TaskContinuationOptions.RunContinuationsAsynchronously and Stack Dives

Question

In this blog post, Stephan Toub describes a new feature that will be included in .NET 4.6 which adds another value to the TaskCreationOptions and TaskContinuationOptions enums called RunContinuationsAsynchronously.

He explains:

"I talked about a ramification of calling {Try}Set* methods on TaskCompletionSource, that any synchronous continuations off of the TaskCompletionSource’s Task could run synchronously as part of the call. If we were to invoke SetResult here while holding the lock, then synchronous continuations off of that Task would be run while holding the lock, and that could lead to very real problems. So, while holding the lock we grab the TaskCompletionSource to be completed, but we don’t complete it yet, delaying doing so until the lock has been released"

And gives the following example to demonstrate:

private SemaphoreSlim _gate = new SemaphoreSlim(1, 1);
private async Task WorkAsync()
{
    await _gate.WaitAsync().ConfigureAwait(false);
    try
    {
        // work here
    }
    finally { _gate.Release(); }
}

Now imagine that you have lots of calls to WorkAsync:

await Task.WhenAll(from i in Enumerable.Range(0, 10000) select WorkAsync());

We've just created 10,000 calls to WorkAsync that will be appropriately serialized on the semaphore. One of the tasks will enter the critical region, and the others will queue up on the WaitAsync call, inside SemaphoreSlim effectively enqueueing the task to be completed when someone calls Release. If Release completed that Task synchronously, then when the first task calls Release, it'll synchronously start executing the second task, and when it calls Release, it'll synchronously start executing the third task, and so on. If the "//work here" section of code above didn't include any awaits that yielded, then we're potentially going to stack dive here and eventually potentially blow out the stack.

I'm having a hard time grasping the part where he talks about executing the continuation synchronously.

Question

How could this possibly cause a stack dive? More so, And what is RunContinuationsAsynchronously effectively going to do in order to solve that problem?

Experimentalize

Yuval, you may want to check How can I prevent synchronous continuations on a Task?, it was a big hit. – Lymphatic

@Noseratio Good question, albeit different than what i wantes to know. – Experimentalize

Yuval, I actually have some additional thoughts, like I dealt with a case where this stack dive was occurring due two objects communicating with each other via await TCS.Task/TCS.SetResult. I'll put it together as time allows :) – Lymphatic

Put the baby to sleep and post it! Waiting.. – Experimentalize

@Noseratio 10 days? C'mon ;P – Experimentalize

These 10 days feel like a single long day :) Was this any useful? – Lymphatic

Answer 1

The key concept here is that a task's continuation may run synchronously on the same thread that completed the antecedent task.

Let's imagine that this is SemaphoreSlim.Release's implementation (it's actually Toub's AsyncSemphore's):

public void Release() 
{ 
    TaskCompletionSource<bool> toRelease = null; 
    lock (m_waiters) 
    { 
        if (m_waiters.Count > 0) 
            toRelease = m_waiters.Dequeue(); 
        else 
            ++m_currentCount; 
    } 
    if (toRelease != null) 
        toRelease.SetResult(true); 
}

We can see that it synchronously completes a task (using TaskCompletionSource). In this case, if WorkAsync has no other asynchronous points (i.e. no awaits at all, or all awaits are on an already completed task) and calling _gate.Release() may complete a pending call to _gate.WaitAsync() synchronously on the same thread you may reach a state in which a single thread sequentially releases the semaphore, completes the next pending call, executes // work here and then releases the semaphore again etc. etc.

This means that the same thread goes deeper and deeper in the stack, hence stack dive.

RunContinuationsAsynchronously makes sure the continuation doesn't run synchronously and so the thread that releases the semaphore moves on and the continuation is scheduled for another thread (which one depends on the other continuation parameters e.g. TaskScheduler)

This logically resembles posting the completion to the ThreadPool:

public void Release() 
{ 
    TaskCompletionSource<bool> toRelease = null; 
    lock (m_waiters) 
    { 
        if (m_waiters.Count > 0) 
            toRelease = m_waiters.Dequeue(); 
        else 
            ++m_currentCount; 
    } 
    if (toRelease != null) 
        Task.Run(() => toRelease.SetResult(true)); 
}

Inoculation

So the point is one thread having to "release" multiple concurrent calls, causing him slowly increase the same threads stack? – Experimentalize

@YuvalItzchakov The calls aren't really concurrent since the semaphore limit is 1. It's more of a recursion where by releasing the semaphore the thread enters the semaphore and releases it which enters it again and so forth and so forth. It's equivalent to void Do(int count) { if count > 0 Do(--count)} – Inoculation

@YuvalItzchakov It will stop at some point, so it's not an infinite loop, but it still may reach a StackOverflowException and inside a lock it can cause a deadlock – Inoculation

I remember when TPL was simple. This feels like a problem that the default scheduler should fix without developers tying their minds in knots to understand another edge case. – Condon

@Condon The TPL does. SemaphoreSlim doesn't actually act in this way, it posts all continuations to the ThreadPool. This option lets you build TPL equivalent constructs yourself. You don't need that in everyday code. – Inoculation

@Condon It seems like this option is provided for special use-cases. This definitely doesn't seem like a common case. – Experimentalize

@YuvalItzchakov at some point every developer now needs to know a) when a use case is special and b) how to kludge around it. – Condon

