【译】The danger of TaskCompletionSource class-编程知识

【译】The danger of TaskCompletionSource class

news/2024/11/15 6:20:20/文章来源:https://www.cnblogs.com/zhangchen-trunk/p/18302405

来自 Sergey Tepliakov的另一篇 https://devblogs.microsoft.com/premier-developer/the-danger-of-taskcompletionsourcet-class/#comments

当使用async/await时，如果您想手动控制任务的生存期，TaskCompletionSource<T>类是一个非常有用的工具。下面是TaskCompletionSource的一个示例，用于将基于事件的异步代码转换为基于任务的模式。

public static Task PerformOperation(this PictureBox pictureBox)
{
    var tcs = new TaskCompletionSource<object>();
            
    // Naive version that does not unsubscribe from the event
    pictureBox.LoadCompleted += (s, ea) =>
    {
        if (ea.Cancelled) tcs.SetCanceled();
        else if (ea.Error != null) tcs.SetException(ea.Error);
        else tcs.SetResult(null);
    };    pictureBox.LoadAsync();    return tcs.Task;
}

事实上， TaskCompletionSource<T> 代表了一个未来的结果并提供了通过调用 SetCanceled SetException 或 SetResult 方法手动设置任务的最终状态的能力。

这个类非常有用，不仅当你需要使旧代码看起来很现代和花哨时。 TaskCompletionSource<T> 用于手动控制一个任务的生命周期的各种情况，例如在不同的通信协议中。所以，让我们模仿其中之一。

假设我们要创建一个自定义数据库的adapter类。adapter类将具有用于处理请求的专用“工作线程”线程， ExecuteAsync 以及客户端可用于计划后台处理工作的方法。这与实际的 Redis 客户端执行的操作非常相似，其他一些数据库客户端也遵循相同的模式，因此这不是一个牵强的场景。

