// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.
namespace Microsoft.VisualStudio.Threading
{
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
///
/// An asynchronous like class with more convenient release syntax.
///
///
/// This semaphore guarantees FIFO ordering.
///
/// This object does *not* need to be disposed of, as it does not hold unmanaged resources.
/// Disposing this object has no effect on current users of the semaphore, and they are allowed to release their hold on the semaphore without exception.
/// An is thrown back at anyone asking to or waiting to enter the semaphore after is called.
///
///
public class AsyncSemaphore : IDisposable
{
///
/// A task that is faulted with an .
///
private static readonly Task DisposedReleaserTask = TplExtensions.FaultedTask(new ObjectDisposedException(typeof(AsyncSemaphore).FullName));
///
/// A task that is canceled without a specific token.
///
private static readonly Task CanceledReleaser = Task.FromCanceled(new CancellationToken(true));
///
/// A task to return for any uncontested request for the lock.
///
private readonly Task uncontestedReleaser;
///
/// The sync object to lock on for mutable field access.
///
private readonly object syncObject = new object();
///
/// A queue of operations waiting to enter the semaphore.
///
private readonly LinkedList waiters = new LinkedList();
///
/// A pool of recycled nodes.
///
private readonly Stack> nodePool = new Stack>();
///
/// A value indicating whether this instance has been disposed.
///
private bool disposed;
///
/// Initializes a new instance of the class.
///
/// The initial number of requests for the semaphore that can be granted concurrently.
public AsyncSemaphore(int initialCount)
{
this.CurrentCount = initialCount;
this.uncontestedReleaser = Task.FromResult(new Releaser(this));
}
///
/// Gets the number of openings that remain in the semaphore.
///
public int CurrentCount { get; private set; }
///
/// Requests access to the lock.
///
/// A token whose cancellation signals lost interest in the lock.
///
/// A task whose result is a releaser that should be disposed to release the lock.
/// This task may be canceled if is signaled.
///
/// Thrown when is canceled before semaphore access is granted.
/// Thrown when this semaphore is disposed before semaphore access is granted.
public Task EnterAsync(CancellationToken cancellationToken = default) => this.EnterAsync(Timeout.InfiniteTimeSpan, cancellationToken);
///
/// Requests access to the lock.
///
/// A timeout for waiting for the lock.
/// A token whose cancellation signals lost interest in the lock.
///
/// A task whose result is a releaser that should be disposed to release the lock.
/// This task may be canceled if is signaled or expires.
///
/// Thrown when is canceled or the expires before semaphore access is granted.
/// Thrown when this semaphore is disposed before semaphore access is granted.
public Task EnterAsync(TimeSpan timeout, CancellationToken cancellationToken = default)
{
if (cancellationToken.IsCancellationRequested)
{
return Task.FromCanceled(cancellationToken);
}
lock (this.syncObject)
{
if (this.disposed)
{
return DisposedReleaserTask;
}
if (this.CurrentCount > 0)
{
this.CurrentCount--;
return this.uncontestedReleaser;
}
else if (timeout == TimeSpan.Zero)
{
return CanceledReleaser;
}
else
{
WaiterInfo info = new WaiterInfo(this, cancellationToken);
LinkedListNode? node = this.GetNode(info);
// Careful: consider that if the token was cancelled just now (after we checked it on entry to this method)
// or the timeout expires,
// then this Register method may *inline* the handler we give it, reversing the apparent order of execution with respect to
// the code that follows this Register call.
info.CancellationTokenRegistration = cancellationToken.Register(s => CancellationHandler(s), info);
if (timeout != Timeout.InfiniteTimeSpan)
{
info.TimerTokenSource = new Timer(s => CancellationHandler(s), info, checked((int)timeout.TotalMilliseconds), Timeout.Infinite);
}
// Only add to the queue if cancellation hasn't already happened.
if (!info.Trigger.Task.IsCanceled)
{
this.waiters.AddLast(node);
info.Node = node;
}
else
{
// Make sure we don't leak the Timer if cancellation happened before we created it.
info.Cleanup();
// Also recycle the unused node.
this.RecycleNode(node);
}
return info.Trigger.Task;
}
}
}
///
/// Requests access to the lock.
///
/// A timeout for waiting for the lock (in milliseconds).
/// A token whose cancellation signals lost interest in the lock.
/// A task whose result is a releaser that should be disposed to release the lock.
/// Thrown when is canceled or the expires before semaphore access is granted.
/// Thrown when this semaphore is disposed before semaphore access is granted.
public Task EnterAsync(int timeout, CancellationToken cancellationToken = default) => this.EnterAsync(TimeSpan.FromMilliseconds(timeout), cancellationToken);
///
/// Faults all pending semaphore waiters with
/// and rejects all subsequent attempts to enter the semaphore with the same exception.
///
public void Dispose()
{
this.Dispose(true);
GC.SuppressFinalize(this);
}
///
/// Disposes managed and unmanaged resources held by this instance.
///
/// true if was called; false if the object is being finalized.
protected virtual void Dispose(bool disposing)
{
if (disposing)
{
List? waitersCopy = null;
lock (this.syncObject)
{
this.disposed = true;
if (this.waiters.Count > 0)
{
waitersCopy = new List(this.waiters.Count);
while (this.waiters.First is { } head)
{
head.Value.Trigger.TrySetException(new ObjectDisposedException(this.GetType().FullName));
waitersCopy.Add(head.Value);
this.waiters.RemoveFirst();
head.Value.Node = null;
}
}
this.nodePool.Clear();
}
if (waitersCopy is object)
{
foreach (WaiterInfo? waitInfo in waitersCopy)
{
waitInfo.Cleanup();
}
}
}
}
private static void CancellationHandler(object? state)
{
var waiterInfo = (WaiterInfo)state!;
// The party that manages to complete or cancel the task is responsible to remove it from the queue.
if (waiterInfo.Trigger.TrySetCanceled(waiterInfo.CancellationToken.IsCancellationRequested ? waiterInfo.CancellationToken : new CancellationToken(true)))
{
// If the node is in the queue, remove it.
// It might not have been added yet if cancellation was already requested by the time we called Register.
lock (waiterInfo.Owner.syncObject)
{
if (waiterInfo.Node is { } node)
{
waiterInfo.Owner.waiters.Remove(node);
waiterInfo.Owner.RecycleNode(node);
}
}
}
// Clear registration and references.
waiterInfo.Cleanup();
}
private void Release()
{
WaiterInfo? info = null;
lock (this.syncObject)
{
if (this.CurrentCount++ == 0)
{
// We loop because the First node may have been canceled.
while (this.waiters.First is { } head)
{
// Remove the head of the queue.
this.waiters.RemoveFirst();
info = head.Value;
this.RecycleNode(head);
if (info.Trigger.TrySetResult(new Releaser(this)))
{
// We successfully let someone enter the semaphore.
this.CurrentCount--;
// We've filled the one slot available in the semaphore. Stop looking for more.
break;
}
}
}
}
// Release memory related to cancellation handling.
info?.Cleanup();
}
private void RecycleNode(LinkedListNode node)
{
Assumes.True(Monitor.IsEntered(this.syncObject));
node.Value.Node = null;
if (this.nodePool.Count < 10)
{
LinkedListNode nullableNode = node!;
nullableNode.Value = null;
this.nodePool.Push(nullableNode);
}
}
private LinkedListNode GetNode(WaiterInfo info)
{
Assumes.True(Monitor.IsEntered(this.syncObject));
if (this.nodePool.Count > 0)
{
LinkedListNode? node = this.nodePool.Pop();
node.Value = info;
return node!;
}
return new LinkedListNode(info);
}
///
/// A value whose disposal triggers the release of a lock.
///
public readonly struct Releaser : IDisposable
{
///
/// The lock instance to release.
///
private readonly AsyncSemaphore? toRelease;
///
/// Initializes a new instance of the struct.
///
/// The lock instance to release on.
internal Releaser(AsyncSemaphore toRelease)
{
this.toRelease = toRelease;
}
///
/// Releases the lock.
///
public void Dispose()
{
if (this.toRelease is object)
{
this.toRelease.Release();
}
}
}
private class WaiterInfo
{
internal WaiterInfo(AsyncSemaphore owner, CancellationToken cancellationToken)
{
this.Owner = owner;
this.CancellationToken = cancellationToken;
}
internal LinkedListNode? Node { get; set; }
internal AsyncSemaphore Owner { get; }
internal TaskCompletionSource Trigger { get; } = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
internal CancellationToken CancellationToken { get; }
internal CancellationTokenRegistration CancellationTokenRegistration { private get; set; }
internal IDisposable? TimerTokenSource { private get; set; }
internal void Cleanup()
{
CancellationTokenRegistration cancellationTokenRegistration;
IDisposable? timerTokenSource;
lock (this)
{
cancellationTokenRegistration = this.CancellationTokenRegistration;
this.CancellationTokenRegistration = default;
timerTokenSource = this.TimerTokenSource;
this.TimerTokenSource = null;
}
cancellationTokenRegistration.Dispose();
timerTokenSource?.Dispose();
}
}
}
}