3
\$\begingroup\$

I have multiple distinct processes that need to access external resources that are rate limited. The processes are all async in nature and run in different applications. In times past I would just use SemaphoreSlim this design doesn't allow for that.

I've found several samples that seem to be half complete, or cut and paste into their code. This was derived from an existing post, but heavily modified to encompass the additional methods and to honor the cancellation token.

Any feedback would be appreciated.

public sealed class SemaphoreAsync : IDisposable
{
  Semaphore _semaphore;

  private SemaphoreAsync(Semaphore sem) => _semaphore = sem;
  public SemaphoreAsync(int initialCount, int maximumCount) => _semaphore = new Semaphore(initialCount, maximumCount);
  public SemaphoreAsync(int initialCount, int maximumCount, string name) => _semaphore = new Semaphore(initialCount, maximumCount, name);
  public SemaphoreAsync(int initialCount, int maximumCount, string name, out bool createdNew, SemaphoreSecurity semaphoreSecurity) => _semaphore = new Semaphore(initialCount, maximumCount, name, out createdNew, semaphoreSecurity);

  public static SemaphoreAsync OpenExisting(string name)
  {
    return new SemaphoreAsync(Semaphore.OpenExisting(name));
  }

  public static SemaphoreAsync OpenExisting(string name, SemaphoreRights rights)
  {
    return new SemaphoreAsync(Semaphore.OpenExisting(name, rights));
  }

  public static bool TryOpenExisting(string name, out SemaphoreAsync result)
  {
    if (Semaphore.TryOpenExisting(name, out Semaphore semaphore))
    {
      result = new SemaphoreAsync(semaphore);
      return true;
    }
    result = null;
    return false;
  }

  public static bool TryOpenExisting(string name, SemaphoreRights rights, out SemaphoreAsync result)
  {
    if (Semaphore.TryOpenExisting(name, rights, out Semaphore semaphore))
    {
      result = new SemaphoreAsync(semaphore);
      return true;
    }
    result = null;
    return false;
  }

  public async Task<bool> WaitOne(TimeSpan timeout, CancellationToken ct)
  {
    DateTime start = DateTime.UtcNow;
    while (!_semaphore.WaitOne(0))
    {
      ct.ThrowIfCancellationRequested();
      if (DateTime.UtcNow < start.Add(timeout))
        return false;
      await Task.Delay(100, ct);
    }
    return true;
  }

  public async Task<bool> WaitOne(int millisecondsTimeout, CancellationToken ct)
  {
    DateTime start = DateTime.UtcNow;
    while (!_semaphore.WaitOne(0))
    {
      ct.ThrowIfCancellationRequested();
      if (millisecondsTimeout > 0)
      {
        if (DateTime.UtcNow < start.AddMilliseconds(millisecondsTimeout))
          return false;
      }
      await Task.Delay(100, ct);
    }
    return true;
  }

  public async Task<bool> WaitOne(CancellationToken ct)
  {
    while (!_semaphore.WaitOne(0))
    {
      ct.ThrowIfCancellationRequested();
      await Task.Delay(100, ct);
    }
    return true;
  }

  public SemaphoreSecurity GetAccessControl()
  {
    return _semaphore.GetAccessControl();
  }

  public int Release()
  {
    return _semaphore.Release();
  }

  public int Release(int releaseCount)
  {
    return _semaphore.Release(releaseCount);
  }

  public void SetAccessControl(SemaphoreSecurity semaphoreSecurity)
  {
    _semaphore.SetAccessControl(semaphoreSecurity);
  }


  #region IDisposable Support
  private bool disposedValue = false; // To detect redundant calls
  void Dispose(bool disposing)
  {
    if (!disposedValue)
    {
      if (disposing)
      {
        // TODO: dispose managed state (managed objects).
        if (_semaphore != null)
        {
          _semaphore.Dispose();
          _semaphore = null;
        }
      }
      disposedValue = true;
    }
  }

  // This code added to correctly implement the disposable pattern.
  public void Dispose()
  {
    // Do not change this code. Put cleanup code in Dispose(bool disposing) above.
    Dispose(true);
  }
  #endregion

}
\$\endgroup\$
  • \$\begingroup\$ You could simply use a ConcurrentDictionary to store SemaphoreSlims. SemaphoreSlim already supports async operations. \$\endgroup\$ – Limeray May 22 at 9:51
  • \$\begingroup\$ The challenge is that SemaphoreSlim is limited to only a single application, we're specifically looking for async locking wrapper for the Semaphore class across multiple applications using a named semaphore (which SemaphoreSlim can't do). \$\endgroup\$ – Gary Smith May 23 at 16:13
3
\$\begingroup\$

You could use ThreadPool.RegisterWaitForSingleObject to register a callback when a the WaitHandle (Semaphore extends WaitHandle) is signaled.

Together with a TaskCompletionSource you could completely remove all your wait loops.

Example:

private async Task Run()
{
    var semaphore = new Semaphore(0, 1);
    await AwaitWaitHandle(semaphore, CancellationToken.None, TimeSpan.FromMilliseconds(-1));
}

private Task AwaitWaitHandle(WaitHandle handle, CancellationToken cancellationToken, TimeSpan timeout)
{
    var taskCompletionSource = new TaskCompletionSource<bool>();

    var reg = ThreadPool.RegisterWaitForSingleObject(handle,
        (state, timedOut) =>
        {
            // Handle timeout
            if (timedOut)
                taskCompletionSource.TrySetCanceled();

            taskCompletionSource.TrySetResult(true);
        }, null, timeout, true);

    // Handle cancellation
    cancellationToken.Register(() =>
    {
        reg.Unregister(handle);
        taskCompletionSource.TrySetCanceled();
    });

    return taskCompletionSource.Task;
}

You could use AwaitWaitHandle in your SemaphoreAsync implementation to await the Semaphore.

\$\endgroup\$
  • \$\begingroup\$ This is an interesting approach. I'm going to try this and the other method presented here. At first I thought there would be issues with multiple calls on the same CancellationToken but I noticed that you're unregistered it on the actual event. \$\endgroup\$ – Gary Smith May 30 at 18:14
2
\$\begingroup\$

Perhaps you could rewrite the wait operations to await both the semaphore or cancellation token.

public async Task<bool> WaitOne(TimeSpan timeout, CancellationToken ct)
  {
    DateTime start = DateTime.UtcNow;
    while (!_semaphore.WaitOne(0))
    {
      ct.ThrowIfCancellationRequested();
      if (DateTime.UtcNow < start.Add(timeout))
        return false;
      await Task.Delay(100, ct);
    }
    return true;
  }
  public async Task<bool> WaitOne(TimeSpan timeout, CancellationToken ct)
  {
      var success = await Task.Run(() =>
      {
          return WaitHandle.WaitTimeout
              != WaitHandle.WaitAny(new[] { _semaphore, ct.WaitHandle }, timeout);
      });
      ct.ThrowIfCancellationRequested();
      return success;
  }
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.