【译】You probably should stop using a custom TaskScheduler

简介: 以更明确的方式控制并发我认为并发控制(又称速率限制)是应用程序非常重要的方面,重要的方面应该是明确的。TaskScheduler 相当低级别的工具,我宁愿拥有更高级别的工具。如果工作是 CPU 密集型的,那么 PLINQ 或类似 ActionBlock TPL DataFlow 的东西可能是更好的选择。 如果工作主要是 IO 绑定和异步的,那么可以使用 Parallel.ForEachAsync 或 Polly.RateLimiting 基于 的 SemaphoreSlim 自定义帮助程序类。结论

如果你不知道什么是TaskScheduler 或你的项目中没有它的自定义实现,你可能可以跳过这篇文章。但如果你不知道它是什么,但你的项目中确实有一两个,那么这篇文章绝对适合你。

让我们从基础开始。任务并行库(也称为TPL)引入于2010年的NET 4.0。当时它主要用于并行编程,而不是异步编程,因为异步编程在C#4和NET 4.0中不是一等公民。

例如,体现在TPL API中,Task.Factory.StartNew的入参为委托,返回voidT,而不是TaskTask<T>:

var task = Task.Factory.StartNew(() =>
   {
    Console.WriteLine("Starting work...");
    Thread.Sleep(1000);
    Console.WriteLine("Done doing work.");
  });

Task.Factory.StartNew 有相当多的重载,其中一个需要 TaskScheduler .这是一种定义如何在运行时执行任务的策略。

默认情况下(如果未传递自定义 TaskScheduler 项,同时 TaskCreationOptions.LongRunning 未传递自定义项),则使用默认 TaskScheduler 。这是一个称为 ThreadPoolTaskScheduler 的内部类型,它使用 .NET 线程池来管理任务。(如果 传递TaskCreationOptions.LongRunning参数 给 Task.Factory.Startnew ,则使用专用线程来避免长时间使用线程池中的线程)。

与任何新技术一样,当 TPL 发布时,书呆子们很兴奋,并试图尽可能多地使用(和滥用)新技术。如果Microsoft给你一个可扩展的库,有些人认为这是一个好主意......你知道的。。。扩展它。

最常见的模式之一是并发限制,它使用固定数量的专用线程来确保您不会超额订阅 CPU:

public sealed class DedicatedThreadsTaskScheduler : TaskScheduler
{
    private readonly BlockingCollection<Task> _tasks = new BlockingCollection<Task>();
    private readonly List<Thread> _threads;
    public DedicatedThreadsTaskScheduler(int threadCount)
    {
        _threads = Enumerable.Range(0, threadCount).Select(i =>
        {
            var t = new Thread(() =>
            {
                foreach (var task in _tasks.GetConsumingEnumerable())
                {
                    TryExecuteTask(task);
                }
            })
            {
                IsBackground = true,
            };
            t.Start();
            return t;
        }).ToList();
    }
    protected override void QueueTask(Task task) => _tasks.Add(task);
    public override int MaximumConcurrencyLevel => _threads.Count;
    protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued) => false;
    protected override IEnumerable<Task> GetScheduledTasks() => _tasks;
}

此外还有很多其他实现执行相同的操作: DedicatedThreadTaskScheduler 、、 DedicatedThreadsTaskSchedulerLimitedConcurrencyLevelTaskScheduler 甚至 IOCompletionPortTaskScheduler 使用 IO 完成端口来限制并发性。

无论实现和幻想如何,它们都做同样的事情:它们最多允许同时执行给定数量的任务。下面是一个示例,说明我们如何使用它来强制最多同时运行 2 个任务:

var sw = Stopwatch.StartNew();
// Passing 2 as the threadCount to make sure we have at most 2 pending tasks.
var scheduler = new DedicatedThreadsTaskScheduler(threadCount: 2);
var tasks = new List<Task>();
for (int i = 0; i < 5; i++)
{
    int num = i;
    var task = Task.Factory.StartNew(() =>
    {
        Console.WriteLine($"{sw.Elapsed.TotalSeconds}: Starting {num}...");
        Thread.Sleep((num + 1) * 1000);
        Console.WriteLine($"{sw.Elapsed.TotalSeconds}: Finishing {num}");
    }, CancellationToken.None, TaskCreationOptions.None, scheduler);
    
    tasks.Add(task);
}
await Task.WhenAll(tasks);

