This post is part of the series 'Async/await in C#'. Be sure to check out the rest of the blog posts of the series!
The previous post decompiled an async method and showed the state machine the compiler generates. That state machine calls GetAwaiter, reads IsCompleted, registers a continuation, and eventually completes a Task. Every one of those calls lands on code that somebody wrote.
This post writes it. MyThreadPool and MyTask, from nothing, in about 150 lines that actually run. Not a simplification of the real types for the sake of a diagram: a working implementation that the C# compiler will happily drive with a real await.
#A thread pool in twenty lines
Task needs somewhere to run continuations. A thread pool is a queue plus a fixed number of threads draining it, and that really is all it is:
C#
public static class MyThreadPool
{
private static readonly BlockingCollection<(Action Work, ExecutionContext? Context)> WorkItems = new();
static MyThreadPool()
{
for (var i = 0; i < Environment.ProcessorCount; i++)
{
var thread = new Thread(WorkerLoop) { IsBackground = true };
thread.Start();
}
}
public static void QueueUserWorkItem(Action work)
=> WorkItems.Add((work, ExecutionContext.Capture()));
private static void WorkerLoop()
{
foreach (var (work, context) in WorkItems.GetConsumingEnumerable())
{
if (context is null)
{
work();
}
else
{
ExecutionContext.Run(context, static state => ((Action)state!)(), work);
}
}
}
}
The queue is the obvious part. The ExecutionContext is the part worth noticing: the thread that runs the work is not the thread that queued it, so anything ambient on the queueing thread would be lost. Capturing at queue time and restoring at run time is what makes AsyncLocal<T> survive the trip. A later post covers ExecutionContext properly; here it is two lines, and skipping them would break the sample's first demo.
IsBackground = true means these threads do not keep the process alive. That is the real thread pool's behaviour too.
#State, result, continuation
MyTask holds exactly what the post on the Task lifecycle said a task holds:
C#
public sealed class MyTask
{
private readonly Lock _lock = new();
private bool _completed;
private Exception? _exception;
private Action? _continuation;
private ExecutionContext? _continuationContext;
}
One continuation, not a list. The real Task supports many; one is enough to make await work, and it keeps the code readable.
Completing the task is a matter of setting the fields and running whatever was waiting:
C#
public void SetResult() => Complete(exception: null);
public void SetException(Exception exception) => Complete(exception);
private void Complete(Exception? exception)
{
Action? continuation;
ExecutionContext? context;
lock (_lock)
{
if (_completed)
throw new InvalidOperationException("The task is already completed");
_completed = true;
_exception = exception;
continuation = _continuation;
context = _continuationContext;
}
// Scheduling happens outside the lock: the continuation must never run while the lock is held.
if (continuation is not null)
Schedule(continuation, context);
}
Those two words in the comment are the whole reason this type has a lock at all. Consider the race:
- Thread A calls
OnCompleted to register a continuation. - Thread B completes the task.
If B wins, there is no continuation to run yet, and A must run it itself. If A wins, B finds one and runs it. What must never happen is both deciding the other will do it, or both doing it. So registration checks the completed flag under the same lock:
C#
private void AddContinuation(Action continuation, ExecutionContext? context)
{
var alreadyCompleted = false;
lock (_lock)
{
if (_completed)
{
alreadyCompleted = true;
}
else
{
_continuation = continuation;
_continuationContext = context;
}
}
// Registering on an already completed task must still run the continuation.
if (alreadyCompleted)
Schedule(continuation, context);
}
Schedule is where the continuation's own captured context wins over the completing thread's:
C#
private static void Schedule(Action continuation, ExecutionContext? context)
{
if (context is null)
{
MyThreadPool.QueueUserWorkItem(continuation);
}
else
{
MyThreadPool.QueueUserWorkItem(
() => ExecutionContext.Run(context, static state => ((Action)state!)(), continuation));
}
}
Note that continuations always go through the pool. They never run inline on the thread that called SetResult. That is the equivalent of TaskCreationOptions.RunContinuationsAsynchronously, and making it the only behaviour removes a whole class of surprise from this implementation. The real Task defaults to the other choice, and that default is why RunContinuationsAsynchronously exists.
#Blocking, once
Somebody has to be able to wait synchronously, if only at the top of Main:
C#
public void Wait()
{
ManualResetEventSlim? completed = null;
lock (_lock)
{
if (!_completed)
{
completed = new ManualResetEventSlim();
_continuation = completed.Set;
_continuationContext = null;
}
}
completed?.Wait();
if (_exception is not null)
ExceptionDispatchInfo.Throw(_exception);
}
This is the one method in the file that burns a thread, and the rest of the series is about not calling it. It is also where the exception comes back out, via ExceptionDispatchInfo.Throw so the original stack trace survives the rethrow.
#Making it awaitable
Everything needed for await is already there. The awaitable pattern asks for four members, and MyTask can supply all of them from its private state:
C#
public Awaiter GetAwaiter() => new(this);
public readonly struct Awaiter(MyTask task) : INotifyCompletion
{
public bool IsCompleted => task.IsCompleted;
public void OnCompleted(Action continuation)
=> task.AddContinuation(continuation, ExecutionContext.Capture());
public void GetResult() => task.Wait();
}
That is it. await someMyTask now compiles and works:
C#
await MyTask.Delay(TimeSpan.FromMilliseconds(200));
GetResult calling Wait looks alarming but never blocks: the compiler only calls it after IsCompleted was true or the continuation fired, so the flag is already set and Wait falls straight through to the exception check.
The awaiter is a nested type so it can touch MyTask's private members, and a readonly struct so awaiting costs no allocation.
What MyTask still cannot do is be returned from an async method. async MyTask Foo() does not compile, because the compiler needs a builder to construct and complete the task on the method's behalf. That is [AsyncMethodBuilder], and it is the next post's subject.
#Run, Delay and WhenAll
With a pool, a completion and a continuation, the familiar static helpers are short.
Run is queue plus complete:
C#
public static MyTask Run(Action action)
{
var task = new MyTask();
MyThreadPool.QueueUserWorkItem(() =>
{
try
{
action();
}
catch (Exception ex)
{
task.SetException(ex);
return;
}
task.SetResult();
});
return task;
}
Delay is the interesting one, because it is the smallest possible demonstration of "there is no thread":
C#
public static MyTask Delay(TimeSpan delay)
{
var task = new MyTask();
Timer? timer = null;
timer = new Timer(_ =>
{
timer!.Dispose();
task.SetResult();
});
timer.Change(delay, Timeout.InfiniteTimeSpan);
return task;
}
Nothing sleeps. A timer is registered, the method returns immediately, and some time later a callback completes the task. Between those two moments there is no thread associated with the delay at all. That is the same shape as a real I/O completion, which an earlier post described from the operating system's side.
The timer is assigned to a local that the callback closes over, which keeps it rooted until it fires. Forget that and the Timer becomes eligible for collection before the delay elapses.
WhenAll is a counter and a list of exceptions:
C#
public static MyTask WhenAll(IReadOnlyList<MyTask> tasks)
{
var all = new MyTask();
if (tasks.Count is 0)
{
all.SetResult();
return all;
}
var remaining = tasks.Count;
var exceptions = new List<Exception>();
foreach (var task in tasks)
{
task.ContinueWith(() =>
{
if (task._exception is not null)
{
lock (exceptions)
{
exceptions.Add(task._exception);
}
}
if (Interlocked.Decrement(ref remaining) is 0)
{
if (exceptions.Count is 0)
{
all.SetResult();
}
else
{
all.SetException(new AggregateException(exceptions));
}
}
});
}
return all;
}
The Interlocked.Decrement is not decoration. Several tasks can complete on several pool threads at the same instant, and exactly one of them has to be the one that completes all. Reaching zero is how that one is chosen. Collecting every exception rather than the first is also deliberate, and it is why the real Task.WhenAll hands you an AggregateException.
#Iterate: async before async
Here is the piece that explains why async/await was added to the language at all.
Before the compiler could rewrite a method, you composed asynchronous steps by hand. ContinueWith works, but chaining it gets unreadable fast, and loops are close to impossible. The trick people used instead was an iterator: yield return a task, and let a driver advance the iterator each time one completes.
C#
public static MyTask Iterate(IEnumerable<MyTask> tasks)
{
var result = new MyTask();
var enumerator = tasks.GetEnumerator();
MoveNext();
return result;
void MoveNext()
{
try
{
if (enumerator.MoveNext())
{
var current = enumerator.Current;
current.ContinueWith(() =>
{
if (current._exception is not null)
{
result.SetException(current._exception);
}
else
{
MoveNext();
}
});
return;
}
}
catch (Exception ex)
{
result.SetException(ex);
return;
}
result.SetResult();
}
}
It lets you write this:
C#
static IEnumerable<MyTask> CountdownAsync()
{
for (var i = 3; i > 0; i--)
{
Console.WriteLine($"{i}...");
yield return MyTask.Delay(TimeSpan.FromMilliseconds(300));
}
Console.WriteLine("Liftoff");
}
await MyTask.Iterate(CountdownAsync());
Read that method and then read the MoveNext above it. The iterator is a state machine, generated by the compiler, whose states are the points between yield returns. Iterate is the driver that resumes it when the awaited thing finishes. That is precisely the pair of things the previous post found in a decompiled async method, with yield return spelled await and the driver spelled AwaitUnsafeOnCompleted.
async/await is this pattern, promoted into the language and made to work with try/catch, using, and a return value.
#Running it
The sample exercises all of it:
inside MyTask.Run: correlationId = request-42
after await: correlationId = request-42
before the delay: thread 16
after the delay: thread 8
Hello
World!
worker 0 done on thread 9
worker 3 done on thread 4
worker 2 done on thread 12
worker 1 done on thread 13
3...
2...
1...
Liftoff
caught at the await: boom
The first two lines are AsyncLocal<T> surviving a hop through MyThreadPool, which is the ExecutionContext.Capture and ExecutionContext.Run pair earning their place. The thread numbers around the delay change, because the continuation was scheduled rather than run inline. The last line is an exception thrown on a pool thread, stored on the task, and rethrown at the await.
#What the real Task adds
MyTask is around 150 lines. Task is several thousand. The difference is not conceptual, it is everything that a general-purpose type has to handle:
- Many continuations, stored in a field that starts as a single object and grows into a list only when needed, to avoid allocating a list for the common case of one.
- Inline completion by default, with
RunContinuationsAsynchronously to opt out. MyTask always schedules, which is safer and slower. TaskScheduler, so a continuation can be sent somewhere other than the thread pool. That is the hook SynchronizationContext and ConfigureAwait use.ICriticalNotifyCompletion, so the state machine can skip the redundant ExecutionContext capture that MyTask.Awaiter.OnCompleted performs on every await.- A result, which means
Task<TResult> and a generic awaiter. - Cancellation, a third terminal state that is neither success nor failure. That is the next post.
- Exception bookkeeping:
AggregateException, UnobservedTaskException, and the machinery that decides whether an exception was ever looked at. - Struct-based completion sources, so that a hot operation that usually finishes synchronously does not allocate at all. That is
ValueTask and IValueTaskSource.
None of those change the model. They make it fast and general.
#What to take away
- A thread pool is a queue and some threads. The interesting part is capturing and restoring
ExecutionContext around each work item. - A task is three fields: completed, exception, continuation. Everything else is built on them.
- The only genuinely hard part is the race between registering a continuation and completing the task. Decide under a lock, run outside it.
- A continuation must never be invoked while holding the lock.
GetAwaiter on a type with those three fields is enough to make await work. Returning it from an async method needs a builder, which is the next topic.Delay with a timer shows "there is no thread" in six lines: nothing waits between the call and the callback.Iterate is what asynchronous code looked like before async/await, and the compiler's rewrite is the same idea generated for you.
The next post adds the one thing MyTask cannot do: stop. Cancellation, built from scratch as well.
#Additional resources
Do you have a question or a suggestion about this post? Contact me!
