How to implement a dynamic semaphore?

Question

redirected from: http://social.msdn.microsoft.com/Forums/en-US/csharpgeneral/thread/fe5c0012-3e57-4498-9fec-5051bdd78ede/

How to implement a dynamic semaphore so that the resources it guards can be dynamically increase or decrease at different time.

I saw there is a way to do it in Java.  But since the framework implement semaphore differently, I wish to know that if I can implement a resizeable semaphore in Dot Net.  

My project needs this functionality that a service is able to release avaliable resource to client, but the number of resources are different at different time (day time/night time).   I think the semaphore is the best way to guard resources.  and I need a adjustable semaphore (or something alike)

I need enlightments of how to implement a resizeable semaphore (maybe from existing Threating.Semaphore, Mutex or WaitHandler), so that I can use it to guard my resources dynamically.

Thanks in advance

 

 

Answer 1

You can roll your own.  Here's my attempt at a "no frills" version.

I wrote this in a hurry -- watch for bugs.

  public class AdjustableSemaphore
  {
    private object m_Monitor = new object();
    private int m_Count = 0;
    private int m_MaximumCount = 0;

    public AdjustableSemaphore(int maximumCount)
    {
      this.MaximumCount = maximumCount;
    }

    public int MaximumCount
    {
      get
      {
        lock (m_Monitor)
          return m_MaximumCount;
      }
      set
      {
        lock (m_Monitor)
        {
          if (value < 1)
            throw new ArgumentException("Must be greater than or equal to 1.", "MaximumCount");

          m_Count += (value - m_MaximumCount); // m_Count can go negative. That's okay.
          m_MaximumCount = value;
          Monitor.PulseAll(m_Monitor);
        }
      }
    }

    public void WaitOne()
    {
      lock (m_Monitor)
      {
        while (m_Count <= 0)
          System.Threading.Monitor.Wait(m_Monitor);

        m_Count--;
      }
    }

    public void Release()
    {
      lock (m_Monitor)
      {
        if (m_Count < m_MaximumCount)
        {
          m_Count++;
          System.Threading.Monitor.Pulse(m_Monitor);
        }
        else
          throw new System.Threading.SemaphoreFullException("Semaphore released too many times.");
      }
    }
  }

Answer 2

Thanks BinaryCoder for the quick solution!

But I think the attempt could not address the following condition:

"Dynamic decrease permits - To be specific, if 10 threads currently have permits and 7 is the new limit, we won’t try to communicate to any threads that they are now in violation of the new limit. What we can do is make sure that the next call to acquire() will block until enough threads (4, to be precise) have called release() to bring the number of outstanding permits to below the new limit. It’s not infeasible to interrupt threads that are holding permits beyond the new upper limit, but that’s beyond the scope of a simple semaphore."

The above argument is from the  "Extending Java’s Semaphore to Allow Dynamic Resizing"

http://eng.genius.com/blog/2009/04/20/javas-semaphore-resizing/

 

In Java, the semaphore class has reducePermits() method. Unfortunately, Microsoft semaphore does not have the same method. We need to find a way to handle it.

 

Any thoughts?

Answer 3

Are you sure this is not handled by my example?

I believe if you reduce MaximumCount such that the semaphore becomes over-acquired, the m_Count goes negative.  This is intentionally not an error.  Further acquisitions will not succeed until releases bring m_Count positive.

 

Answer 4

