Counting Semaphore (Dijkstra) in C#

• Download source files - 3.38 Kb

Introduction

Here is a C# implementation of a Dijkstra counting Semaphore. This can also be used as a Reverse Sensing Semaphore as a thread can block until the “Count” reaches zero. Any thread(s) waiting on WaitForStarvation() will be Pulsed by the first thread that “Released” the count to zero. In addition, this has many helper methods and overloads normally not available in other implementations, such as: current Count, Acquire with timeouts, AquireAll(), ReleaseAll(), WaitForStarvation(), and “Try” versions of the Acquire and Release methods.

The Code:

using System;
using System.Threading;

namespace SyncPrimitives
{
    /// <summary>
    /// Author: William Stacey (staceyw@mvps.org)
    /// Date: 06/10/04
    /// The Dijkstra semaphore (also called a counting
    /// semaphore) is used to control access to
    /// a set of resources. A Dijkstra semaphore
    /// has a count associated with it and each
    /// Acquire() call reduces the count. A thread
    /// that tries to Acquire() the semaphore
    /// with a zero count blocks until someone else
    /// calls Release() to increase the count.
    /// <seealso cref="http://www.fawcette.com/javapro/
    ///            2002_02/magazine/features/krangaraju/"/>
    /// <seealso cref="http://www.mcs.drexel.edu/~shartley/
    ///         MCS361/Lectures/designingJavaSemaphore.html"/>
    /// </summary>
    public sealed class Semaphore
    {
        #region Fields
        // Current count available.
        private int count;
        // Max slots in the semaphore.
        private int maxCount;
        // Object used for sync.
        private readonly object syncLock;
        // Object used for starvation sync.
        private readonly object starvationLock;
        #endregion

        #region Constructors
        /// <summary>
        /// Creates semaphore object with a maxCount
        /// and set initial count to maxCount.
        /// </summary>
        /// <param name="maxCount">
        /// Maximum count for the semaphore object.
        /// This value must be greater than zero.
        /// </param>
        public Semaphore(int maxCount) : this(maxCount, maxCount)
        {
        }

        /// <summary>
        /// Creates semaphore object with
        /// a maximum count and initial count.
        /// </summary>
        /// <param name="maxCount">
        /// Maximum count for the semaphore object.
        /// This value must be greater than zero.
        /// </param>
        /// <param name="initialCount">
        /// Initial count for the semaphore object.
        /// This value must be zero or greater
        /// and less than or equal to maximumCount.
        /// </param>
        public Semaphore(int initialCount, int maxCount)
        {
            if ( initialCount < 0 )
                throw new 
                  ArgumentOutOfRangeException("initialCount must be >= 0.");
            if ( maxCount < 1 )
                throw new ArgumentOutOfRangeException("maxCount must be >= 1.");
            if ( initialCount > maxCount)
                throw new 
                  ArgumentOutOfRangeException("initialCount" + 
                  " must be <= maxCount.");
            count = initialCount;
            this.maxCount = maxCount;
            syncLock = new object();
            starvationLock = new object();
        }

        #endregion

        #region Properties
        /// <summary>
        /// Gets the current available count (or slots)
        /// in the semaphore. A count of zero means that no slots
        /// are available and calls to Acquire will block until
        /// other thread(s) call Release.
        /// Example:
        /// A semaphore with a count of 2 will allow
        /// 2 more Acquire calls before blocking.
        /// </summary>
        public int Count
        {
            get
            {
                lock(syncLock)
                {
                    return count;
                }
            }
        }

        /// <summary>
        /// Gets the maximum count of the semaphore
        /// set during construction.
        /// </summary>
        public int MaxCount
        {
            get { return maxCount; }
        }

        #endregion

        #region Public Methods

        /// <summary>
        /// Acquires semaphore and decrements count by 1.
        /// If count is zero, this will
        /// block indefinitely until another thread executes
        /// a Release() to increase the count.
        /// </summary>
        /// <returns>true if the call returned because
        /// the caller reacquired the lock for the
        /// specified object. This method does not return
        /// if the lock is not reacquired.</returns>
        public bool Acquire()
        {
            return Acquire(Timeout.Infinite);
        }

