The blog of dlaa.me

In my notes for the release of the ComputeFileHashes tool (and source code), I mentioned that I'd written a synchronization object for managing the parallel computation of checksums across multiple threads. I called this class WaitingRoom (after failing to come up with anything better) and thought I'd write about the pattern it implements so others might use it as well.

To understand the motivation behind the WaitingRoom class, it's helpful to understand a bit about how ComputeFileHashes works. For those who haven't read the implementation notes, the basic goal is to enable parallel computation of multiple checksum algorithms for sequential chunks of a file. So there's a primary thread which is responsible for opening the file and making consecutive blocks of it available for processing and there are one or more worker threads which are responsible for processing the data in each block that becomes available. For performance reasons, there are two buffers for these blocks and they're swapped repeatedly: one is the "current" block (which has valid data) and the other is the "next" block (which is being filled-in). It's important to ensure the worker threads only access a block when it's valid, so the primary thread needs a way to tell the worker threads when it's safe to start as well as a way to find out when they've all finished with a block.

The WaitingRoom class makes this synchronization process easy by exposing three methods: Wait, Release, and Arrive. Both Wait and Release are used by the primary thread: it calls Wait to block until all worker threads are "in the waiting room" (during which time they're all blocked) and then calls Release to "open the waiting room door" and unblock the worker threads. [Okay, the analogy breaks down a bit here because most doctors admit only *one* patient at a time. :) ] The worker threads call Arrive to signal they've "entered the waiting room" and automatically block until the primary thread releases them. What's great about using WaitingRoom is that the order of the threads doesn't matter: the primary thread can be ready first, or the worker threads can be ready first, or the primary thread can be ready after some - but not all - of the worker threads are ready. Whatever the order, WaitingRoom coordinates things smoothly!

As it happens, ComputeFileHashes uses two WaitingRoom instances. The first instance is used to wait for the worker threads to be ready to process a new file. When this is going on, the primary thread has nothing else to do, so it calls Wait and then immediately calls Release when the wait completes. The second instance is used to wait for the worker threads to finish processing a block of data. In this case, the primary thread has some work it can do in parallel with the workers: it reads the next block of data from disk and updates the status display. So it makes a call to Release (which gets the worker threads going), does its own work while they're busy, then calls Wait to wait for them to finish. After that, it's safe to swap the "current" and "next" buffers and repeat the same steps with the next block.

The complete implementation of WaitingRoom is below. It uses only the standard .NET Monitor class - along with the C# lock statement to simplify the syntax a bit. There are a few Debug.Asserts in there to try to keep callers honest, but it's up to the developer to ensure that Wait and Release are only called by the primary thread and that Arrive is only called by the worker threads.

Here's what it looks like:

using System.Diagnostics;
using System.Threading;

namespace ComputeFileHashes
{
    /// <summary>
    /// Implements a synchronization object that allows an owner
    /// thread to synchronize with its worker threads.
    /// </summary>
    class WaitingRoom
    {
        // Number of worker threads
        private readonly int _capacity;
        // Object on which to lock for entry
        private readonly object _entryLock = new object();
        // Object on which to lock for exit
        private readonly object _exitLock = new object();
        // Current count of worker threads
        private int _count;
        // "Sign" of owner/worker threads
        private bool _sign;

        /// <summary>
        /// Initializes a new instance.
        /// </summary>
        /// <param name="capacity">Number of worker threads.</param>
        public WaitingRoom(int capacity)
        {
            Debug.Assert(0 < capacity);
            _capacity = capacity;
        }

        /// <summary>
        /// Waits for all worker threads to call the Arrive method.
        /// </summary>
        public void Wait()
        {
            // Claim entry lock
            lock (_entryLock)
            {
                Debug.Assert((0 <= _count) && (_count <= _capacity));

                // Block if all worker threads have not arrived
                if (_count < _capacity)
                {
                    Monitor.Wait(_entryLock);
                }
            }
        }

        /// <summary>
        /// Signals the presence/availability of a worker thread.
        /// </summary>
        public void Arrive()
        {
            // Claim the entry lock
            bool sign;
            lock (_entryLock)
            {
                Debug.Assert((0 <= _count) && (_count < _capacity));

                // Capture sign
                sign = _sign;

                // Wake owner thread if all worker threads present
                _count++;
                if (_count == _capacity)
                {
                    Monitor.Pulse(_entryLock);
                }
            }

            // Claim the exit lock
            lock (_exitLock)
            {
                // Block if owner has not yet released the worker threads
                if (sign == _sign)
                {
                    Monitor.Wait(_exitLock);
                }
            }
        }

        public void Release()
        {
            // Claim the exit lock
            lock (_exitLock)
            {
                Debug.Assert(_count == _capacity);

                // Reset count and flip sign
                _count = 0;
                _sign = !_sign;

                // Wake worker threads
                Monitor.PulseAll(_exitLock);
            }
        }
    }
}

There's probably an official name for this kind of synchronization primitive, but I don't know what it is. :| I poked around a bit on Wikipedia's "Currency Control" page just now and found something called room synchronization that sounds close... I think WaitingRoom may be a type of room synchronization - with some specialized restrictions for the specifics of the ComputeFileHashes scenario. But whatever you call it, WaitingRoom is a pretty handy class to work with. Feel free to use WaitingRoom in your own code - maybe even drop me a note if you find other interesting uses. And if anyone reading knows the real name for what it's doing, please let me know! :)