@Condon Not necessarily. I don't see many developers implementing synchronization primitives and async wrappers are synchronous code. – Experimentalize

@YuvalItzchakov so this whole issue is limited to a semaphore implementation? I can't believe that. This async stuff has become the leakiest of abstractions. – Condon

@Condon on the contrary. If you only use async-await you would never encounter that problem. If you use ContinueWith directly though, you may need to handle that (though very very unlikely). – Inoculation

Answer 2

How could this possibly cause a stack dive? More so, And what is RunContinuationsAsynchronously effectively going to do in order to solve that problem?

i3arnon provides a very good explanation of the reasons behind introducing RunContinuationsAsynchronously. My answer is rather orthogonal to his; in fact, I'm writing this for my own reference as well (I myself ain't gonna remember any subtleties of this in half a year from now :)

First of all, let's see how TaskCompletionSource's RunContinuationsAsynchronously option is different from Task.Run(() => tcs.SetResult(result)) or the likes. Let's try a simple console application:

using System;
using System.Threading;
using System.Threading.Tasks;

namespace ConsoleApplications
{
    class Program
    {
        static void Main(string[] args)
        {
            ThreadPool.SetMinThreads(100, 100);

            Console.WriteLine("start, " + new { System.Environment.CurrentManagedThreadId });

            var tcs = new TaskCompletionSource<bool>();

            // test ContinueWith-style continuations (TaskContinuationOptions.ExecuteSynchronously)
            ContinueWith(1, tcs.Task);
            ContinueWith(2, tcs.Task);
            ContinueWith(3, tcs.Task);

            // test await-style continuations
            ContinueAsync(4, tcs.Task);
            ContinueAsync(5, tcs.Task);
            ContinueAsync(6, tcs.Task);

            Task.Run(() =>
            {
                Console.WriteLine("before SetResult, " + new { System.Environment.CurrentManagedThreadId });
                tcs.TrySetResult(true);
                Thread.Sleep(10000);
            });
            Console.ReadLine();
        }

        // log
        static void Continuation(int id)
        {
            Console.WriteLine(new { continuation = id, System.Environment.CurrentManagedThreadId });
            Thread.Sleep(1000);
        }

        // await-style continuation
        static async Task ContinueAsync(int id, Task task)
        {
            await task.ConfigureAwait(false);
            Continuation(id);
        }

        // ContinueWith-style continuation
        static Task ContinueWith(int id, Task task)
        {
            return task.ContinueWith(
                t => Continuation(id),
                CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
        }
    }
}

Note how all continuations run synchronously on the same thread where TrySetResult has been called:

start, { CurrentManagedThreadId = 1 }
before SetResult, { CurrentManagedThreadId = 3 }
{ continuation = 1, CurrentManagedThreadId = 3 }
{ continuation = 2, CurrentManagedThreadId = 3 }
{ continuation = 3, CurrentManagedThreadId = 3 }
{ continuation = 4, CurrentManagedThreadId = 3 }
{ continuation = 5, CurrentManagedThreadId = 3 }
{ continuation = 6, CurrentManagedThreadId = 3 }

Now what if we don't want this to happen, and we want each continuation to run asynchronously (i.e., in parallel with other continuations and possibly on another thread, in the absence of any synchronization context)?

There's a trick that could do it for await-style continuations, by installing a fake temporary synchronization context (more details here):

public static class TaskExt
{
    class SimpleSynchronizationContext : SynchronizationContext
    {
        internal static readonly SimpleSynchronizationContext Instance = new SimpleSynchronizationContext();
    };