public class DatabaseFacade : IDisposable
{private readonly BlockingCollection<(string item, TaskCompletionSource<string> result)> _queue =new BlockingCollection<(string item, TaskCompletionSource<string> result)>();private readonly Task _processItemsTask;public DatabaseFacade() => _processItemsTask = Task.Run(ProcessItems);public void Dispose() => _queue.CompleteAdding();public Task SaveAsync(string command){var tcs = new TaskCompletionSource<string>();_queue.Add((item: command, result: tcs));return tcs.Task;}private async Task ProcessItems(){foreach (var item in _queue.GetConsumingEnumerable()){Console.WriteLine($"DatabaseFacade: executing '{item.item}'...");// Waiting a bit to emulate some IO-bound operationawait Task.Delay(100);item.result.SetResult("OK");Console.WriteLine("DatabaseFacade: done.");}}
}

该代码并不能用于生产环境，但它是基于 BlockingCollection 的生产者-消费者模式的一个很好的示例。

假设我们有另一个组件，比如说一个logger。logger通常也使用生产者-消费者模式实现。出于性能原因，我们不希望在每次方法调用时刷新消息，而是可以使用阻塞集合和专用线程将数据保存到外部源。其中一个外部来源可能是数据库。

public class Logger : IDisposable
{private readonly DatabaseFacade _facade;private readonly BlockingCollection<string> _queue =new BlockingCollection<string>();private readonly Task _saveMessageTask;public Logger(DatabaseFacade facade) =>(_facade, _saveMessageTask) = (facade, Task.Run(SaveMessage));public void Dispose() => _queue.CompleteAdding();public void WriteLine(string message) => _queue.Add(message);private async Task SaveMessage(){foreach (var message in _queue.GetConsumingEnumerable()){// "Saving" message to the fileConsole.WriteLine($"Logger: {message}");// And to our database through the facadeawait _facade.SaveAsync(message);}}
}

这个logger的实现非常粗浅，无论如何，您不应该自己另写一个logger。我在这里的目标是展示两个生产者-消费者队列如何相互影响，logger是一个相当广泛、易于理解的概念。

译者注：即logger作为生产者, adapter作为消费者。

问题是：你能在这里看到代码有问题吗？一个非常严重的问题！

让我们尝试运行以下代码：

using (var facade = new DatabaseFacade())
using (var logger = new Logger(facade))
{logger.WriteLine("My message");await Task.Delay(100);await facade.SaveAsync("Another string");Console.WriteLine("The string is saved");
}

输出为：

Logger: My message
DatabaseFacade: executing 'My message'...

我们永远无法将"Another string"保存到数据库中。为什么？因为DatabaseFacade的线程被Logger的线程阻塞。

TaskCompletionSource 类型有一个非常奇特的行为：默认情况下，当调用方法时 SetResult后，后续的Task的“异步”延续都会以同步的方式进行。这就是我们的例子中发生的事情（ SetCancelled 和 SetException 以及它们的 TrySetXXX 对应物也是如此）：

这意味着两个“队列”隐式链接在一起，logger的队列会阻止adapter的队列。

译者注：这里的错误比较抽象，即item.result.SetResult("OK");语句执行后，该线程跳去执行await _facade.SaveAsync(message);的下一句（也就是去foreach了），因为“当调用方法时 SetResult后，后续的Task的“异步”延续都会以同步的方式进行”。

不幸的是，这样的情况相对普遍，我在我的项目中遇到过几次。当任务的“延续”（注1）以 TaskCompletionSource<T> 的方式实现时，可能会出现此问题，从而阻止调用 SetResult 的线程。

注1: 正如我们稍后将看到的，不同类型的Task延续表现不同。

这些问题的主要挑战是很难理解根本原因。一旦从生产机器抓取出问题的dump，您可能根本看不到任何明显的问题。你可能有一堆线程在等待内核对象，而在任何堆栈跟踪中没有任何相关用户的代码。

现在让我们看看为什么会发生这种情况，以及我们如何缓解这个问题。

每个Task都有一个名为 m_stateFlags 的字段，该字段表示任务的当前状态（如 RanToCompletion 、 Cancelled Failed 等）。但这并不是该字段的唯一作用：它还包含一组在任务创建期间通过指定 TaskCreationOptions 的标志。这些标志控制不同的方面，例如是否在专用线程（ TaskCreationOptions.LongRunning ）中运行任务，是否将工作项调度到全局队列而不是线程本地队列（ TaskCreationOptions.PreferFairness ），或者是否强制任务继续始终异步运行（ TaskCreationOptions.RunContinuationsAsynchronously）。

显然，我们对后一个方面感兴趣，我们将立即看到如何指定此标志。但要完全理解这个问题，我们还需要看另一个方面：我们需要了解任务的延续性。

static async Task WithAsync()
{var task = Task.Run(() => { Sleep(100); });await task;Console.WriteLine("After task await");
}static Task WithContinueWith()
{var task = Task.Run(() => { Sleep(100); });return task.ContinueWith(t => { Console.WriteLine("Inside ContinueWith"); },TaskContinuationOptions.OnlyOnRanToCompletion);
}

您可能认为这两种实现是等效的，因为编译器只是将 await 之后代码块“移动”到通过调度的实现的ContinueWith 中。但这只是一部分事实，实际的逻辑更复杂一些。

在我的另一篇文章“剖析 C# 中的异步方法”中更详细地描述了 C# 编译器对异步方法所做的实际转换，在这里我们将重点介绍一个特定方面：延续调度。

当一个正在await的Task未完成时，生成的状态机将调用 TaskAwaiter.UnsafeOnCompleted 并传递一个回调，该回调在等待任务完成时调用，以向前移动状态机的状态。此方法调用 Task.SetContinuationForAwait 将给定操作添加为任务的延续：

// Now register the continuation, and if we couldn't register it because the task is already completing,
// process the continuation directly (in which case make sure we schedule the continuation
// rather than inlining it, the latter of which could result in a rare but possible stack overflow).
if (tc != null)
{if (!AddTaskContinuation(tc, addBeforeOthers: false))tc.Run(this, bCanInlineContinuationTask: false);
}
else
{Contract.Assert(!flowExecutionContext, "We already determined we're not required to flow context.");if (!AddTaskContinuation(continuationAction, addBeforeOthers: false))AwaitTaskContinuation.UnsafeScheduleAction(continuationAction, this);
}

局部变量 tc 在有同步上下文时不为 null，否则将调用该 else 块。首先， AddTaskContinuation 方法将被调用，该方法在当前任务未完成时返回 true （以防止栈溢出），同时一个指定的action委托成功添加为当前任务的延续。否则 UnsafeScheduleAction 方法将被调用创建 AwaitTaskContinuation 实例。

译者注：这段代码主要应该走if内的逻辑，也就是在当前任务完成时直接执行后续任务，否则就先添加作为延续。

在一般情况下（稍后会详细介绍）， System.Action 实例被添加为任务延续，并且延续存储在 Task.m_continuationObject 中。

现在，让我们看看任务完成后会发生什么（代码片段来自 Task.FinishContinuations ）：

internal void FinishContinuations()
{// Atomically store the fact that this task is completing.  From this point on, the adding of continuations will// result in the continuations being run/launched directly rather than being added to the continuation list.// 取出延续的任务，并将当前延续任务置为空object continuationObject = Interlocked.Exchange(ref m_continuationObject, s_taskCompletionSentinel);// If continuationObject == null, then we don't have any continuations to processif (continuationObject != null){// Skip synchronous execution of continuations if this task's thread was abortedbool bCanInlineContinuations = // 取非即为 任务未被中止且TaskCreationOptions未明确置为异步延续!(// 前两个为任务被中止((m_stateFlags & TASK_STATE_THREAD_WAS_ABORTED) != 0) ||(Thread.CurrentThread.ThreadState == ThreadState.AbortRequested) ||// 最后一个为TaskCreationOptions为异步延续((m_stateFlags & (int)TaskCreationOptions.RunContinuationsAsynchronously) != 0));// Handle the single-Action case// 把延续任务转为委托Action singleAction = continuationObject as Action;if (singleAction != null){AwaitTaskContinuation.RunOrScheduleAction(singleAction, bCanInlineContinuations, ref t_currentTask);return;}// The rest of the body}
}

FinishContinuations 函数检查任务创建标志，如果 RunContinuationsAsynchronously 未指定，则同步运行单个操作继续！这种行为对于案例中的 async / await 和task.ContinueWith 是不同的。使用同步上下文或非默认TaskScheduler时，除非任务已完成（注2），异步方法的延续Task都会同步调用。这意味着当等待的任务未完成时，“异步”延续几乎一直同步运行！

注2：这是一种罕见的竞争条件，如果等待的任务在“await site”完成（如在await finishedTask ），则异步方法将继续同步执行。仅当任务在调用生成的状态机的过程中 MoveNext 完成时，才有可能出现这种情况。（译者注：这里没太懂，大概就是一种特殊情况也会同步执行）

但是，安排 Task.ContinueWith 的延续的逻辑是不同的：在这种情况下，将创建一个 StandardTaskContinuation 实例并添加为任务的延续。除非 TaskContinuationOptions.ExecuteSynchronously 指定了标志，否则此延续将异步运行，而不考虑任务创建选项。

我们实际上可以检查我们一开始遇到的问题是否与 TaskCompletionSource 本身无关，并且实际上表现为在没有 TaskCreationOptions.RunContinuationsAsynchronously 以下情况下创建的任何任务：

static async Task WithAsync()
{Print("WithAsync");var task = Task.Run(() => { Sleep(100); Print("In Task.Run"); });await Task.Yield();await task;await Task.Yield();Print("After task await");
}static Task WithContinueWith()
{Print("WithContinueWith");var task = Task.Run(() => { Sleep(100); Print("In Task.Run"); });var result = task.ContinueWith(t => { Print("Inside ContinueWith"); });return result;
}await WithContinueWith();
await WithAsync();

输出：

WithAsync: 1
In Task.Run: 3
After task await: 3WithContinueWith: 3
In Task.Run: 4
Inside ContinueWith: 5

正如我们所看到的，在不同await语句之间的代码块和剩余的方法运行在 Task.Run同一线程中。但是task.ContinueWith 继续在不同的线程中异步运行。我们可以通过使用 Task.Factory.StartNew 和提供 TaskCreationOptions.RunContinuationsAsynchronously 以下方式来改变行为：

static async Task WithAsync()
{Print("WithAsync");var task = Task.Factory.StartNew(() => { Sleep(100); Print("In Task.Factory.StartNew"); },TaskCreationOptions.RunContinuationsAsynchronously);await task;Print("After task await");
}

WithAsync: 1
In Task.Factory.StartNew: 3
After task await: 4

如何解决这个问题？

正如我们已经讨论过的，这里涉及两个部分：1）任务创建时的参数（即TaskCreationOptions.RunContinuationsAsynchronously） 2）延续的类型（即刚刚的两种情况）。

您完全无法控制 async / await 执行时的行为。如果您无法控制任务的创建，但想要Task运行异步延续，则可以在await 之后显式调用 Task.Yield() 或者完全切换到自定义的Task（这两种方式几乎不可行）。

译者注：即第一个例子的问题这么改也能解决，大家可以自行尝试。但这样改需要对Task有足够的认识。

private async Task SaveMessage()
{foreach (var message in _queue.GetConsumingEnumerable()){Console.WriteLine($"Logger: {message}");await _facade.SaveAsync(message);// 显式调用 Task.Yield()异步延续await Task.Yield();}
}

但是，如果可以的话，您应该在每次使用 TaskCompletionSource<T> 时都提供TaskCreationOptions：

static async Task WithAsync(TaskCreationOptions options)
{Print($"WithAsync. Options: {options}");var tcs = new TaskCompletionSource<object>(options);var setTask = Task.Run(() => {Sleep(100);Print("Setting task's result");tcs.SetResult(null);Print("Set task's result");});//await Task.Yield();await tcs.Task;Print("After task await");await setTask;
}await WithAsync(TaskCreationOptions.None);
await WithAsync(TaskCreationOptions.RunContinuationsAsynchronously);

输出为：

WithAsync. Options: None: 1
Setting task's result: 3
After task await: 3
Set task's result: 3WithAsync. Options: RunContinuationsAsynchronously: 3
Setting task's result: 4
Set task's result: 4
After task await: 3

从 .NET 4.6.1 TaskCompletionSource 开始接受 TaskCreationFlags .如果指定了该标志 TaskCreationOptions.RunContinuationsAsynchronously ，则所有延续（包括“异步”延续）都将异步执行。这将消除当许多异步方法链接在一起并且该链中的一个任务基于 TaskCompletionSource时可能发生的隐式耦合。