        /// <summary>
        /// Returns a value indicating if Semephore
        /// can be acquired within the timeout period.
        /// </summary>
        /// <returns>true if the lock was acquired before
        /// the specified time elapsed; otherwise, false.</returns>
        /// <exception cref="ArgumentOutOfRangeException">
        /// The value of the millisecondsTimeout parameter
        /// is negative, and is not equal to Infinite.
        /// </exception>
        public bool Acquire(int millisecondsTimeout)
        {
            lock(syncLock)
            {
                // Use spin lock instead of an if test, to handle
                // rogue/barging threads that can enter
                // syncLock before a thread that was notified by a pulse.
                // That rogue thread would
                // decrease the count, then our "Pulsed" thread
                // would regain the lock and continue and
                // decrease the count to -1 which is an error.
                // The while loop/test prevents this.
                while ( count == 0 )
                    try
                    {
                        if (!Monitor.Wait(syncLock, millisecondsTimeout))
                            return false;
                    }
                    catch
                    {
                        // If we get interupted or aborted,
                        // we may have been pulsed before.
                        // If we just exit, that pulse would get lost and
                        // possibly result in a "live" lock
                        // where other threads are waiting
                        // on syncLock, and never get a pulse.
                        // Regenerate a Pulse as we consumed it.
                        // Even if we did not get
                        // pulsed, this does not hurt as any thread
                        // will check again for count = 0.
                        Monitor.Pulse(syncLock);
                        // Rethrow the exception for caller.
                        // Now semaphore state is same as if
                        // this call never happened. Caller must
                        // decide how to handle exception.
                        throw;
                    }
                count--;
                if ( count == 0 )
                    lock(starvationLock) { Monitor.PulseAll(starvationLock); }
                return true;
            }
        }

        /// <summary>
        /// Acquires all the semaphores and brings
        /// count to zero. This has the effect
        /// of block other threads until we release one or more slots.
        /// <seealso cref="Acquire()"/>
        /// <seealso cref="ReleaseAll()"/>
        /// </summary>
        /// <returns>true if the acquired maxCount slots.
        /// This method does not return until
        /// all slots are acquired.</returns>
        public bool AcquireAll()
        {
            // Aquires all slots or blocks for Timeout.Infinite.
            return AcquireAll(Timeout.Infinite);
        }

        /// <summary>
        /// Tries to acquire (maxCount) slots
        /// in semaphore. If any single attempt to
        /// acquire a semaphore slot exceeds
        /// millisecondsTimeout, then return is false.
        /// Return is true if we acquire maxCount slots.
        /// Normally this method would be paired
        /// with the ReleaseAll method.
        /// </summary>
        /// <param name="millisecondsTimeout"></param>
        /// <returns>true if maxCount slots are acquired
        /// before the specified time elapsed;
        /// otherwise, false.</returns>
        public bool AcquireAll(int millisecondsTimeout)
        {
            int slotsGot = 0;
            int elapsedMS = 0;
            DateTime start = DateTime.Now;
            int timeout = millisecondsTimeout;
            for (int i = 0; i < maxCount; i++)
            {
                try
                {
                    if (! Acquire(timeout) )
                    {
                        // Could not acquire all slots,
                        // release any we may already have got.
                        if ( slotsGot > 0 )
                            Release(slotsGot);
                        return false;
                    }
                    else
                    {
                        elapsedMS = (int)((TimeSpan)
                           (DateTime.Now - start)).TotalMilliseconds;
                        timeout = millisecondsTimeout - elapsedMS;
                        // Next wait will be a smaller timeout.

                        if ( timeout < 0 )
                            timeout = 0;
                            // Next Acquire will return
                            // false if we have to wait;

                        slotsGot++;
                        // If we get all remaining slots
                        // with no timeout, we just keep going.
                    }
                }
                catch
                {
                    // Catch any exception during Acquire wait.
                    if ( slotsGot > 0 )
                        Release(slotsGot);
                    throw;
                }
            } // end for.
            // Count is not zero, so notify any/all starvation consumers.
            lock(starvationLock) { Monitor.PulseAll(starvationLock); }
            return true;
        }

