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.

Solution 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.

Solution 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);
    }
}

Solution 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;
}

Solution 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()}";
    }
}

View more solutions

Comments

[Dark Falcon]

For one-time-use, a TaskCompletionSource whose result is ignored by the awaiting task.

[Matthew Watson]

gist.github.com/AArnott/1084951 maybe?

[Mihai Caracostea]

@MatthewWatson I see it uses a lock, which will block a thread from the thread pool. I was hoping for something not involving a blocked thread.

[Dark Falcon]

A lock doesn't necessarily mean a thread is blocked.

[Mihai Caracostea]

@DarkFalcon True. And in this case it might even not block any thread.

[Mihai Caracostea]

@MatthewWatson This extension method uses ThreadPool.RegisterWaitForSingleObject which "Registers a delegate that is waiting for a WaitHandle.". I understand from this bit that a thread from the pool will block on that waithandle. Am I wrong?

[Scott Chamberlain]

Why wouldn't a new SemaphoreSlim(1) work, WaitOne() is WaitAsync() and Set() becomes Release()

[Stephen Cleary]

AREs and Semaphores are very similar (though usually used differently). The semantic difference comes in if the primitive is signalled when it is already set.

[HelloSam]

I think this.waiters should be lock'ed at the in the Remove(tcs) manipulation path?

[Chris Gillum]

@HelloSam I think you are right! Fixed. Thanks for pointing this out.

[Andy]

I don't have a lot of time to debug this, but, be forewarned: i am getting dead-lock using this. When a new thread calls event.Set(), it hangs on toRelease.SetResult(true);

[Chris Gillum]

@Andy thanks for the comment. There is an additional fix I made since I originally posted this which I suspect addresses your deadlock (in my case, it was a StackOverflowException). The fix was to wrap the SetResult(true) call in a Task.Run(...).

[Ashley Jackson]

anything wrong with await Task.Run(() => loginWaiter.WaitOne(TimeSpan.FromSeconds(75)));

[Stephen Cleary]

@AshleyJackson: That approach does use another thread. Some synchronization primitives do not allow this (e.g., Mutex, Monitor), but since this is an AutoResetEvent, it should work.

[M.kazem Akhgary]

I think those who are named "Stephen" are born for asynchronous anything.

[user1713059]

Why can you use regular lock in awaitable code? Can't the same task continue as a different thread here and go around the lock?

[Drew Noakes]

@user1713059 note that WaitAsync isn't actually an async method. That means it doesn't yield control midway through processing. Instead, it obtains a Task from the TaskCompletionSource and returns it before releasing the lock.

[user1713059]

Ah sure, so even if I do "await WaitAsync()" it is sure that the whole method gets executed by the same thread, because it's not actually async - is that right? The "Async" method suffix led me astray, but from what I see it's used in methods without the "async" keyword too.

[Drew Noakes]

It's still an asynchronous method because it returns a task which may not be completed by the time the method returns. However the method is not async, which means the method won't yield at some point within its body while it awaits some other Task's completion. It's convention for methods that return Task (or Task<T>) to have an Async suffix.

[Drew Noakes]

With respect to your original comment, the lock is released before the Task is returned to the caller, so there's no way for that caller to get around the lock.

[Klepto]

Stephen Toubs post seems to have been moved here

[Theodor Zoulias]

Using a Channel as a substitute of a TaskCompletionSource seems like a clever idea. But is also unnecessary, and the implementation seems susceptible to visibility problems. I am not sure that all threads will "see" the latest value of the non-volatile private T Result field in all cases.

[Theodor Zoulias]

Example: Thread A enters the GetResult() method, reads the value of Result in an out-of-order fashion, and then gets suspended by the OS. Thread B enters and exits the SetResult method. Thread A resumes, executes synchronously the await _channel.Reader.WaitToReadAsync() line, and returns a Task having default(T) as its value. Is this scenario impossible based on the C# ECMA-334 specification? I have no idea!

[Mr.Wang from Next Door]

@TheodorZoulias certainly it works, you can try it online : dotnetfiddle.net/uyQRG1

[Theodor Zoulias]

I am sure that it works. I am not sure that it is guaranteed to work correctly on all CPU architectures. Visibility problems are notoriously difficult to debug. You can read this article by Igor Ostrovsky to get an idea why.

[Yarek T]

Why is this considered bad?

[Test]

@YarekT I remembered the reason at the time I wrote this answer months ago, but not now. I don't think this is bad, though there are more than one context switching (by WaitOne() and by await keyword) performance issue in this.

[Yarek T]

