Efficiently waiting for one or more resources to become available

Question

Before I spend too long reinventing the wheel, I wanted to check in case there is already a class in .Net that does what I want.

What I want is something a bit like a Semaphore (or perhaps even like a CountdownEvent), but slightly different.

I have a requirement where I have a varying number of "resources" available, and I want a thread to wait efficiently when there are zero resources available. In the meantime, another thread can free up a resource, which should immediately release the other waiting thread.

This sounds a lot like a Semaphore, but its not because a Semaphore (as far as I can see) treats each thread as a "resource" in terms of counting them.

Anyway, here's my first simple implementation of what I want. It has no disposal, code contracts, error handling, timeout support or cancellation support yet, but it should demonstrate what I want:

public sealed class ResourceCounter
{
    /// <summary>Create with the specified number of resources initially available.</summary>

    public ResourceCounter(int resourceCount)
    {
        _resourceCount = resourceCount;

        if (_resourceCount > 0)
        {
            _resourceAvailable.Set();
        }
    }

    /// <summary>Acquires a resource. Waits forever if necessary.</summary>

    public void Acquire()
    {
        while (true)
        {
            _resourceAvailable.Wait();

            lock (_lock)
            {
                if (_resourceCount > 0)
                {
                    if (--_resourceCount == 0)
                    {
                        _resourceAvailable.Reset();
                    }

                    return;
                }
            }
        }
    }

    /// <summary>Releases a resource.</summary>

    public void Release()
    {
        lock (_lock)
        {
            ++_resourceCount;
            _resourceAvailable.Set();
        }
    }

    private int _resourceCount;
    private readonly object _lock = new object(); 
    private readonly ManualResetEventSlim _resourceAvailable = new ManualResetEventSlim();
}

The usage pattern is very simple:

1. Construct a ResourceCounter with the required initial resource count (which can be zero or more).

2. A thread which wants to acquire a resource calls ResourceCounter.Acquire(), which will not return until a resource is available and has been acquired.

3. A thread which wants to release a resource calls ResourceCounter.Release(), which will release a resource and return immediately.

Note that any thread can release a resource; it doesn't have to be the one that acquired the resource.

I'm using this as part of some multithreaded pipeline code where one thread is responsible for enqueuing work items, several threads are processing the work items, and another thread is outputting the processed work items. The thread which outputs the processed work items has to multiplex them (since the processing threads may output completed items in any order), and I needed a mechanism to stop work items from being queued endlessly while the multiplexer waits for a tardy item.

(See Pipelines, multiplexing, and unbounded buffering for some background on this.)

Anyway, is there anything already available to do this, or should I continue to develop my own class for it?

---

[EDIT]

As noted below, a SemaphoreSlim does exactly the right thing. I'd rejected it because I thought that the thread that called Wait() had to be the one that called Release(), but that wasn't the case. This is what I get for coding on a Sunday... ;)

Answer 1

A multi-stage pipeline architecture is more easily constructed using queues to communicate. A producer thread places items into a work queue, one or more worker threads dequeue and process items, and add them to an output queue. A final thread reads the output queue and outputs the data.

In .NET, this is easily accomplished with BlockingCollection.

See https://stackoverflow.com/a/5108487/56778 for an example of a two-stage pipeline. Adding another stage is straightforward.

To handle the problem of the output thread getting things out of order, I made the output queue a priority queue, using a min heap. My items were identified by a sequential record number, so the output thread knew which record number was to be output next. It would wait on an AutoResetEvent for an item to be placed on the queue (the worker process would set the event when an item was enqueued). The output thread would then peek at the top item to see if it matched the expected item. If not, it would wait on the event again.

That worked very well, because it eliminated the second queue. The block was in the output queue, where it belonged. Performance was very good for my purposes. Enqueuing an item is an O(log n) operation, but in practice n is very small. Even when the queue had 100,000 items in it, the amount of time required to enqueue an item was insignificant compared to the amount of time it took to process a record.

You can still use BlockingCollection for this. You just have to make a binary heap class implement the IProducerConsumerCollection interface. I did that by adding locks to the simple binary heap class I published in A Generic BinaryHeap class. You can then provide one of those to the BlockingCollection constructor, like this:

BlockingCollection<MyRecord> = 
    new BlockingCollection<ConcurrentBinaryHeap<MyRecord>>(
    new ConcurrentBinaryHeap<MyRecord>, MaxQueueSize);

There is a potential deadlock here, though. If the queue fills up (i.e. exceeds the maximum you set when you initialized the BlockingCollection), then the tardy thread can't enqueue the item and all work comes to a complete halt. This never happened to me in practice, because although my per-record processing times varied, they didn't vary that much.

If it's a concern, you can either increase the queue size (only works if you can say with certainty that you won't ever fill the queue), or provide an alternate channel for the next expected item to be posted if the queue is full. I made that work, but for my purposes it was easier just to increase the queue size.

If you're interested, I can dig through my archives to find that ConcurrentBinaryHeap class.

Answer 2

The way threads communicate with each other is unrelated to the resources or their locking mechanism. Nothing prevents you to pass semaphores and resources around within the same process, using an ad-hoc messaging system (message queues, events, or whatever fits your need).