在本例中,我们在循环中创建任务,实际上它可能在某种请求中。下面是输出:

0.0154143: Starting 0...
0.0162219: Starting 1...
1.0262272: Finishing 0
1.0265169: Starting 2...
2.0224863: Finishing 1
2.0227441: Starting 3...
4.0417418: Finishing 2
4.041956: Starting 4...
6.0332304: Finishing 3
9.0453789: Finishing 4

正如你所看到的,一旦任务 0 完成,我们会立即安排任务 1 等,所以实际上我们在这里限制了并发性。

但是让我们做一点点小小的改动:

static async Task FooBarAsync()
{
    await Task.Run(() => 42);
}
...
var task = Task.Factory.StartNew(() =>
{
    Console.WriteLine($"{sw.Elapsed.TotalSeconds}: Starting {num}...");
    Thread.Sleep((num + 1) * 1000);
    FooBarAsync().GetAwaiter().GetResult();
    Console.WriteLine($"{sw.Elapsed.TotalSeconds}: Finishing {num}");
}, CancellationToken.None, TaskCreationOptions.None, scheduler);

输出为:

0.0176502: Starting 1...
0.0180366: Starting 0...

是的。死锁了!为什么?让我们更新一个示例以更好地查看问题:让我们跟踪当前 TaskScheduler 并将循环中创建的任务数减少到 1:

static void Trace(string message) => 
    Console.WriteLine($"{message}, TS: {TaskScheduler.Current.GetType().Name}");
static async Task FooBarAsync()
{
    Trace("Starting FooBarAsync");
    await Task.Run(() => 42);
    Trace("Finishing FooBarAsync");
}
static async Task Main(string[] args)
{
    var sw = Stopwatch.StartNew();
    var scheduler = new DedicatedThreadsTaskScheduler(threadCount: 2);
    var tasks = new List<Task>();
    for (int i = 0; i < 1; i++)
    {
        int num = i;
        var task = Task.Factory.StartNew(() =>
        {
            Trace($"{sw.Elapsed.TotalSeconds}: Starting {num}...");
            Thread.Sleep((num + 1) * 1000);
            FooBarAsync().GetAwaiter().GetResult();
            Trace($"{sw.Elapsed.TotalSeconds}: Finishing {num}...");
        }, CancellationToken.None, TaskCreationOptions.None, scheduler);
        
        tasks.Add(task);
    }
  Trace("Done scheduling tasks...");
    await Task.WhenAll(tasks);
}

输出为:

0.018728: Starting 0..., TS: DedicatedThreadsTaskScheduler
Starting FooBarAsync, TS: DedicatedThreadsTaskScheduler
Finishing FooBarAsync, TS: DedicatedThreadsTaskScheduler
1.028004: Finishing 0..., TS: DedicatedThreadsTaskScheduler
Done scheduling tasks..., TS: ThreadPoolTaskScheduler

现在应该相对容易理解发生了什么以及为什么当我们尝试运行超过 2 个任务时会陷入死锁。请记住,异步方法中的每个步骤(关键字 await后的代码)本身就是一个任务,由任务调度程序逐个执行。默认情况下,任务调度程序是粘性的:如果TaskScheduler是在创建任务时提供的,那么所有后续的Task都将使用相同的TaskScheduler这意味着TaskScheduler贯穿所有异步方法中的 awaits。

在我们的例子中,这意味着当完成 FooAsync时 ,我们 DedicatedThreadsTaskScheduler 被调用来运行它的后续的Task(译者注:即await Task.Run(() => 42);)。但是它已经忙于运行所有任务,因此它无法在 FooAsync 末尾运行一段微不足道的代码。而且由于 FooAsync 无法完成,我们无法立即完成Task。导致死锁。

我们能做些什么来解决这个问题?

解决方案

有几种方法可以避免此问题:

1. Use ConfigureAwait(false)

