Building Async Coordination Primitives, Part 1: AsyncManualResetEvent
The Task-based Async Pattern (TAP) isn’t just about asynchronous operations that you initiate and then asynchronously wait for to complete. More generally, tasks can be used to represent all sorts of happenings, enabling you to await for any matter of condition to occur. We can even use Tasks to build simple coordination primitives like their synchronous counterparts but that allow the waiting to be done asynchronously.
One of the more basic coordination primitives is an event, and there are a few of these in the .NET Framework. ManualResetEvent and AutoResetEvent wrap their Win32 counterparts, and then more recently .NET 4 saw the addition of ManualResetEventSlim, which is a lighter-weight version of ManualResetEvent. An event is something that one party can wait on for another party to signal. In the case of a manual-reset event, the event remains signaled after it’s been set and until it’s explicitly reset; until it is reset, all waits on the event succeed.
TaskCompletionSource<TResult> is itself a form of an event, just one without a reset. It starts in the non-signaled state; its Task hasn’t been completed, and thus all waits on the Task won’t complete until the Task is completed. Then the {Try}Set* methods act as a signal, moving the Task into the completed state, such that all waits complete. Thus, we can easily build an AsyncManualResetEvent on top of TaskCompletionSource<TResult>; the only behavior we’re really missing is the “reset” capability, which we’ll provide by swapping in a new TaskCompletionSource<TResult> instance.
Here’s the shape of the type we’re aiming to build:
public class AsyncManualResetEvent
{
public Task WaitAsync();
public void Set();
public void Reset();
}
WaitAsync and Set are both easy. Wrapping a TaskCompletionSource<bool>, Set will complete the TaskCompletionSource<bool> with TrySetResult, and WaitAsync will return the completion source’s Task:
public class AsyncManualResetEvent
{
private volatile TaskCompletionSource<bool> m_tcs = new TaskCompletionSource<bool>();public Task WaitAsync() { return m_tcs.Task; }
public void Set() { m_tcs.TrySetResult(true); }
…
}
That leaves just the Reset method. Our goal for Reset is to make the Tasks returned from subsequent calls to WaitAsync not completed, and since Tasks never transition from completed to not-completed, we need to swap in a new TaskCompletionSource<bool>. In doing so, though, we need to make sure that, if multiple threads are calling Reset, Set, and WaitAsync concurrently, no Tasks returned from WaitAsync are orphaned (meaning that we wouldn’t want someone to call WaitAsync and get back a Task that won’t be completed the next time someone calls Set). To achieve that, we’ll make sure to only swap in a new Task if the current one is already completed, and we’ll make sure that we do the swap atomically. (There are of course other policies that would be valid here; this is simply the one I’ve chosen for this particular example.)
public class AsyncManualResetEvent
{
private volatile TaskCompletionSource<bool> m_tcs = new TaskCompletionSource<bool>();public Task WaitAsync() { return m_tcs.Task; }
public void Set() { m_tcs.TrySetResult(true); }
public void Reset()
{
while (true)
{
var tcs = m_tcs;
if (!tcs.Task.IsCompleted ||
Interlocked.CompareExchange(ref m_tcs, new TaskCompletionSource<bool>(), tcs) == tcs)
return;
}
}
}
With that, our type is done. However, there is one more potentially important behavior to keep in mind. In previous posts, I’ve talked about continuations and how they can be made to execute synchronously, meaning that the continuation will execute as part of the task’s completion, synchronously on the same thread that’s completed the task. In the case of TaskCompletionSource<TResult>, that means that synchronous continuations can happen as part of a call to {Try}Set*, which means in our AsyncManualResetEvent example, those continuations could execute as part of the Set method. Depending on your needs (and whether callers of Set may be ok with a potentially longer-running Set call as all synchronous continuations execute), this may or may not be what you want. If you don’t want this to happen, there are a few alternative approaches. One approach is to run the completion asynchronously, having the call to set block until the task being completed is actually finished (not including the task’s synchronous continuations, just the task itself), e.g.
public void Set()
{
var tcs = m_tcs;
Task.Factory.StartNew(s => ((TaskCompletionSource<bool>)s).TrySetResult(true),
tcs, CancellationToken.None, TaskCreationOptions.PreferFairness, TaskScheduler.Default);
tcs.Task.Wait();
}
There are of course other possible approaches, and what you do depends on your needs.
Next time, we’ll take a look at implementing an async auto-reset event.
Light
Dark
9 comments
is it possible to extend this example exposing something like “WaitAsync(int msTimeout)” many thanks anyway for your interesting posts!!!
> is it possible to extend this example exposing something like “WaitAsync(int msTimeout)”
Sure, e.g. https://github.com/dotnet/corefx/issues/33008
Hi Stephen!
Suppose we have too many threads calling the Set method.
Could this cause performance issues?
You mean lots of threads all concurrently invoking Set? It does involve synchronization, but it’s about as light as synchronization point as you’ll find, basically one interlocked operation.
Is it possible to extend this example to use a cancellation token for the wait?
Would this be an acceptable way to do it?
public Task WaitAsync(CancellationToken cancellationToken = default(CancellationToken))
{
var task = m_tcs.Task;
if (cancellationToken != null)
{
cancellationToken.Register(() => { m_tcs.TrySetCanceled(); });
}
return task;
}
> Is it possible to extend this example to use a cancellation token for the wait?
The blog post at https://devblogs.microsoft.com/pfxteam/how-do-i-cancel-non-cancelable-async-operations/ gives examples of implementing a `WithCancellation` method. With that, your desired WaitAsync could just be:
> Would this be an acceptable way to do it?
That has a few issues:
1. It’s going to cancel the actual task associated with this instance, which may or may not be the behavior you want.
2. Register returns a disposable instance which lets you unregister from the cancellationToken; not doing so in the successful case will end up leaking a registration into that cancellation token, which could be an issue if it’s longer-lived.
this comment has been deleted.
CompareExchange returns the original value in the ref location, not the new one. So you compare it against the snapshotted original value, as if they match, it will have been successfully replaced.