    public static void TrySetResult<TResult>(this TaskCompletionSource<TResult> @this, TResult result, bool asyncAwaitContinuations)
    {
        if (!asyncAwaitContinuations)
        {
            @this.TrySetResult(result);
            return;
        }

        var sc = SynchronizationContext.Current;
        SynchronizationContext.SetSynchronizationContext(SimpleSynchronizationContext.Instance);
        try
        {
            @this.TrySetResult(result);
        }
        finally
        {
            SynchronizationContext.SetSynchronizationContext(sc);
        }
    }
}

Now, using tcs.TrySetResult(true, asyncAwaitContinuations: true) in our test code:

start, { CurrentManagedThreadId = 1 }
before SetResult, { CurrentManagedThreadId = 3 }
{ continuation = 1, CurrentManagedThreadId = 3 }
{ continuation = 2, CurrentManagedThreadId = 3 }
{ continuation = 3, CurrentManagedThreadId = 3 }
{ continuation = 4, CurrentManagedThreadId = 4 }
{ continuation = 5, CurrentManagedThreadId = 5 }
{ continuation = 6, CurrentManagedThreadId = 6 }

Note how await continuations now run in parallel (albeit, still after all synchronous ContinueWith continuations).

This asyncAwaitContinuations: true logic is a hack and it works for await continuations only. The new RunContinuationsAsynchronously makes it work consistently for any kind of continuations, attached to TaskCompletionSource.Task.

Another nice aspect of RunContinuationsAsynchronously is that any await-style continuations scheduled to be resumed on specific synchronization context will run on that context asynchronously (using SynchronizationContext.Post, even if TCS.Task completes on the same context (unlike the current behavior of TCS.SetResult). ContinueWith-style continuations will be also be run asynchronously by their corresponding task schedulers (most often, TaskScheduler.Default or TaskScheduler.FromCurrentSynchronizationContext). They won't be inlined via TaskScheduler.TryExecuteTaskInline. I believe Stephen Toub has clarified that in the comments to his blog post, and it also can be seen here in CoreCLR's Task.cs.

Why should we be worrying about imposing asynchrony on all continuations?

I usually need it when I deal with async methods which execute cooperatively (co-routines).

A simple example is a pause-able asynchronous processing: one async process pauses/resumes the execution of another. Their execution workflow synchronizes at certain await points, and TaskCompletionSource is used for such kind of synchronization, directly or indirectly.

Below is some ready-to-play-with sample code which uses an adaptation of Stephen Toub's PauseTokenSource. Here, one async method StartAndControlWorkAsync starts and periodically pauses/resumes another async method, DoWorkAsync. Try changing asyncAwaitContinuations: true to asyncAwaitContinuations: false and see the logic being completely broken:

using System;
using System.Threading;
using System.Threading.Tasks;

namespace ConsoleApp
{
    class Program
    {
        static void Main()
        {
            StartAndControlWorkAsync(CancellationToken.None).Wait();
        }

        // Do some work which can be paused/resumed
        public static async Task DoWorkAsync(PauseToken pause, CancellationToken token)
        {
            try
            {
                var step = 0;
                while (true)
                {
                    token.ThrowIfCancellationRequested();
                    Console.WriteLine("Working, step: " + step++);
                    await Task.Delay(1000).ConfigureAwait(false);
                    Console.WriteLine("Before await pause.WaitForResumeAsync()");
                    await pause.WaitForResumeAsync();
                    Console.WriteLine("After await pause.WaitForResumeAsync()");
                }
            }
            catch (Exception e)
            {
                Console.WriteLine("Exception: {0}", e);
                throw;
            }
        }

        // Start DoWorkAsync and pause/resume it
        static async Task StartAndControlWorkAsync(CancellationToken token)
        {
            var pts = new PauseTokenSource();
            var task = DoWorkAsync(pts.Token, token);

            while (true)
            {
                token.ThrowIfCancellationRequested();

                Console.WriteLine("Press enter to pause...");
                Console.ReadLine();

                Console.WriteLine("Before pause requested");
                await pts.PauseAsync();
                Console.WriteLine("After pause requested, paused: " + pts.IsPaused);

                Console.WriteLine("Press enter to resume...");
                Console.ReadLine();

                Console.WriteLine("Before resume");
                pts.Resume();
                Console.WriteLine("After resume");
            }
        }

        // Based on Stephen Toub's PauseTokenSource
        // http://blogs.msdn.com/b/pfxteam/archive/2013/01/13/cooperatively-pausing-async-methods.aspx
        // the main difference is to make sure that when the consumer-side code - which requested the pause - continues, 
        // the producer-side code has already reached the paused (awaiting) state.
        // E.g. a media player "Pause" button is clicked, gets disabled, playback stops, 
        // and only then "Resume" button gets enabled

        public class PauseTokenSource
        {
            internal static readonly Task s_completedTask = Task.Delay(0);

            readonly object _lock = new Object();

            bool _paused = false;

            TaskCompletionSource<bool> _pauseResponseTcs;
            TaskCompletionSource<bool> _resumeRequestTcs;

            public PauseToken Token { get { return new PauseToken(this); } }

            public bool IsPaused
            {
                get
                {
                    lock (_lock)
                        return _paused;
                }
            }

            // request a resume
            public void Resume()
            {
                TaskCompletionSource<bool> resumeRequestTcs = null;

                lock (_lock)
                {
                    resumeRequestTcs = _resumeRequestTcs;
                    _resumeRequestTcs = null;

                    if (!_paused)
                        return;
                    _paused = false;
                }

                if (resumeRequestTcs != null)
                    resumeRequestTcs.TrySetResult(true, asyncAwaitContinuations: true);
            }

            // request a pause (completes when paused state confirmed)
            public Task PauseAsync()
            {
                Task responseTask = null;

                lock (_lock)
                {
                    if (_paused)
                        return _pauseResponseTcs.Task;
                    _paused = true;

                    _pauseResponseTcs = new TaskCompletionSource<bool>();
                    responseTask = _pauseResponseTcs.Task;

                    _resumeRequestTcs = null;
                }

                return responseTask;
            }

            // wait for resume request
            internal Task WaitForResumeAsync()
            {
                Task resumeTask = s_completedTask;
                TaskCompletionSource<bool> pauseResponseTcs = null;

                lock (_lock)
                {
                    if (!_paused)
                        return s_completedTask;

                    _resumeRequestTcs = new TaskCompletionSource<bool>();
                    resumeTask = _resumeRequestTcs.Task;

                    pauseResponseTcs = _pauseResponseTcs;

                    _pauseResponseTcs = null;
                }

                if (pauseResponseTcs != null)
                    pauseResponseTcs.TrySetResult(true, asyncAwaitContinuations: true);

                return resumeTask;
            }
        }

        // consumer side
        public struct PauseToken
        {
            readonly PauseTokenSource _source;

            public PauseToken(PauseTokenSource source) { _source = source; }

            public bool IsPaused { get { return _source != null && _source.IsPaused; } }

            public Task WaitForResumeAsync()
            {
                return IsPaused ?
                    _source.WaitForResumeAsync() :
                    PauseTokenSource.s_completedTask;
            }
        }


    }

    public static class TaskExt
    {
        class SimpleSynchronizationContext : SynchronizationContext
        {
            internal static readonly SimpleSynchronizationContext Instance = new SimpleSynchronizationContext();
        };

        public static void TrySetResult<TResult>(this TaskCompletionSource<TResult> @this, TResult result, bool asyncAwaitContinuations)
        {
            if (!asyncAwaitContinuations)
            {
                @this.TrySetResult(result);
                return;
            }

            var sc = SynchronizationContext.Current;
            SynchronizationContext.SetSynchronizationContext(SimpleSynchronizationContext.Instance);
            try
            {
                @this.TrySetResult(result);
            }
            finally
            {
                SynchronizationContext.SetSynchronizationContext(sc);
            }
        }
    }
}

I didn't want to use Task.Run(() => tcs.SetResult(result)) here, because it would be redundant to push continuations to ThreadPool when they're already scheduled to run asynchronously on a UI thread with a proper synchronization context. At the same time, if both StartAndControlWorkAsync and DoWorkAsync run on the same UI synchronization context, we'd also have a stack dive (if tcs.SetResult(result) is used without Task.Run or SynchronizationContext.Post wrapping).

Now, RunContinuationsAsynchronously is probably the best solution to this problem.

Lymphatic

Splendid! I'm gonna try and execute some of that code tomorrow, see if i have any question. Thanks! – Experimentalize