static async Task FooBarAsync()
{
    Trace("Starting FooBarAsync");
    await Task.Run(() => 42);
    Trace("Finishing FooBarAsync");
}

我们在这里看到的问题与UI案例中的死锁非常相似,当任务被阻塞并且单个UI线程无法运行继续时。

我们可以通过确保每个异步方法都有 ConfigureAwait(false) 来避免这个问题。下面是具有以下 FooBarAsync 的实现时的输出。

static async Task FooBarAsync()
{
    Trace("Starting FooBarAsync");
    await Task.Run(() => 42).ConfigureAwait(false);
    Trace("Finishing FooBarAsync");
}
0.0397394: Starting 0..., TS: DedicatedThreadsTaskScheduler
Starting FooBarAsync, TS: DedicatedThreadsTaskScheduler
**Finishing FooBarAsync, TS: ThreadPoolTaskScheduler**
1.0876967: Finishing 0..., TS: DedicatedThreadsTaskScheduler

有人可能会说这是解决这个问题的正确方法,但我不同意。在我们的一个项目中,有一个实际案例,一个很难修复的库代码中存在阻塞异步方法。你可以通过使用分析器来确保你的代码遵循最佳实践,但期望每个人都遵循这些最佳实践是不切实际的。

(译者注:同样可以使用Fody来自动实现追加.ConfigureAwait(false);)

这里最大的问题是,这是一个不常见的情况。有许多后端系统在没有 ConfigureAwait(false) 的情况下工作得很好,因为团队没有任何带有同步上下文的 UI,而且任务调度程序的行为方式相同这一事实并不广为人知。

我只是觉得有更好的选择。

2. 以更明确的方式控制并发

我认为并发控制(又称速率限制)是应用程序非常重要的方面,重要的方面应该是明确的。

TaskScheduler 相当低级别的工具,我宁愿拥有更高级别的工具。如果工作是 CPU 密集型的,那么 PLINQ 或类似 ActionBlock TPL DataFlow 的东西可能是更好的选择。

 

如果工作主要是 IO 绑定和异步的,那么可以使用 Parallel.ForEachAsyncPolly.RateLimiting 基于 的 SemaphoreSlim 自定义帮助程序类。

结论

自定义TaskScheduler 只是一个工具,与任何工具一样,它可能被正确或错误地使用。如果您需要一个了解 UI 的调度程序,那TaskScheduler 适合您。但是,是否应该在应用中使用一个进行并发和并行控制?我会投反对票。如果团队可能在多年前有正当理由来使用,但请仔细检查这些理由今天是否存在。

是的,请记住,阻塞异步调用可能会以多种方式反噬,TaskScheduler 只是其中之一。因此,我建议对每个阻塞异步调用的地方进行备注,解释为什么您认为这样做既安全又有用。

分类: datausus.net

相关文章
|
5月前
|
Kubernetes 容器
Warning FailedScheduling 14m (x12 over 16m) default-scheduler 0/1 nodes are available: 1 node(s
Warning FailedScheduling 14m (x12 over 16m) default-scheduler 0/1 nodes are available: 1 node(s
114 0
|
5月前
|
前端开发 JavaScript 算法
shouldComponentUpdate 是做什么的?
shouldComponentUpdate 是做什么的?
146 0
|
10月前
|
Kubernetes 算法 调度
基于kube-scheduler-simulator编写自己的调度程序
基于kube-scheduler-simulator编写自己的调度程序
111 0
|
数据库
This scheduler instance (XXXXX) is still active but was recovered by another
This scheduler instance (XXXXX) is still active but was recovered by another
215 0
|
数据库
This scheduler instance is still active but was recovered by another instance in the cluster
This scheduler instance is still active but was recovered by another instance in the cluster
833 0
|
Java 调度
Leetcode-Medium 621. Task Scheduler
Leetcode-Medium 621. Task Scheduler
112 0
Leetcode-Medium 621. Task Scheduler
|
JavaScript 安全 前端开发
What Is ElectronJS and Why Should You Use It?
In this three-part tutorial, we will explore how to create a fully functional invoice application using ElectronJS and ApsaraDB for MongoDB.
2656 0
What Is ElectronJS and Why Should You Use It?