Thread-Safe Buffered Observable Priority Queue?

Question

I'm writing a program where one thread needs to push items onto the queue, and one or more threads pops items off the queue and processes them. To avoid running out of memory, I'd like the producer thread to sleep when the queue gets full. Some items have a higher priority than others, so I'd like those to be processed first. If the items have the same priority, I'd like the one that was added first to be processed first.

I want to display the top 100 items or so in a WPF DataGrid, so it needs to be accessed by a UI thread as well. Would be nice if it could notify the UI thread that there's been an update as well, i.e., implements IObservable.

Is there a container class that will do all this?

For bonus points, I'm pretty sure the entire queue doesn't need to be locked both when enqueing and dequeing.

.NET 4 implementations are fine.

Answer 1

If you're on .NET 4 you should seriously consider the Task Parallel Library with a custom scheduler like the example QueuedTaskScheduler. I'm not sure it meets all your requirements, but it would be a good start.

Answer 2

You are out of luck looking for a container - you have to implement one yourself. Be carefull with priorities - sorting gets slow fast. What I do is I have a queue class implemented myself that internally uses multiple arrays (one per priority - coded low, middle, high). This way I never sort. Avoid locks if you can (multi core assumed) and go for Spinlocks (.NET 4.0), they are faster / carry less overhead in a queue scenario.

Answer 3

What I've done in the past is wrapped multiple ConcurrentQueue<T> collections into one -- sort of what TomTom suggests. This is pretty reasonable when the number of priorities you intend to have is low. For instance in some cases it may even be sufficient to have two: high and low. Then your TryDequeue method just looks something like this:

public bool TryDequeue(out T item)
{
    return _highItems.TryDequeue(out item) || _lowItems.TryDequeue(out item);
}

This isn't exactly a comprehensive answer to your question, but maybe it can help you get started.