纯粹的计算任务顺序执行

Question

我有几个计算密集的方法,我试图使用async-await并行运行.

我有一个大约80,000个对象的列表,我将其输入到一个返回任务的函数中:

public static void Main(string[] args)
{
    //...blah blah blah...

    var runner = new Runner(); //in a nutshell, I manage to get an object that has an async method on it.
    runner.Run().Wait(); //and I wait for it to complete.

    //...blah blah blah...
}

我在我的跑步者对象中有以下方法(或多或少......这是一个人为的例子):

public async Task Run()
{
    var items = ... //this is my list
    var tasks = items.Select(i => this.RunItemAsync(i)).ToArray();

    //I don't get here until the tasks are all finished...every single one...

    await Task.WhenAll(tasks).ConfigureAwait(false);
}

private async Task RunItemAsync(Item i)
{
    var subItems = i.GetSubItems();

    var tasks = subItems.Select(s => s.RunSubItemAsync(s)).ToArray();

    //I don't get here until the sub item tasks are all finished...

    await Task.WhenAll(tasks).ConfigureAwait(false);

    //does computations, doesn't wait on any async i/o, etc
    await this.ProcessAsync(i).ConfigureAwait(false);
}

private async Task RunSubItemAsync(SubItem s)
{
    //does computations, doesn't wait on any async i/o, etc
    ...
}

在过去一年左右的时间里,我一直在努力克服异步等待,有时候使用TPL Dataflow实现了出色的性能和一些非常酷的东西,但每隔一段时间我就会遇到这样的事情,我似乎无法得到"激活"其并行功能的任务.这个特殊项目将在具有~16个内核的服务器上运行,所以我真的想利用它.我的开发虚拟机只有2个核心分配给它,但是它仍然应该允许任务激活并并行运行(并且它在过去已经存在).

我的观察

• 我设法通过await Task.Delay(1).ConfigureAwait(false)在RunItemAsync方法的开头插入一个小的并行运行.据我所知,这创造了某种形式的"呼吸空间",允许其他任务使用该线程.然而,这是不够的,因为它很脏,不可靠,并且要求我有不可接受的延迟.
• 没有上述Delay调用,任务都在上面运行Main Thread.这对我来说是显而易见的,因为Main它是启动它的功能.我对此没有任何问题,但我曾经历过以下经验:在new Thread创建的线程上运行任务导致它无法使用默认任务调度程序运行,并且每个任务最终都在该线程上顺序运行.也许Main Thread这个类别的跌幅？

我的问题

我知道运行ToArray本身不会执行异步代码.但是,我想要发生的是当我的RunItemAsync方法达到它的第一个await时它将"停止"并允许调用的下一次迭代ToArray运行.

我也明白添加await Task.Delay工作因为它引起了我想要的上面.必须有一些方法来做到这一点,而不诉诸await Task.Delay......

如何并行启动所有这些计算绑定的任务,而不会无意中导致它们按顺序运行？

Answer 1

目前有四种主要的并发库/技术.

• async 最适合自然异步单一操作,如I/O.
• 任务并行库(TPL)最适合并行化CPU绑定工作.
• TPL Dataflow跨越async并行,为处理数据提供网格/流水线抽象.
• Reactive Extensions(Rx)在概念上类似于TPL Dataflow,但没有并行功能,而是具有大量与时间相关的功能.

在您的情况下,您想要使用TPL.一个简单的Parallel.ForEach应该足够好.

最后,同步代码(包括CPU绑定的并行代码)应该有一个同步API; 和异步代码应该有一个异步API.所以你希望你的API看起来是同步的,而不是异步的.

所以,像这样:

public static void Main(string[] args)
{
  var runner = new Runner();
  runner.Run();
}

public void Run()
{
  var items = ...
  Parallel.ForEach(items, i => this.RunItem(i));
}

private void RunItem(Item i)
{
  var subItems = i.GetSubItems();
  Parallel.ForEach(subItems, s => s.RunSubItem(s));
  this.Process(i);
}

private void RunSubItem(SubItem s)
{
  SemaphoreSlim.Wait(); // instead of WaitAsync
  ...
}