Can I TryTake a group of items from a BlockingCollection<T>?

Question

In an application I'm working on, thousands of updates are being received per second. Reflecting these updates on the UI immediately and one by one is performance overkill.

The following code yields very bad performance, because handling each update requires invoking the UI thread and adding a new item to an ObservableCollection which in turn triggers the CollectionChanged event.

foreach (var update in updates.GetConsumingEnumerable())
{
    // Handle Update
}

What I'm trying to do is to make the consumer wait for a little bit of time (i.e. 50 millisecond) to give a chance to the publisher to add more items, and then handle these updates in chunks.

Currently, I'm using the following code, but I find it hard to predict for how long the items will stay in the collection before being consumed, I'm afraid this approach may produce another performance bottleneck.

List<Tuple<T, List<string>>> updatesToHandle = new List<Tuple<T, List<string>>>();
while (!updates.IsAddingCompleted)
{
    Tuple<T, List<string>> item = null;
    if (updates.TryTake(out item, 5)) // Try Take for 5 Milliseconds
    {
        updatesToHandle.Add(item);
    }
    else
    {
        var newItems = new List<T>(updatesToHandle.Count);
        var updatedItems = new List<Tuple<T, List<string>>>(updatesToHandle.Count);
        foreach (var update in updatesToHandle)
        {
            try
            {
                // Handle Update
            }
            finally
            {
                updatesToHandle.Clear();
                Thread.Sleep(50);
            }
        }
    }
}

Answer 1

I would consider using Reactive Extensions ReactiveExtensions. This link solves a similar problem Stock Trading Example

Answer 2

I would suggest two changes to your code.

First, rather than having the thread sleep, create a timer that fires every 50 milliseconds. That timer handler will then loop on TryTake with no delay, collecting all of the items that are currently in the collection (or up to some maximum). With no delay, TryTake will return immediately if the collection is empty.

Second, do not call the "update UI" once for each update. Modify the update handler so that it accepts the entire list of updates rather than one update at a time. That will avoid the time involved in waiting for the UI thread.

The above assumes, of course, that you've modified your update handler so that it can add multiple items to the ObservableCollection in one shot, preventing multiple CollectionChanged events.

Answer 3

I second Rx. Specifically you want BufferByTime (if aggregating when incoming rate is high) or Throttle (if throwing away values when incoming rate is too high). Also, your UI should probably be bound to BehaviorSubject, which will always cache the last value so new subscribers get that last cached value immediately when they subscribe.