Nope, It did not work :(

I tested with following code, the result should allow 5 threads working after the main thread change the semaphore's maximum count to 5.  but the result still showes only 1 thread is doing work.

public class Example
 {
  // A semaphore that simulates a limited resource pool.
  private static AdjustableSemaphore _pool;

  // A padding interval to make the output more orderly.
  private static int _padding;

  public static void Main()
  {
   // Create a semaphore that can satisfy up to one concurrent requests.
   _pool = new AdjustableSemaphore(1);

   // Create and start five numbered threads.
   for (int i = 1; i <= 20; i++)
   {
    Thread t = new Thread(new ParameterizedThreadStart(Worker));
    t.Start(i);
   }

   Thread.Sleep(5000);
 
   _pool.MaximumCount = 5;

   Console.WriteLine("Main thread exits.");
  }

  private static void Worker(object num)
  {
   Console.WriteLine("Thread {0} begins and waits for the semaphore.", num);
   _pool.WaitOne();

   // A padding interval to make the output more orderly.
   int padding = Interlocked.Add(ref _padding, 100);

   Console.WriteLine("Thread {0} enters the semaphore.", num);

   Thread.Sleep(1000 + padding);

   Console.WriteLine("Thread {0} releases the semaphore.", num);
   Console.WriteLine("Thread {0} previous semaphore count: {1}", num, _pool.Release());
  }
 }

Answer 5

You have found a bug.  I have edited my original post to include a call to Monitor.PulseAll in the MaximumCount setter.

 

Answer 6

Thanks a lot!

I have run the code, it seems working.  I will need test the code more thoroughly in our project before I get back here.

stay tuned..

Answer 7

Hi BinaryCoder,

The implementation was correct and works great in my project!

Just wondering do you have any idea to add the WaitOne(timespan), or WaitOne(timespan, bool) into our implentation of the adjustable semaphore?

Thanks in advance!

Answer 8

I'm not sure how the bool parameter works... I looked at it briefly and read some really obscure stuff about ContextBoundObject and SynchronizationAttribute. You might be able to just pass it through to the Monitor.Wait parameter, but I'm a little hesitant because I have zero familiarity with this parameter.

As far as the TimeSpan, you should be able to just add the parameter to the WaitOne method and follow through with passing that on to the corresponding Monitor.Wait parameter.

 

 

Answer 9

Here is a something a little interesting when using a timeout.  I don't know if this is a real problem or not but wanted to bring it up.  I'd be interested to know what someone very familiar with the CLR monitor implementation thinks.

The documentation for Wait states that:

true if the lock was reacquired before the specified time elapsed

Since time can pass between the call to Pulse and the Monitor.Exit of the pulsing thread, taken literally this suggests that it might be possible that a single waiting thread is successfully pulsed but still unable to reacquire the lock within the specified timeout.  If the thread that timed out does not soon attempt to re-enter the monitor, the program may hang until it does so.  If the thread exits before doing so, the hang would be permanent.  (Alas, I don't have time to experiment with this right now, sorry...)

If this is a problem, I think replacing the Monitor.Pulse with a Monitor.PulseAll could help avoid this, but I'm not sure if that would completely resolve for all cases.

 

Answer 10

How about this implemenation?  It seems to run well in the demo code, can you see any potential bugs when multithreading?

I got the idea from : http://stackoverflow.com/questions/299198/implement-c-generic-timeout

The two method I added are:  public bool WaitOne(int timeoutMilliseconds)     and     static bool CallWithTimeout(Action action, int timeoutMilliseconds)

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;

namespace AdjustableSemaphore
{
 public class Example
 {
  // A semaphore that simulates a limited resource pool.
  private static AdjustableSemaphore _pool;

  public static void Main()
  {
   // Create a semaphore that can satisfy up to three concurrent requests.
   _pool = new AdjustableSemaphore(1);

   // Create and start five numbered threads.
   for (int i = 1; i <= 2; i++)
   {
    Thread t = new Thread(new ParameterizedThreadStart(Worker));
    t.Start(i);
   }

   System.Diagnostics.Debug.WriteLine("Main thread exits.");
  }

  private static void Worker(object num)
  {
   Console.WriteLine("Thread {0} begins and waits for the semaphore.", num);
   if (_pool.WaitOne(2000))
   {
    Console.WriteLine("Thread:" + num + " got semaphore.");
   }
   else
   {
    Console.WriteLine("Thread:" + num  + " timed out.");
    return;
   }

   Console.WriteLine("Thread {0} enters the semaphore.", num);

   Thread.Sleep(5000);

   Console.WriteLine("Thread {0} releases the semaphore.", num);
   Console.WriteLine("Thread {0} previous semaphore count: {1}", num, _pool.Release());
  }
 }

 public class AdjustableSemaphore
 {
  private object m_Monitor = new object();
  private int m_Count = 0;
  private int m_MaximumCount = 0;

  public AdjustableSemaphore(int maximumCount)
  {
   this.MaximumCount = maximumCount;
  }

  public int MaximumCount
  {
   get
   {
    lock (m_Monitor)
     return m_MaximumCount;
   }
   set
   {
    lock (m_Monitor)
    {
     if (value < 1)
      throw new ArgumentException("Must be greater than or equal to 1.", "MaximumCount");

     m_Count += (value - m_MaximumCount); // m_Count can go negative. That's okay.
     m_MaximumCount = value;
     Monitor.PulseAll(m_Monitor);
    }
   }
  }

  public bool WaitOne(int timeoutMilliseconds)
  {
   return CallWithTimeout(WaitOne, timeoutMilliseconds);
  }

  public void WaitOne()
  {
   lock (m_Monitor)
   {
    while (m_Count <= 0)
     System.Threading.Monitor.Wait(m_Monitor);

    m_Count--;
   }
  }

  public int Release()
  {
   lock (m_Monitor)
   {
    if (m_Count < m_MaximumCount)
    {
     m_Count++;
     System.Threading.Monitor.Pulse(m_Monitor);
    }
    else
     throw new System.Threading.SemaphoreFullException("Semaphore released too many times.");
   }
   return m_Count;
  }

  public void Release(int count)
  {
   lock (m_Monitor)
   {
    if (m_Count + count  < m_MaximumCount)
    {
     m_Count += count;
     System.Threading.Monitor.Pulse(m_Monitor);
    }
    else
     throw new System.Threading.SemaphoreFullException("Semaphore released too many times.");
   }
  }

  static bool CallWithTimeout(Action action, int timeoutMilliseconds)
  {
   Thread threadToKill = null;
   Action wrappedAction = () =>
   {
    threadToKill = Thread.CurrentThread;
    action();
   };

   IAsyncResult result = wrappedAction.BeginInvoke(null, null);
   if (result.AsyncWaitHandle.WaitOne(timeoutMilliseconds))
   {
    wrappedAction.EndInvoke(result);
    return true;
   }
   else
   {
    threadToKill.Abort();
    return false;  // throw new TimeoutException();
   }
  }
 }
}

Answer 11

> How about this implemenation?

Bad.  Really bad.  Thread.Abort can cause unpredictable problems because you can't be sure what the target thread was in the middle of doing when you aborted it.  The badness is explained even on the page you referenced.

See my post from not long ago for a more reasonable technique.

I take it you might be new to multithreaded programming.  I strongly recommend that you read up on multithreading.  It's a complex topic and it takes a while to get the "intuition" of how to do things so that they work reliably.  Learn about how the Monitor class in particular.

Answer 12

You are absolutely right, the last time I coded for multi-threading was 10 years ago when I was a college student. at the time I was fairly good at the concepts and the coding.

but now, the knowledge has all gone.  I definitely need to pick them up.

Thanks for your suggestion.  I have added the new method as you suggested:

 

///

 

<summary>

 

/// Blocks the current thread until the current WaitHandle receives a signal,

 

/// using a 32-bit signed integer to measure the time interval.

 

/// </summary>

 

/// <param name="millisecondsTimeOut">The number of milliseconds to wait, or Timeout.Infinite (-1) to wait indefinitely.</param>

 

/// <returns>true if the current instance receives a signal; otherwise, false.</returns>

 

public bool WaitOne(int millisecondsTimeOut)

{

 

lock (m_Monitor)

{

 

while (m_Count <= 0)

{

 

bool waitSucceed;

waitSucceed = System.Threading.

Monitor.Wait(m_Monitor, millisecondsTimeOut);

 

if (waitSucceed)

{

m_Count--;

}

 

 

return waitSucceed;

}

m_Count--;

 

return true;

}

}

Answer 13

Final version and run in production for years.

namespace Soap.Service.Extension
{
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;


/// <summary>
/// The AdjustableSemaphore is a semaphore that can be dynamically change the available
/// counter at run-time.
/// </summary>
public class AdjustableSemaphore
{
private object m_Monitor = new object();
private int m_Count = 0;
private int m_MaximumCount = 0;

public AdjustableSemaphore(int maximumCount)
{
this.MaximumCount = maximumCount;
}

public int MaximumCount
{
get
{
lock (m_Monitor)
return m_MaximumCount;
}
set
{
lock (m_Monitor)
{
if (value < 1)
throw new ArgumentException("Must be greater than or equal to 1.", "MaximumCount");

m_Count += (value - m_MaximumCount); // m_Count can go negative. That's okay.
m_MaximumCount = value;
Monitor.PulseAll(m_Monitor);
}
}
}

public void WaitOne()
{
lock (m_Monitor)
{
while (m_Count <= 0)
System.Threading.Monitor.Wait(m_Monitor);

m_Count--;
}
}

public void Release()
{
lock (m_Monitor)
{
if (m_Count < m_MaximumCount)
{
m_Count++;
System.Threading.Monitor.Pulse(m_Monitor);
}
else
throw new System.Threading.SemaphoreFullException("Semaphore released too many times.");
}
}

public void GetSemaphoreInfo(out int totalResource, out int usedResource, out int avaliableResource)
{
lock (m_Monitor)
{
totalResource = m_MaximumCount;
usedResource = m_MaximumCount - m_Count;
avaliableResource = m_MaximumCount - usedResource;
}
}
}
}