Removing items from a dictionary after a timeout in a CPU friendly way

Question

I have a usecase, in which I need to remove items from a dictionary either when a certain number of items are reached (so, like half of the oldest items in the dictionary will be removed), or when items have stayed in dictionary for lets say 10 seconds.

New items are continuously added to the dictionary, and the reason why I am using a ConcurrentDictionary here is to have as much distinct set of items while also streaming.

I have already accomplished this by using a setup like ConcurrentDictionary where I am using message's key (which is like a small json with max 3 fields) as the key for dictionary and the message and the time stored as a tuple as the value for dictionary. I can now use the key to check for duplicates, have a spillover scenario which I handle like (not the actual code, writing this by memory)

foreach (var item in dictionary.OrderByDescending(kvp => kvp.Value.Item2).Take(Threshold / 2))
{
    dictionary.TryRemove(kvp.Key, out var _);
    // add to output queue
}

and I check the old items by having a code like

foreach (var kvpin dictionary)
{
    if (DateTime.UtcNow.Subtract(kvp.Value.Item2) >= TimeSpan.FromSeconds(10))
    {
        dictionary.TryRemove(distinctMessageKVP.Key, out var _)
        // do something with the removed item
    }
}

And this works. Problem is, this is VERY CPU intensive. I plan to reduce CPU usage by switching to single thread dictionary access and using dictionary instead of concurrent dictionary. Then I also plan to increase the threshold, so we don't keep transferring items and wasting cycles there. I also suspect iterating through all the items for removing items from dictionary is an expensive process. If that's the case, is there any other way I can remove items from the dictionary? Is there any other data structure I can use which will help me solve the issue?

EDIT : Right now, we have 5 tasks writing into the ConcurrentDictionary, and 1 task removing stuff from dictionary. If it's a better setup (to avoid locks), I will have the 5 tasks write into a ConcurrentQueue, then 1 task into Dictionary (not concurrent) and the same task to depopulate the dictionary as well.

EDIT2 : We are looking at something like 1000 items going in and out of the dictionary per second.

user_11178549 · Answer

This problem seems like a tempting case for using the Reactive Extensions library. There is a related question here, that contains this custom implementation:

public static IObservable DistinctFor(this IObservable src,
    TimeSpan validityPeriod)

It doesn't cover the requirement for eviction of old entries though.

Another library that could offer the required functionality as an easy-to-use class is the TPL Dataflow. Something like a BufferBlock that keeps unique items. But I don't see any implementation to exist.

Going back to your existing code, there are two major optimizations that I could suggest:

Don't OrderByDescending the contents of the ConcurrentDictionary. Sorting is an expensive operation. A simple enumeration with foreach seems sufficient for finding and removing the timed-out entries.
Avoid calling frequently the DateTime.UtcNow, it is quite expensive too. In your case you don't actually care about the exact date and time that each entry was entered (it makes no difference if it was day or night, or Monday or weekend). You are just interested about its age. So instead of storing a DataTime field I would consider storing a TimeSpan, generated by a Stopwatch. AFAIK accessing the property Stopwatch.Elapsed is more efficient than accessing the DateTime.UtcNow. If you follow this path you should probably use a lock every time you read this property, because there is a debate whether it is thread-safe or not.

Removing items from a dictionary after a timeout in a CPU friendly way

1 Answers