        /// <summary>
        /// Increases the count of the semaphore object by one.
        /// </summary>
        public void Release()
        {
            Release(1);
        }

        /// <summary>
        /// Increases the count of the semaphore
        /// object by a specified amount.
        /// </summary>
        /// <param name="count">Amount by which the semaphore
        /// object's current count is to be increased.</param>
        /// <exception cref="ArgumentOutOfRangeException">
        /// The releaseCount must be one or greater.
        /// </exception>
        /// <exception cref="ArgumentOutOfRangeException">
        /// The releaseCount would cause
        /// the semaphore's count to exceed maxCount. 
        /// </exception>
        public void Release(int releaseCount)
        {
            if ( releaseCount < 1 )
                throw new 
                  ArgumentOutOfRangeException("releaseCount must be >= 1.");

            lock(syncLock)
            {
                if ( (count + releaseCount) > maxCount )
                    throw new 
                      ArgumentOutOfRangeException("releaseCount" + 
                      " would cause the semaphore's count to exceed maxCount.");
                count += releaseCount;
                Monitor.PulseAll(syncLock);
            }
        }

        /// <summary>
        /// Returns indication if we could
        /// release one slot in the semaphore.
        /// </summary>
        /// <returns>true if we released
        /// one slot; otherwise false.</returns>
        public bool TryRelease()
        {
            return TryRelease(1);
        }

        /// <summary>
        /// Returns indication if we could release
        /// releaseCount slots in the semaphore.
        /// </summary>
        /// <param name="releaseCount"></param>
        /// <returns>true if we released releaseCount
        /// slots; otherwise false.</returns>
        public bool TryRelease(int releaseCount)
        {
            if ( releaseCount <= 0 )
                return false;

            lock(syncLock)
            {
                if ( (count + releaseCount) > maxCount )
                    return false;
                else
                    count += releaseCount;
                Monitor.PulseAll(syncLock);
                return true;
            }
        }

        /// <summary>
        /// Releases all remaining semaphores
        /// not currently owned. This would normally be
        /// called by a thread that previously
        /// called AcquireAll(). Note:  Be carefull when
        /// using this method as it will release
        /// all threads waiting on an Aquire method,
        /// which may or may not be what you want.
        /// An alternative would be to spin on
        /// TryRelease() until it returns false.
        /// </summary>
        public void ReleaseAll()
        {
            lock(syncLock)
            {
                count = maxCount;
                Monitor.PulseAll(syncLock);
                // We PulseAll instead of calling pulse
                // with exact number of times needed.
                // This can be slightly inefficient,
                // but is safe and simple.
                // See http://www.mcs.drexel.edu/~shartley/
                //   MCS361/Lectures/designingJavaSemaphore.html
            }
        }

        /// <summary>
        /// This method blocks the calling thread
        /// until the semaphore count drops to zero.
        /// A drop to zero will not be recognized
        /// if a release happens before this call.
        /// You can use this to get notified when
        /// semephore's count reaches zero.  This
        /// is also known as a "reverse-sensing" semaphore.
        /// </summary>
        public void WaitForStarvation()
        {
            lock(starvationLock)
            {
                // We will block until count is 0.
                // We use Interlocked just to be sure
                // we test for zero correctly as we
                // are not in the syncLock context.
                if ( Interlocked.CompareExchange(ref count, 0, 0) != 0 )
                    Monitor.Wait(starvationLock);
                // Any Exception during wait will
                // just go to caller.  Do not need to signal
                // any other threads as PulseAll(starvationLock) is used.
                // Also note we don't do a spin
                // while() test as we only care that 
                // count *did go to zero at some instant.
            }
        }
        #endregion
    } // class SemephoreDijkstra
}

History

• Version 1.1 - 06/20/04 - Changed AcquireAll(ms) to respect user's milliseconds total.