Element order in BlockingCollection<>

Question

I have a Download Queue implemented with BlockingCollection<>. Now I want to prioritize some Download once in a while. I thought it might be great to move some elements 'up' the Collection, like in a list, but there is no method like Remove()/AddFirst() or Move().

What's the preferred way of arranging items in a BlockingCollection<> ?

Catastrophism

Related: Concurrent collection with priority. – Laskowski

Answer 1

BlockingCollection<T> works by wrapping an internal IProducerConsumerCollection<T>. The default is to use a ConcurrentQueue<T> internally, but you can provide your own implementation via this constructor.

If you provide your own threadsafe collection, you can use any collection type you want. This would allow to you prioritize elements as needed.

While there are no built-in collections that will implement your desired functionality, you could probably wrap a pair of ConcurrentQueue<T> collections into a class that implements IProducerConsumerCollection<T>. This would allow you to have "high priority" and "low priority" elements.

Kinase

Answer 2

Unfortunately there is no way to rearrange the queue in the manner you desire. What you really need is a PriorityBlockingCollection implemented as a priority queue, but alas that does not exist either.

What you can do is exploit the TakeFromAny method to get the priority behavior you want. TakeFromAny will dequeue the first available item from an array of BlockingCollection instances. It will give priority to queues listed first in the array.

var low = new BlockingCollection<object> { "low1", "low2" };
var high = new BlockingCollection<object> { "high1", "high2" };
var array = new BlockingCollection<object>[] { high, low };
while (true)
{
  object item;
  int index = BlockingCollection<object>.TakeFromAny(array, out item);
  Console.WriteLine(item);
}

The example above will print:

high1
high2
low1
low2

It forces you to use to multiple queues so it is not the most elegant solution.

Winters

Answer 3

There is no way to implement a priority queue directly on top of a BlockingCollection<T>. A BlockingCollection<T> is best viewed as a strict queue for which no reordering can be achieved.

However you could use a combination of a priority queue and a BlockingCollection<T> to achieve the same effect. Lets assume for a second you implemented a simple PriorityQueue<T> which correctly orders your downloads. The following could be used to add priority to the handling of the receiving side

class DownloadManager {
  private PriorityQueue<Download> m_priorityQueue;
  private BlockingCollection<Download> m_downloadCollection;

  public bool TryGetNext(ref Download download) {
    PumpDownloadCollection();
    if (m_priorityQueue.IsEmpty) {
      download = null;
      return false;
    }

    download = m_priorityQueue.Dequeue();
    return true;
  }

  private void PumpDownloadCollection() {
    T value;
    while (m_downloadCollection.TryTake(out value)) {
      m_priorityQueue.Enqueue(value);
    }
  }

Note: PriorityQueue<T> is not a type that actually exists in the .Net Framework. It's something you'd need to write yourself based on the priority scheduling of downloaded items.

Vikiviking

Answer 4

Reed is correct in telling you that you need to implement the IProducerConsumerCollection<T>. However, there is a class that can help you. It's not built in, but it's featured on MSDN. Just pass this ConcurrentPriorityQueue to your BlockingCollection.

This is how I used it:

private readonly BlockingCollection<KeyValuePair<int, ICommand>> _commands 
    = new BlockingCollection<KeyValuePair<int, ICommand>>(
        new ConcurrentPriorityQueue<int, ICommand>());

The ICommand is an interface in my project.

Now this allows you to add items like this:

_actions.Add(new KeyValuePair<int, ICommand>(1, command1));
_actions.Add(new KeyValuePair<int, ICommand>(2, command2));
_actions.Add(new KeyValuePair<int, ICommand>(1, command3));

Items with a lower integer value as priority will be executed first. In the above example:

command1
command3
command2

When looping over your BlockingCollection, you will no longer get the single elements (ICommand in my case), but a KeyValuePair. This might require some code changes of course. A nice thing is that you have its original priority:

foreach (var command in _queue) 
{
    var priority = command.Key;
    var actualCommand = command.Value;
}

Marillin