Awaitable AutoResetEvent

Question

What would be the async (awaitable) equivalent of AutoResetEvent?

If in the classic thread synchronization we would use something like this:

    AutoResetEvent signal = new AutoResetEvent(false);

    void Thread1Proc()
    {
        //do some stuff
        //..
        //..

        signal.WaitOne(); //wait for an outer thread to signal we are good to continue

        //do some more stuff
        //..
        //..
    }

    void Thread2Proc()
    {
        //do some stuff
        //..
        //..

        signal.Set(); //signal the other thread it's good to go

        //do some more stuff
        //..
        //..
    }

I was hoping that in the new async way of doing things, something like this would come to be:

SomeAsyncAutoResetEvent asyncSignal = new SomeAsyncAutoResetEvent();

async void Task1Proc()
{
    //do some stuff
    //..
    //..

    await asyncSignal.WaitOne(); //wait for an outer thread to signal we are good to continue

    //do some more stuff
    //..
    //..
}

async void Task2Proc()
{
    //do some stuff
    //..
    //..

    asyncSignal.Set(); //signal the other thread it's good to go

    //do some more stuff
    //..
    //..
}

I've seen other custom made solutions, but what I've managed to get my hands on, at some point in time, still involves locking a thread. I don't want this just for the sake of using the new await syntax. I'm looking for a true awaitable signaling mechanism which does not lock any thread.

Is it something I'm missing in the Task Parallel Library?

EDIT: Just to make clear: SomeAsyncAutoResetEvent is an entirely made up class name used as a placeholder in my example.

Answer 1

If you want to build your own, Stephen Toub has the definitive blog post on the subject.

If you want to use one that's already written, I have one in my AsyncEx library. AFAIK, there's no other option as of the time of this writing.

Answer 2

Here's the source for Stephen Toub's AsyncAutoResetEvent, in case his blog goes offline.

public class AsyncAutoResetEvent
{
    private static readonly Task s_completed = Task.FromResult(true);
    private readonly Queue<TaskCompletionSource<bool>> m_waits = new Queue<TaskCompletionSource<bool>>();
    private bool m_signaled;

    public Task WaitAsync()
    {
        lock (m_waits)
        {
            if (m_signaled)
            {
                m_signaled = false;
                return s_completed;
            }
            else
            {
                var tcs = new TaskCompletionSource<bool>();
                m_waits.Enqueue(tcs);
                return tcs.Task;
            }
        }
    }

    public void Set()
    {
        TaskCompletionSource<bool> toRelease = null;

        lock (m_waits)
        {
            if (m_waits.Count > 0)
                toRelease = m_waits.Dequeue();
            else if (!m_signaled)
                m_signaled = true;
        }

        toRelease?.SetResult(true);
    }
}

Answer 3

I think there is good example on MSDN: https://msdn.microsoft.com/en-us/library/hh873178%28v=vs.110%29.aspx#WHToTap

public static Task WaitOneAsync(this WaitHandle waitHandle)
{
    if (waitHandle == null) 
        throw new ArgumentNullException("waitHandle");

    var tcs = new TaskCompletionSource<bool>();
    var rwh = ThreadPool.RegisterWaitForSingleObject(waitHandle, 
        delegate { tcs.TrySetResult(true); }, null, -1, true);
    var t = tcs.Task;
    t.ContinueWith( (antecedent) => rwh.Unregister(null));
    return t;
}

Answer 4

Here is a version I cooked up which allows you to specify a timeout. It is derived from Stephen Toub's solution. We currently use this in production workloads.

public class AsyncAutoResetEvent
{
    readonly LinkedList<TaskCompletionSource<bool>> waiters = 
        new LinkedList<TaskCompletionSource<bool>>();

    bool isSignaled;

    public AsyncAutoResetEvent(bool signaled)
    {
        this.isSignaled = signaled;
    }

    public Task<bool> WaitAsync(TimeSpan timeout)
    {
        return this.WaitAsync(timeout, CancellationToken.None);
    }

