{"id":1315,"date":"2018-10-01T13:16:54","date_gmt":"2018-10-01T05:16:54","guid":{"rendered":"https:\/\/blogs.msdn.microsoft.com\/seteplia\/?p=1315"},"modified":"2019-06-11T21:28:17","modified_gmt":"2019-06-12T04:28:17","slug":"the-danger-of-taskcompletionsourcet-class","status":"publish","type":"post","link":"https:\/\/devblogs.microsoft.com\/premier-developer\/the-danger-of-taskcompletionsourcet-class\/","title":{"rendered":"The danger of TaskCompletionSource class"},"content":{"rendered":"

… when used with async\/await.<\/p>\n

TaskCompletionSource<T><\/code> class is a very useful facility if you want to control the lifetime of a task manually. Here is a canonical example when TaskCompletionSource<\/code>is used for converting the event-based asynchronous code to the Task-based pattern:<\/p>\n

public\u00a0static\u00a0Task PerformOperation(this\u00a0PictureBox pictureBox)\r\n{\r\n\u00a0\u00a0\u00a0 var tcs =\u00a0new\u00a0TaskCompletionSource<object>();\r\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 \r\n\u00a0\u00a0\u00a0 \/\/ Naive version that does not unsubscribe from the event\r\n\u00a0\u00a0\u00a0 pictureBox.LoadCompleted += (s, ea) =>\r\n\u00a0\u00a0\u00a0 {\r\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 if (ea.Cancelled) tcs.SetCanceled();\r\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 else\u00a0if (ea.Error !=\u00a0null) tcs.SetException(ea.Error);\r\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 else tcs.SetResult(null);\r\n\u00a0\u00a0\u00a0 };\r\n \r\n\u00a0\u00a0\u00a0 pictureBox.LoadAsync();\r\n \r\n\u00a0\u00a0\u00a0 return tcs.Task;\r\n}<\/pre>\n
Effectively, TaskCompletionSource<T><\/code> represents a future result and gives an ability to set the final state of the underlying task manually by calling SetCanceled<\/code>, SetException<\/code> or SetResult<\/code> methods.<\/p>\n
This class is very useful not only when you need to make an old code to look modern and fancy. TaskCompletionSource<T><\/code> is used in a variety of cases when an operation’s lifetime is controlled manually, for instance, in different communication protocols. So, let’s mimic one of them.<\/p>\n
Let suppose we want to create a custom database adapter. The adapter will have a dedicated “worker” thread for processing requests and ExecuteAsync<\/code> method that a client can use to schedule work for background processing. This is quite similar to what an actual Redis client<\/a> does and some other database clients follow the same pattern, so this is not a far-fetched scenario.<\/p>\n
public class DatabaseFacade : IDisposable\r\n{\r\n    private readonly BlockingCollection<(string item, TaskCompletionSource<string> result)> _queue =\r\n        new BlockingCollection<(string item, TaskCompletionSource<string> result)>();\r\n    private readonly Task _processItemsTask;\r\n \r\n    public DatabaseFacade() => _processItemsTask = Task.Run(ProcessItems);\r\n \r\n    public void Dispose() => _queue.CompleteAdding();\r\n \r\n    public Task SaveAsync(string command)\r\n    {\r\n        var tcs = new TaskCompletionSource<string>();\r\n        _queue.Add((item: command, result: tcs));\r\n        return tcs.Task;\r\n    }\r\n \r\n    private async Task ProcessItems()\r\n    {\r\n        foreach (var item in _queue.GetConsumingEnumerable())\r\n        {\r\n            Console.WriteLine($\"DatabaseFacade: executing '{item.item}'...\");\r\n \r\n            \/\/ Waiting a bit to emulate some IO-bound operation\r\n            await Task.Delay(100);\r\n            item.result.SetResult(\"OK\");\r\n            Console.WriteLine(\"DatabaseFacade: done.\");\r\n        }\r\n    }\r\n}<\/pre>\n
The code is not quite production ready, but its a good example of a producer-consumer pattern based on BlockingCollection<\/code>.<\/p>\n
Suppose we have another component, let’s say a logger. A logger is usually implemented using the producer-consumer pattern as well. For performance reasons, we don’t want to flush the messages on each method call, and instead we can use a blocking collection and a dedicated thread for saving data to external sources. And one of the external sources could be a database.<\/p>\n
public class Logger : IDisposable\r\n{\r\n    private readonly DatabaseFacade _facade;\r\n    private readonly BlockingCollection<string> _queue =\r\n        new BlockingCollection<string>();\r\n \r\n    private readonly Task _saveMessageTask;\r\n \r\n    public Logger(DatabaseFacade facade) =>\r\n        (_facade, _saveMessageTask) = (facade, Task.Run(SaveMessage));\r\n \r\n    public void Dispose() => _queue.CompleteAdding();\r\n \r\n    public void WriteLine(string message) => _queue.Add(message);\r\n \r\n    private async Task SaveMessage()\r\n    {\r\n        foreach (var message in _queue.GetConsumingEnumerable())\r\n        {\r\n            \/\/ \"Saving\" message to the file\r\n            Console.WriteLine($\"Logger: {message}\");\r\n \r\n            \/\/ And to our database through the facade\r\n            await _facade.SaveAsync(message);\r\n        }\r\n    }\r\n}<\/pre>\n
The logger’s implementation is extremely naive and by all means, you should not write yet another logger yourself. My goal here is to show how two producer-consumer queues may affect each other and the logger is quite a widely spread concept that is easy to understand.<\/p>\n
The question is: can you see the issue here? Like a really serious issue!<\/p>\n
Let’s try to run the following code:<\/p>\n
using (var facade = new DatabaseFacade())\r\nusing (var logger = new Logger(facade))\r\n{\r\n    logger.WriteLine(\"My message\");\r\n    await Task.Delay(100);\r\n \r\n    await facade.SaveAsync(\"Another string\");\r\n    Console.WriteLine(\"The string is saved\");\r\n}<\/pre>\n
The output is:<\/p>\n
Logger: My message\r\nDatabaseFacade: executing 'My message'...<\/pre>\n
We never save “Another string” into the database. Why? Because the database facade’s thread is blocked by the logger’s thread.<\/p>\n
TaskCompletionSource<\/code> type has a very peculiar behavior: by default, when SetResult<\/code> method is called then all the task’s “async” continuations are invoked … synchronously. That’s what happening in our case (the same is true for SetCancelled<\/code>, SetException<\/code>, as well as their TrySetXXX<\/code> counterparts):<\/p>\n
\"The<\/a><\/code><\/p>\n
 <\/p>\n
It means that the two “queues” are implicitly linked together and the logger’s queue blocks the adapter’s queue.<\/p>\n
Unfortunately, the situation like this is relatively common and I’ve faced it several times in my projects. The issue may occur when “a continuation” (*) of a task, backed by TaskCompletionSource<T><\/code>, blocks the thread in one way or another, thereby blocking a thread that calls SetResult<\/code>.<\/p>\n
(*) As we’ll see in a moment different types of continuations behave differently.<\/p>\n
The main challenge with such issues that it’s very hard to understand the root cause. Once you have a dump of a process from a production machine you may see no obvious issues at all. You could have a bunch of threads waiting on kernel objects without any relevant user’s code in any of the stack traces.<\/p>\n
Now let’s see why this is happening and how can we mitigate the issue.<\/p>\n
Each task has a state field called m_stateFlags<\/code> field<\/a> that represents the current state of a task (like RanToCompletion<\/code>, Cancelled<\/code>, Failed<\/code> etc). But this is not the only role of the field: it also contains a set of flags specified during task creation via TaskCreationOptions<\/code><\/a>. These flags control different aspects, like whether to run the task in a dedicated thread ([TaskCreationOptions.LongRunning<\/code><\/a>), to schedule work item into a global queue instead of a thread-local one (TaskCreationOptions.PreferFairness<\/code><\/a>), or whether to force task continuations always run asynchronously (TaskCreationOptions.RunContinuationsAsynchronously<\/a>).<\/p>\n
Obviously, we’re interested in the latter aspect and we’ll see at the moment how you can specify this flag. But to fully understand the issue we need to look at the other aspect as well: we need to understand task continuations.<\/p>\n
static async Task WithAsync()\r\n{\r\n    var task = Task.Run(() => { Sleep(100); });\r\n    await task;\r\n    Console.WriteLine(\"After task await\");\r\n}\r\n \r\nstatic Task WithContinueWith()\r\n{\r\n    var task = Task.Run(() => { Sleep(100); });\r\n    return task.ContinueWith(\r\n        t => { Console.WriteLine(\"Inside ContinueWith\"); },\r\n        TaskContinuationOptions.OnlyOnRanToCompletion);\r\n}<\/pre>\n
You may think that the two implementations are equivalent because the compiler just “moves” the block of code between await<\/code> statements into a continuation, scheduled via ContinueWith<\/code>. But this is only kind-of true and the actual logic is a bit more involved.<\/p>\n
An actual transformation that the C# compiler does for async methods is described in more details in my other post “Dissecting async methods in C#”<\/a> and here we’ll focus on one particular aspect: continuations scheduling.<\/p>\n
When an awaited task is not finished, the generated state machine calls TaskAwaiter.UnsafeOnCompleted<\/code><\/a> and passes a call back that is called when the awaited task is done to move the state machine forward. This method calls Task.SetContinuationForAwait<\/code><\/a> to add a given action as a task’s continuation:<\/p>\n
\/\/ Now register the continuation, and if we couldn't register it because the task is already completing,\r\n\/\/ process the continuation directly (in which case make sure we schedule the continuation\r\n\/\/ rather than inlining it, the latter of which could result in a rare but possible stack overflow).\r\nif (tc != null)\r\n{\r\n    if (!AddTaskContinuation(tc, addBeforeOthers: false))\r\n        tc.Run(this, bCanInlineContinuationTask: false);\r\n}\r\nelse\r\n{\r\n    Contract.Assert(!flowExecutionContext, \"We already determined we're not required to flow context.\");\r\n    if (!AddTaskContinuation(continuationAction, addBeforeOthers: false))\r\n        AwaitTaskContinuation.UnsafeScheduleAction(continuationAction, this);\r\n}<\/pre>\n
Local variable tc<\/code> is not null if synchronization context is involved, otherwise, the else<\/code>block is called. First, AddTaskContinuation<\/code><\/a> method is called that returns true<\/code> when a current task is not finished (to prevent stack overflow) and a given action is successfully added as a continuation for the current task. Otherwise UnsafeScheduleAction<\/code> is called that creates AwaitTaskContinuation<\/code> instance.<\/p>\n
In a common case (more details later) a System.Action<\/code> instance is added as a task continuation and the continuation is stored in Task.m_continuationObject<\/code><\/a>.<\/p>\n
Now, let’s see what is happening when a task is finished (code snippet from Task.FinishContinuations<\/code><\/a>):<\/p>\n
internal void FinishContinuations()\r\n{\r\n    \/\/ Atomically store the fact that this task is completing.  From this point on, the adding of continuations will\r\n    \/\/ result in the continuations being run\/launched directly rather than being added to the continuation list.\r\n    object continuationObject = Interlocked.Exchange(ref m_continuationObject, s_taskCompletionSentinel);\r\n \r\n    \/\/ If continuationObject == null, then we don't have any continuations to process\r\n    if (continuationObject != null)\r\n    {\r\n        \/\/ Skip synchronous execution of continuations if this task's thread was aborted\r\n        bool bCanInlineContinuations = !(((m_stateFlags & TASK_STATE_THREAD_WAS_ABORTED) != 0) ||\r\n                                            (Thread.CurrentThread.ThreadState == ThreadState.AbortRequested) ||\r\n                                            ((m_stateFlags & (int)TaskCreationOptions.RunContinuationsAsynchronously) != 0));\r\n \r\n        \/\/ Handle the single-Action case\r\n        Action singleAction = continuationObject as Action;\r\n        if (singleAction != null)\r\n        {\r\n            AwaitTaskContinuation.RunOrScheduleAction(singleAction, bCanInlineContinuations, ref t_currentTask);\r\n            return;\r\n        }\r\n        \/\/ The rest of the body\r\n    }\r\n}<\/pre>\n
FinishContinuations<\/code> method checks the task creation flags and if RunContinuationsAsynchronously<\/code> was not specified then it runs a single action continuation synchronously! The behavior is different for async<\/code>\/await<\/code> and for task.ContinueWith<\/code> cases. A continuation of the async method is invoked synchronously unless the task is finished (**), synchronization context or non-default task scheduler are used. It means that an “async” continuation runs synchronously almost all the time when the awaited task is not finished!<\/p>\n
(**) This is kind-of a rare race condition. If an awaited task is finished at ‘await site’ (like at await finishedTask<\/code>) then an async method continues it’s execution synchronously. This situation is only possible when a task is finished in the middle of a MoveNext<\/code> call of a generated state machine.<\/p>\n
But the logic for continuations scheduled by Task.ContinueWith<\/code> is different: in this case, a StandardTaskContinuation<\/code><\/a> instance is created and added as a task’s continuation. This continuation runs asynchronously unless TaskContinuationOptions.ExecuteSynchronously<\/code><\/a> flag is specified regardless of the task creation options<\/strong>.<\/p>\n
We can actually check that the issue we faced at the beginning has nothing to do with TaskCompletionSource<\/code> per se and actually manifests itself with any tasks created without TaskCreationOptions.RunContinuationsAsynchronously<\/code>:<\/p>\n
static async Task WithAsync()\r\n{\r\n    Print(\"WithAsync\");\r\n    var task = Task.Run(\r\n        () => { Sleep(100); Print(\"In Task.Run\"); });\r\n    await Task.Yield();\r\n    await task;\r\n    await Task.Yield();\r\n    Print(\"After task await\");\r\n}\r\n \r\nstatic Task WithContinueWith()\r\n{\r\n    Print(\"WithContinueWith\");\r\n    var task = Task.Run(\r\n        () => { Sleep(100); Print(\"In Task.Run\"); });\r\n    var result = task.ContinueWith(\r\n        t => { Print(\"Inside ContinueWith\"); });\r\n    return result;\r\n}\r\n\r\n\r\n    \r\nawait WithContinueWith();\r\nawait WithAsync();<\/pre>\n
The output:<\/p>\n
WithAsync: 1\r\nIn Task.Run: 3\r\nAfter task await: 3\r\n\r\nWithContinueWith: 3\r\nIn Task.Run: 4\r\nInside ContinueWith: 5<\/pre>\n
As we can see, the block between await<\/code> statement and the rest of the method runs synchronously in the same thread that runs Task.Run<\/code>. But the continuation scheduled with task.ContinueWith<\/code> runs asynchronously in a different thread. We can change the behavior by using Task.Factory.StartNew<\/code> and providing TaskCreationOptions.RunContinuationsAsynchronously<\/code>:<\/p>\n
static async Task WithAsync()\r\n{\r\n    Print(\"WithAsync\");\r\n    var task = Task.Factory.StartNew(\r\n        () => { Sleep(100); Print(\"In Task.Factory.StartNew\"); },\r\n        TaskCreationOptions.RunContinuationsAsynchronously);\r\n    await task;\r\n    Print(\"After task await\");\r\n}<\/pre>\n
WithAsync: 1\r\nIn Task.Factory.StartNew: 3\r\nAfter task await: 4<\/pre>\n
How to solve the issue?<\/h4>\n
As we discussed already, there are two pieces involved here: 1) task creation options that control how to run continuation (synchronously, if possible, by default) and 2) a type of continuation.<\/p>\n
There is nothing you can do to control the behavior of async<\/code>\/await<\/code>. If you can’t control a task’s creation but want to run the continuations asynchronously you can explicitly call Task.Yield()<\/code> right after await<\/code> or switch to custom tasks altogether (which is hardly an option).<\/p>\n
But if you can, you should provide task creation options every time you use TaskCompletionSource<T><\/code>:<\/p>\n
static async Task WithAsync(TaskCreationOptions options)\r\n{\r\n    Print($\"WithAsync. Options: {options}\");\r\n    var tcs = new TaskCompletionSource<object>(options);\r\n \r\n    var setTask = Task.Run(\r\n        () => {\r\n            Sleep(100);\r\n            Print(\"Setting task's result\");\r\n            tcs.SetResult(null);\r\n            Print(\"Set task's result\");\r\n        });\r\n                \/\/await Task.Yield();\r\n \r\n    await tcs.Task;\r\n\r\n    Print(\"After task await\");\r\n    await setTask;\r\n}\r\n \r\nawait WithAsync(TaskCreationOptions.None);\r\nawait WithAsync(TaskCreationOptions.RunContinuationsAsynchronously);<\/pre>\n
The output is:<\/p>\n
WithAsync. Options: None: 1\r\nSetting task's result: 3\r\nAfter task await: 3\r\nSet task's result: 3\r\n\r\nWithAsync. Options: RunContinuationsAsynchronously: 3\r\nSetting task's result: 4\r\nSet task's result: 4\r\nAfter task await: 3<\/pre>\n
Starting from .NET 4.6.1 TaskCompletionSource<\/code> accepts TaskCreationFlags<\/code>. If the flag TaskCreationOptions.RunContinuationsAsynchronously<\/code> is specified, then all the continuations (including “async” continuations) are executed asynchronously. This will remove an implicit coupling that may occur when many async methods are chained together and one of the tasks in this chain is based on TaskCompletionSource<\/code>.<\/p>\n
Conclusion<\/h4>\n
    \n
  • TaskCompletionSource<\/code> class was introduced in .NET 4.0 in a pre async<\/code>-era for controlling a task’s lifetime manually.<\/li>\n
  • By default all the task’s continuations are executed synchronously unless TaskCreationOptions.RunContinuationsAsynchronously<\/code> option is specified.<\/li>\n
  • All the “async” continuations (blocks between await<\/code> statements) always run in a thread of an awaited task.<\/li>\n
  • TaskCompletionSource<\/code> instanced created with default constructor may cause deadlocks and other threading issues by running all “async” continuations in the thread that sets the result of a task.<\/li>\n
  • If you use .NET 4.6.1+ you should always provide TaskCreationOptions.RunContinuationsAsynchronously<\/code> when creating TaskCompletionSource<\/code> instances.<\/li>\n<\/ul>\n","protected":false},"excerpt":{"rendered":"
    … when used with async\/await.
    \nTaskCompletionSource class is a very useful facility if you want to control the lifetime of a task manually. I share a canonical example when TaskCompletionSource is used for converting the event-based asynchronous code to the Task-based pattern. <\/p>\n","protected":false},"author":4004,"featured_media":37840,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"_acf_changed":false,"footnotes":""},"categories":[6700,3912],"tags":[6695],"class_list":["post-1315","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-net-internals","category-async","tag-seteplia"],"acf":[],"blog_post_summary":"
    … when used with async\/await.
    \nTaskCompletionSource class is a very useful facility if you want to control the lifetime of a task manually. I share a canonical example when TaskCompletionSource is used for converting the event-based asynchronous code to the Task-based pattern. <\/p>\n","_links":{"self":[{"href":"https:\/\/devblogs.microsoft.com\/premier-developer\/wp-json\/wp\/v2\/posts\/1315","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/devblogs.microsoft.com\/premier-developer\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/devblogs.microsoft.com\/premier-developer\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/premier-developer\/wp-json\/wp\/v2\/users\/4004"}],"replies":[{"embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/premier-developer\/wp-json\/wp\/v2\/comments?post=1315"}],"version-history":[{"count":0,"href":"https:\/\/devblogs.microsoft.com\/premier-developer\/wp-json\/wp\/v2\/posts\/1315\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/premier-developer\/wp-json\/wp\/v2\/media\/37840"}],"wp:attachment":[{"href":"https:\/\/devblogs.microsoft.com\/premier-developer\/wp-json\/wp\/v2\/media?parent=1315"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/premier-developer\/wp-json\/wp\/v2\/categories?post=1315"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/premier-developer\/wp-json\/wp\/v2\/tags?post=1315"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}