No worries. I've been recently looking more into Tasks in C#. From what I can gather its bad because it wastes a thread by creating one, then immediately making it blocked by the wait. I've seen a a few solutions floating around that avoid this by somehow using a timer, but they all seem very complicated. Anyway, heres an upvote

[Felix K.]

Definitely the best answer.

[Theodor Zoulias]

Is the AutoResetEventAsync class thread-safe? If yes, what can happen if two threads call WaitAsync() at the same time? Isn't it possible that both will read the IsSignaled field as true, before any of them executes the IsSignaled = false; line? Also the if (Q.Contains(s)) Q.Dequeue().Dispose(); line searches if the s exists in the queue, and then dequeues and disposes some other semaphore (most likely). Is this intentional?

[Harry]

@TheodorZoulias : Yes, because even if 2 threads can enter WaitAsync at the same time, they cannot pass the Q access at the same time. Notice that Q can be only accessed with a single thread. That makes the flow simple and direct. This also implies the internal await is accessible to a single thread only. Thus, it is impossible invalid semaphore is dequeued. The multiple tests I performed on this class haven't yet fail, but this doesn't prove it's valid. I think that the single tread access to Q does.

[Theodor Zoulias]

I am talking about this line: if (IsSignaled) { IsSignaled = false; return; }. This is not protected by a lock. The IsSignaled is not even a volatile field. As for the if (Q.Contains(s)), if you are sure that the s can only be in the head of the queue, if (Q.Peak() == s) would be faster and more expressive regarding the intentions of the code. Btw what will happen if the cancellationToken is canceled and the WaitAsync throws?

[Harry]

You've found some interesting edge cases. I'll try to fix them and edit my example... BRB.

[Harry]

Fixed. I had to add if (Q.Count > 0) because Peek() throws when Q is empty, and it is empty in most common case the Set() is invoked.

[Theodor Zoulias]

Your implementation transitioned from certainly flawed to potentially correct. But after closer inspection one race condition still exists (and maybe more). Assuming an initial state IsSignaled = false, one thread invokes the WaitAsync and another invokes the Set. The expected behavior after both invocations is that the first thread will not be in a waiting state. Alas your implementation allows this to happen, in case the Set is invoked by the second thread just after the first thread has executed the CheckSignaled(), and just before executing the Q.Enqueue(s = new(0, 1)).

[Theodor Zoulias]

Btw I am not sure what the Reset method is supposed to do, but mutating a non-volatile field without synchronization opens the can of worms that is called memory models. Are you an expert in coherency protocols, and can guarantee that this is not a problem? If not, then neither am I, and so there is no chance that I would use your class (as is) in a production environment.

[Harry]

The Reset seems to be redundant here, so it goes. About the race condition - I give up. WaitAsync can't block Set(). So how could I possibly prevent that particular condition from happening? BTW, that would happen only if I invoke Set() parallelly with WaitAsync() that would be just asking for a race condition. In my code I use this class to proceed when I/O operation completes. This event is used for one client session, so this will never happen. The I/O operation will never be completed before WaitAsync() enters Enqueue() part. Even if it throws.

[Theodor Zoulias]

Harry since you realize that implementing a custom asynchronous AutoResetEvent is challenging, why are you trying to roll your own, and don't just use Stephen Cleary's AsyncAutoResetEvent?

[Harry]

BTW, this class is not intended to be used to synchronize threads. Mixing manual thread synchronization with async pattern is not a good idea, because I think it's just extremely difficult to do correctly. That's probably why there is no WaitAsync() method in AutoResetEvent class in the first place.

[Harry]

Why? Because I needed a version that supports timeout and CancellationToken. I just needed it for my code, I use it and I just thought I could share it.

[Theodor Zoulias]

Hmm, Stephen Cleary's AsyncAutoResetEvent supports CancellationTokens, so the timeout functionality can be built on top of that. I just posted here an extension method that supports timeout and CancellationToken (optionally). Btw when you have an asynchronous implementation, implementing the synchronous functionality is trivial. Just call the asynchronous with .GetAwaiter.GetResult(). It's not very efficient, but gets the job done.

[Harry]

Thank you for the insight, I will use the better version with your extension then. Anyway, it was totally worth it as a learning experience. You're MVP.

[Hugh Jeffner]

Am I mistaken or is it not auto-resetting where it returns true after if (winner == tcs.Task)?

[Hugh Jeffner]

Never mind, I didn't understand how it works. Turns out if there are tasks waiting it never bothers setting isSignaled

[vpalmu]

Note that this only works correctly on ManualResetEvent not AutoResetEvent. On AutoResetEvent you need to WaitOne the thing inside the delegate; else the event is still signalled next time somebody calls WaitOne on it.