    public async Task<bool> WaitAsync(TimeSpan timeout, CancellationToken cancellationToken)
    {
        TaskCompletionSource<bool> tcs;

        lock (this.waiters)
        {
            if (this.isSignaled)
            {
                this.isSignaled = false;
                return true;
            }
            else if (timeout == TimeSpan.Zero)
            {
                return this.isSignaled;
            }
            else
            {
                tcs = new TaskCompletionSource<bool>();
                this.waiters.AddLast(tcs);
            }
        }

        Task winner = await Task.WhenAny(tcs.Task, Task.Delay(timeout, cancellationToken));
        if (winner == tcs.Task)
        {
            // The task was signaled.
            return true;
        }
        else
        {
            // We timed-out; remove our reference to the task.
            // This is an O(n) operation since waiters is a LinkedList<T>.
            lock (this.waiters)
            {
                bool removed = this.waiters.Remove(tcs);
                Debug.Assert(removed);
                return false;
            }
        }
    }

    public void Set()
    {
        lock (this.waiters)
        {
            if (this.waiters.Count > 0)
            {
                // Signal the first task in the waiters list. This must be done on a new
                // thread to avoid stack-dives and situations where we try to complete the
                // same result multiple times.
                TaskCompletionSource<bool> tcs = this.waiters.First.Value;
                Task.Run(() => tcs.SetResult(true));
                this.waiters.RemoveFirst();
            }
            else if (!this.isSignaled)
            {
                // No tasks are pending
                this.isSignaled = true;
            }
        }
    }

    public override string ToString()
    {
        return $"Signaled: {this.isSignaled.ToString()}, Waiters: {this.waiters.Count.ToString()}";
    }
}

Answer 5

I was also looking for an AsyncAutoResetEvent class and it seems there is now one available in namespace Microsoft.VisualStudio.Threading

// Summary:
//     An asynchronous implementation of an AutoResetEvent.
[DebuggerDisplay("Signaled: {signaled}")]
public class AsyncAutoResetEvent

Answer 6

It also works, but this way may fade the purpose of using async and await.

AutoResetEvent asyncSignal = new AutoResetEvent();

async void Task1Proc()
{
    //do some stuff
    //..
    //..

    await Task.Run(() => asyncSignal.WaitOne()); //wait for an outer thread to signal we are good to continue

    //do some more stuff
    //..
    //..
}

Answer 7

This could be much simpler implementation with SemaphoreSlim, I haven't tested this in production yet but it should work fine.

public class AsyncAutoResetEvent : IDisposable
{
    private readonly SemaphoreSlim _waiters;

    public AsyncAutoResetEvent(bool initialState)
        => _waiters = new SemaphoreSlim(initialState ? 1 : 0, 1);

    public Task<bool> WaitOneAsync(TimeSpan timeout, CancellationToken cancellationToken = default)
        => _waiters.WaitAsync(timeout, cancellationToken);

    public Task WaitOneAsync(CancellationToken cancellationToken = default)
        => _waiters.WaitAsync(cancellationToken);

    public void Set()
    {
        lock (_waiters)
        {
            if (_waiters.CurrentCount == 0)
                _waiters.Release();
        }
    }

    public override string ToString()
        => $"Signaled: {_waiters.CurrentCount != 0}"; 

    private bool _disposed;
    protected virtual void Dispose(bool disposing)
    {
        if (!_disposed)
        {
            if (disposing)
            {
                _waiters.Dispose();
            }

            _disposed = true;
        }
    }

    public void Dispose()
    {
        // Do not change this code. Put cleanup code in 'Dispose(bool disposing)' method
        Dispose(disposing: true);
        GC.SuppressFinalize(this);
    }
}

Answer 8

Install the Microsoft.VisualStudio.Threading package (i did it using Nuget), then you will be able to use the AsyncAutoResetEvent class, which by its own documentation states:

An asynchronous implementation of an AutoResetEvent.

docs: https://learn.microsoft.com/en-us/dotnet/api/microsoft.visualstudio.threading.asyncautoresetevent

Answer 9

I extended the example from MSDN provided by Oleg Gordeev with an optional Timeout (ms):

