Async/await as a replacement of coroutines
Updated, a follow-up blog post: Asynchronous coroutines with C# 8.0 and IAsyncEnumerable.
I use C# iterators as a replacement for coroutines, and it has been working great. I want to switch to async/await as I think the syntax is cleaner and it gives me type safety...
IMO, it's a very interesting question, although it took me awhile to fully understand it. Perhaps, you didn't provide enough sample code to illustrate the concept. A complete app would help, so I'll try to fill this gap first. The following code illustrates the usage pattern as I understood it, please correct me if I'm wrong:
using System;
using System.Collections;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
namespace ConsoleApplication
{
// https://stackoverflow.com/q/22852251/1768303
public class Program
{
class Resource : IDisposable
{
public void Dispose()
{
Console.WriteLine("Resource.Dispose");
}
~Resource()
{
Console.WriteLine("~Resource");
}
}
private IEnumerator Sleep(int milliseconds)
{
using (var resource = new Resource())
{
Stopwatch timer = Stopwatch.StartNew();
do
{
yield return null;
}
while (timer.ElapsedMilliseconds < milliseconds);
}
}
void EnumeratorTest()
{
var enumerator = Sleep(100);
enumerator.MoveNext();
Thread.Sleep(500);
//while (e.MoveNext());
((IDisposable)enumerator).Dispose();
}
public static void Main(string[] args)
{
new Program().EnumeratorTest();
GC.Collect(GC.MaxGeneration, GCCollectionMode.Forced, true);
GC.WaitForPendingFinalizers();
Console.ReadLine();
}
}
}
Here, Resource.Dispose gets called because of ((IDisposable)enumerator).Dispose(). If we don't call enumerator.Dispose(), then we'll have to uncomment //while (e.MoveNext()); and let the iterator finish gracefully, for proper unwinding.
Now, I think the best way to implement this with async/await is to use a custom awaiter:
using System;
using System.Collections;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
namespace ConsoleApplication
{
// https://stackoverflow.com/q/22852251/1768303
public class Program
{
class Resource : IDisposable
{
public void Dispose()
{
Console.WriteLine("Resource.Dispose");
}
~Resource()
{
Console.WriteLine("~Resource");
}
}
async Task SleepAsync(int milliseconds, Awaiter awaiter)
{
using (var resource = new Resource())
{
Stopwatch timer = Stopwatch.StartNew();
do
{
await awaiter;
}
while (timer.ElapsedMilliseconds < milliseconds);
}
Console.WriteLine("Exit SleepAsync");
}
void AwaiterTest()
{
var awaiter = new Awaiter();
var task = SleepAsync(100, awaiter);
awaiter.MoveNext();
Thread.Sleep(500);
//while (awaiter.MoveNext()) ;
awaiter.Dispose();
task.Dispose();
}
public static void Main(string[] args)
{
new Program().AwaiterTest();
GC.Collect(GC.MaxGeneration, GCCollectionMode.Forced, true);
GC.WaitForPendingFinalizers();
Console.ReadLine();
}
// custom awaiter
public class Awaiter :
System.Runtime.CompilerServices.INotifyCompletion,
IDisposable
{
Action _continuation;
readonly CancellationTokenSource _cts = new CancellationTokenSource();
public Awaiter()
{
Console.WriteLine("Awaiter()");
}
~Awaiter()
{
Console.WriteLine("~Awaiter()");
}
public void Cancel()
{
_cts.Cancel();
}
// let the client observe cancellation
public CancellationToken Token { get { return _cts.Token; } }
// resume after await, called upon external event
public bool MoveNext()
{
if (_continuation == null)
return false;
var continuation = _continuation;
_continuation = null;
continuation();
return _continuation != null;
}
// custom Awaiter methods
public Awaiter GetAwaiter()
{
return this;
}
public bool IsCompleted
{
get { return false; }
}
public void GetResult()
{
this.Token.ThrowIfCancellationRequested();
}
// INotifyCompletion
public void OnCompleted(Action continuation)
{
_continuation = continuation;
}
// IDispose
public void Dispose()
{
Console.WriteLine("Awaiter.Dispose()");
if (_continuation != null)
{
Cancel();
MoveNext();
}
}
}
}
}
When it's time to unwind, I request the cancellation inside Awaiter.Dispose and drive the state machine to the next step (if there's a pending continuation). This leads to observing the cancellation inside Awaiter.GetResult (which is called by the compiler-generated code). That throws TaskCanceledException and further unwinds the using statement. So, the Resource gets properly disposed of. Finally, the task transitions to the cancelled state (task.IsCancelled == true).
IMO, this is a more simple and direct approach than installing a custom synchronization context on the current thread. It can be easily adapted for multithreading (some more details here).
This should indeed give you more freedom than with IEnumerator/yield. You could use try/catch inside your coroutine logic, and you can observe exceptions, cancellation and the result directly via the Task object.
Updated, AFAIK there is no analogy for the iterator's generated IDispose, when it comes to async state machine. You really have to drive the state machine to an end when you want to cancel/unwind it. If you want to account for some negligent use of try/catch preventing the cancellation, I think the best you could do is to check if _continuation is non-null inside Awaiter.Cancel (after MoveNext) and throw a fatal exception out-of-the-band (using a helper async void method).
Updated, this has evolved to a blog post: Asynchronous coroutines with C# 8.0 and IAsyncEnumerable.
It's 2020 and my other answer about await and coroutines is quite outdated by today's C# language standards. C# 8.0 has introduced support for asynchronous streams with new features like:
IAsyncEnumerableIAsyncEnumeratorawait foreachIAsyncDisposableawait using
To familiarize yourself with the concept of asynchronous streams, I could highly recommend reading "Iterating with Async Enumerables in C# 8", by Stephen Toub.
Together, these new features provide a great base for implementing asynchronous co-routines in C# in a much more natural way.
Wikipedia provides a good explanation of what co-routines (aka corotines) generally are. What I'd like to show here is how co-routines can be async, suspending their execution flow by using await and arbitrary swapping the roles of being producer/consumer to each other, with C# 8.0.
The code fragment below should illustrate the concept. Here we have two co-routines, CoroutineA and CoroutineB which execute cooperatively and asynchronously, by yielding to each other as their pseudo-linear execution flow goes on.
namespace Tests
{
[TestClass]
public class CoroutineProxyTest
{
const string TRACE_CATEGORY = "coroutines";
/// <summary>
/// CoroutineA yields to CoroutineB
/// </summary>
private async IAsyncEnumerable<string> CoroutineA(
ICoroutineProxy<string> coroutineProxy,
[EnumeratorCancellation] CancellationToken token)
{
await using (var coroutine = await coroutineProxy.AsAsyncEnumerator(token))
{
const string name = "A";
var i = 0;
// yielding 1
Trace.WriteLine($"{name} about to yeild: {++i}", TRACE_CATEGORY);
yield return $"{i} from {name}";
// receiving
if (!await coroutine.MoveNextAsync())
{
yield break;
}
Trace.WriteLine($"{name} received: {coroutine.Current}", TRACE_CATEGORY);
// yielding 2
Trace.WriteLine($"{name} about to yeild: {++i}", TRACE_CATEGORY);
yield return $"{i} from {name}";
// receiving
if (!await coroutine.MoveNextAsync())
{
yield break;
}
Trace.WriteLine($"{name} received: {coroutine.Current}", TRACE_CATEGORY);
// yielding 3
Trace.WriteLine($"{name} about to yeild: {++i}", TRACE_CATEGORY);
yield return $"{i} from {name}";
}
}
/// <summary>
/// CoroutineB yields to CoroutineA
/// </summary>
private async IAsyncEnumerable<string> CoroutineB(
ICoroutineProxy<string> coroutineProxy,
[EnumeratorCancellation] CancellationToken token)
{
await using (var coroutine = await coroutineProxy.AsAsyncEnumerator(token))
{
const string name = "B";
var i = 0;
// receiving
if (!await coroutine.MoveNextAsync())
{
yield break;
}
Trace.WriteLine($"{name} received: {coroutine.Current}", TRACE_CATEGORY);
// yielding 1
Trace.WriteLine($"{name} about to yeild: {++i}", TRACE_CATEGORY);
yield return $"{i} from {name}";
// receiving
if (!await coroutine.MoveNextAsync())
{
yield break;
}
Trace.WriteLine($"{name} received: {coroutine.Current}", TRACE_CATEGORY);
// yielding 2
Trace.WriteLine($"{name} about to yeild: {++i}", TRACE_CATEGORY);
yield return $"{i} from {name}";
// receiving
if (!await coroutine.MoveNextAsync())
{
yield break;
}
Trace.WriteLine($"{name} received: {coroutine.Current}", TRACE_CATEGORY);
}
}
/// <summary>
/// Testing CoroutineA and CoroutineB cooperative execution
/// </summary>
[TestMethod]
public async Task Test_Coroutine_Execution_Flow()
{
// Here we execute two cotoutines, CoroutineA and CoroutineB,
// which asynchronously yield to each other
//TODO: test cancellation scenarios
var token = CancellationToken.None;
using (var apartment = new Tests.ThreadPoolApartment())
{
await apartment.Run(async () =>
{
var proxyA = new CoroutineProxy<string>();
var proxyB = new CoroutineProxy<string>();
var listener = new Tests.CategoryTraceListener(TRACE_CATEGORY);
Trace.Listeners.Add(listener);
try
{
// start both coroutines
await Task.WhenAll(
proxyA.Run(token => CoroutineA(proxyB, token), token),
proxyB.Run(token => CoroutineB(proxyA, token), token))
.WithAggregatedExceptions();
}
finally
{
Trace.Listeners.Remove(listener);
}
var traces = listener.ToArray();
Assert.AreEqual(traces[0], "A about to yeild: 1");
Assert.AreEqual(traces[1], "B received: 1 from A");
Assert.AreEqual(traces[2], "B about to yeild: 1");
Assert.AreEqual(traces[3], "A received: 1 from B");
Assert.AreEqual(traces[4], "A about to yeild: 2");
Assert.AreEqual(traces[5], "B received: 2 from A");
Assert.AreEqual(traces[6], "B about to yeild: 2");
Assert.AreEqual(traces[7], "A received: 2 from B");
Assert.AreEqual(traces[8], "A about to yeild: 3");
Assert.AreEqual(traces[9], "B received: 3 from A");
});
}
}
}
}
The test's output looks like this:
coroutines: A about to yeild: 1 coroutines: B received: 1 from A coroutines: B about to yeild: 1 coroutines: A received: 1 from B coroutines: A about to yeild: 2 coroutines: B received: 2 from A coroutines: B about to yeild: 2 coroutines: A received: 2 from B coroutines: A about to yeild: 3 coroutines: B received: 3 from A
I currently use asynchronous co-routines in some of my automated UI testing scenarios. E.g., I might have an asynchronous test workflow logic that runs on a UI thread (that'd be CouroutineA) and a complimentary workflow that runs on a ThreadPool thread as a part of a [TestMethod] method (that'd be CouroutineB).
Then I could do something like await WaitForUserInputAsync(); yield return true; to synchronize at certain key points of CouroutineA and CouroutineB cooperative execution flow.
Without yield return I'd have to use some form of asynchronous synchronization primitives, like Stephen Toub's AsyncManualResetEvent. I personally feel using co-routines is a more natural way of doing such kind of synchronization.
The code for CoroutineProxy (which drives the execution of co-routines) is still a work-in-progress. It currently uses TPL Dataflow's BufferBlock as a proxy queue to coordinate the asynchronous execution, and I am not sure yet if it is an optimal way of doing that. Currently, this is what it looks like this:
using System;
using System.Collections.Generic;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
#nullable enable
namespace Tests
{
public interface ICoroutineProxy<T>
{
public Task<IAsyncEnumerable<T>> AsAsyncEnumerable(CancellationToken token = default);
}
public static class CoroutineProxyExt
{
public async static Task<IAsyncEnumerator<T>> AsAsyncEnumerator<T>(
this ICoroutineProxy<T> @this,
CancellationToken token = default)
{
return (await @this.AsAsyncEnumerable(token)).GetAsyncEnumerator(token);
}
}
public class CoroutineProxy<T> : ICoroutineProxy<T>
{
readonly TaskCompletionSource<IAsyncEnumerable<T>> _proxyTcs =
new TaskCompletionSource<IAsyncEnumerable<T>>(TaskCreationOptions.RunContinuationsAsynchronously);
public CoroutineProxy()
{
}
private async IAsyncEnumerable<T> CreateProxyAsyncEnumerable(
ISourceBlock<T> bufferBlock,
[EnumeratorCancellation] CancellationToken token)
{
var completionTask = bufferBlock.Completion;
while (true)
{
var itemTask = bufferBlock.ReceiveAsync(token);
var any = await Task.WhenAny(itemTask, completionTask);
if (any == completionTask)
{
// observe completion exceptions if any
await completionTask;
yield break;
}
yield return await itemTask;
}
}
async Task<IAsyncEnumerable<T>> ICoroutineProxy<T>.AsAsyncEnumerable(CancellationToken token)
{
using (token.Register(() => _proxyTcs.TrySetCanceled(), useSynchronizationContext: true))
{
return await _proxyTcs.Task;
}
}
public async Task Run(Func<CancellationToken, IAsyncEnumerable<T>> routine, CancellationToken token)
{
token.ThrowIfCancellationRequested();
var bufferBlock = new BufferBlock<T>();
var proxy = CreateProxyAsyncEnumerable(bufferBlock, token);
_proxyTcs.SetResult(proxy); // throw if already set
try
{
//TODO: do we need to use routine(token).WithCancellation(token) ?
await foreach (var item in routine(token))
{
await bufferBlock.SendAsync(item, token);
}
bufferBlock.Complete();
}
catch (Exception ex)
{
((IDataflowBlock)bufferBlock).Fault(ex);
throw;
}
}
}
}