public static Task WaitOneAsync(this WaitHandle waitHandle, double timeout = 0)
        {
            if (waitHandle == null) throw new ArgumentNullException("waitHandle");

            var tcs = new TaskCompletionSource<bool>();

            if (timeout > 0) 
            {
                var timer = new System.Timers.Timer(timeout) 
                { Enabled = true, AutoReset = false };

                ElapsedEventHandler del = default;
                del = delegate (object x, System.Timers.ElapsedEventArgs y)
                {
                    tcs.TrySetResult(true);
                    timer.Elapsed -= del; 
                    timer.Dispose();
                };

                timer.Elapsed += del;
            }
        
            var rwh = ThreadPool.RegisterWaitForSingleObject(waitHandle,
                      delegate { tcs.TrySetResult(true); },
                      null, -1, true);

            var t = tcs.Task;
            t.ContinueWith((antecedent) => rwh.Unregister(null));

            return t;
        }

Answer 10

Here is my version of one-time event can be await by multiple threads. It internally relies on BoundedChannel.

public class AsyncOneTimeEvent<T>
{
    private T Result { get; set; }

    private readonly Channel<bool> _channel = Channel.CreateBounded<bool>(new BoundedChannelOptions(1)
    {
        SingleReader = false,
        SingleWriter = true,
        FullMode = BoundedChannelFullMode.DropWrite,
    });

    public async Task<T> GetResult()
    {
        await _channel.Reader.WaitToReadAsync().ConfigureAwait(false);

        return this.Result;
    }

    public void SetResult(T result)
    {
        this.Result = result;
        _channel.Writer.Complete();
    }

    public void SetError(Exception ex)
    {
        _channel.Writer.Complete(ex);
    }
}

Answer 11

Here's my COMPLETE implementation using SemaphoreSlim, using all SemaphoreSlim.WaitAsync overrides:

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

/// <summary>
/// Represents an event that, when signaled, resets automatically after releasing a single waiting task.
/// </summary>
public sealed class AutoResetEventAsync : IDisposable {

    /// <summary>
    /// Waits asynchronously until a signal is received.
    /// </summary>
    /// <returns>Task completed when the event is signaled.</returns>
    public async ValueTask WaitAsync() {
        if (CheckSignaled()) return;
        SemaphoreSlim s;
        lock (Q) Q.Enqueue(s = new(0, 1));
        await s.WaitAsync();
        lock (Q) if (Q.Count > 0 && Q.Peek() == s) Q.Dequeue().Dispose();
    }

    /// <summary>
    /// Waits asynchronously until a signal is received or the time runs out.
    /// </summary>
    /// <param name="millisecondsTimeout">The number of milliseconds to wait, <see cref="System.Threading.Timeout.Infinite"/>
    /// (-1) to wait indefinitely, or zero to return immediately.</param>
    /// <returns>Task completed when the event is signaled or the time runs out.</returns>
    public async ValueTask WaitAsync(int millisecondsTimeout) {
        if (CheckSignaled()) return;
        SemaphoreSlim s;
        lock (Q) Q.Enqueue(s = new(0, 1));
        await s.WaitAsync(millisecondsTimeout);
        lock (Q) if (Q.Count > 0 && Q.Peek() == s) Q.Dequeue().Dispose();
    }

    /// <summary>
    /// Waits asynchronously until a signal is received, the time runs out or the token is cancelled.
    /// </summary>
    /// <param name="millisecondsTimeout">The number of milliseconds to wait, <see cref="System.Threading.Timeout.Infinite"/>
    /// (-1) to wait indefinitely, or zero to return immediately.</param>
    /// <param name="cancellationToken">The <see cref="System.Threading.CancellationToken"/> to observe.</param>
    /// <returns>Task completed when the event is signaled, the time runs out or the token is cancelled.</returns>
    public async ValueTask WaitAsync(int millisecondsTimeout, CancellationToken cancellationToken) {
        if (CheckSignaled()) return;
        SemaphoreSlim s;
        lock (Q) Q.Enqueue(s = new(0, 1));
        try {
            await s.WaitAsync(millisecondsTimeout, cancellationToken);
        }
        finally {
            lock (Q) if (Q.Count > 0 && Q.Peek() == s) Q.Dequeue().Dispose();
        }
    }

    /// <summary>
    /// Waits asynchronously until a signal is received or the token is cancelled.
    /// </summary>
    /// <param name="cancellationToken">The <see cref="System.Threading.CancellationToken"/> to observe.</param>
    /// <returns>Task completed when the event is signaled or the token is cancelled.</returns>
    public async ValueTask WaitAsync(CancellationToken cancellationToken) {
        if (CheckSignaled()) return;
        SemaphoreSlim s;
        lock (Q) Q.Enqueue(s = new(0, 1));
        try {
            await s.WaitAsync(cancellationToken);
        }
        finally {
            lock (Q) if (Q.Count > 0 && Q.Peek() == s) Q.Dequeue().Dispose();
        }
    }

    /// <summary>
    /// Waits asynchronously until a signal is received or the time runs out.
    /// </summary>
    /// <param name="timeout">A <see cref="System.TimeSpan"/> that represents the number of milliseconds to wait,
    /// a <see cref="System.TimeSpan"/> that represents -1 milliseconds to wait indefinitely, or a System.TimeSpan
    /// that represents 0 milliseconds to return immediately.</param>
    /// <returns>Task completed when the event is signaled or the time runs out.</returns>
    public async ValueTask WaitAsync(TimeSpan timeout) {
        if (CheckSignaled()) return;
        SemaphoreSlim s;
        lock (Q) Q.Enqueue(s = new(0, 1));
        await s.WaitAsync(timeout);
        lock (Q) if (Q.Count > 0 && Q.Peek() == s) Q.Dequeue().Dispose();
    }

    /// <summary>
    /// Waits asynchronously until a signal is received, the time runs out or the token is cancelled.
    /// </summary>
    /// <param name="timeout">A <see cref="System.TimeSpan"/> that represents the number of milliseconds to wait,
    /// a <see cref="System.TimeSpan"/> that represents -1 milliseconds to wait indefinitely, or a System.TimeSpan
    /// that represents 0 milliseconds to return immediately.</param>
    /// <param name="cancellationToken">The <see cref="System.Threading.CancellationToken"/> to observe.</param>
    /// <returns>Task completed when the event is signaled, the time runs out or the token is cancelled.</returns>
    public async ValueTask WaitAsync(TimeSpan timeout, CancellationToken cancellationToken) {
        if (CheckSignaled()) return;
        SemaphoreSlim s;
        lock (Q) Q.Enqueue(s = new(0, 1));
        try {
            await s.WaitAsync(timeout, cancellationToken);
        }
        finally {
            lock (Q) if (Q.Count > 0 && Q.Peek() == s) Q.Dequeue().Dispose();
        }
    }

    /// <summary>
    /// Sets the state of the event to signaled, allowing one or more waiting tasks to proceed.
    /// </summary>
    public void Set() {
        SemaphoreSlim? toRelease = null;
        lock (Q) {
            if (Q.Count > 0) toRelease = Q.Dequeue();
            else if (!IsSignaled) IsSignaled = true;
        }
        toRelease?.Release();
    }

    /// <summary>
    /// Sets the state of the event to non nonsignaled, making the waiting tasks to wait.
    /// </summary>
    public void Reset() => IsSignaled = false;

    /// <summary>
    /// Disposes any semaphores left in the queue.
    /// </summary>
    public void Dispose() {
        lock (Q) {
            while (Q.Count > 0) Q.Dequeue().Dispose();
        }
    }

    /// <summary>
    /// Checks the <see cref="IsSignaled"/> state and resets it when it's signaled.
    /// </summary>
    /// <returns>True if the event was in signaled state.</returns>
    private bool CheckSignaled() {
        lock (Q) {
            if (IsSignaled) {
                IsSignaled = false;
                return true;
            }
            return false;
        }
    }

    private readonly Queue<SemaphoreSlim> Q = new();
    private volatile bool IsSignaled;

}

I used SemaphoreSlim because it gives the time out and cancellation token support "for free". It could be even better if I just modified the original .NET source code of SemaphoreSlim to behave like AutoResetEvent but nah, that's it. Let me know if you find any bugs.