TPL-定义的ExecutionDataflowBlockOptions BoundedCapacity降低了性能

Question

有什么方法可以通过TPL节流来限制性能下降吗？

我有一个复杂的组件管道，并试图限制所需的内存需求。我从多个文件中并行读取，管道中的组件可能会从这些文件的随机部分中读取一些内容，其余组件则进行CPU绑定操作。

我使用通用测试方法将性能测试平台简化为这些测试。

private void TPLPerformaceTest(int generateNumbers, ExecutionDataflowBlockOptions transformBlockOptions)
{
    var transformBlock = new TransformBlock<int, int>(i => i, transformBlockOptions);

    var storedCount = 0;
    var generatedCount = 0;
    var store = new ActionBlock<int>(i => Interlocked.Increment(ref storedCount));

    transformBlock.LinkTo(store);
    transformBlock.Completion.ContinueWith(_ => store.Complete());

    for (int i = 0; i < generateNumbers; i++)
    {
        transformBlock.SendAsync(i).Wait(); //To ensure delivery
        Interlocked.Increment(ref generatedCount);
    }
    transformBlock.Complete();
    store.Completion.Wait();

    Assert.IsTrue(generatedCount == generateNumbers);
    Assert.IsTrue(storedCount == generateNumbers);
}

第一个没有节流。在我的CPU上，大约需要12秒钟才能完成，消耗约800MB的RAM，平均CPU利用率约为35％。

[Test]
public void TPLPerformaceUnlimitedTest()
{
    var generateNumbers = 100000000;
    this.TPLPerformaceTest(generateNumbers,new ExecutionDataflowBlockOptions());
}

仅将BoundedCapacity设置为int.MaxValue的第二项测试完全没有限制，需要20到30秒才能完成，消耗2.1GB的RAM，平均CPU利用率约为50％。根据手册，BoundedCapacity默认情况下应设置为int.MaxValue，因此我看不到性能下降的原因。

[Test]
[Sequential]
public void TPLPerformaceBounedCapacityTest()
{
    var generateNumbers = 100000000;
    this.TPLPerformaceTest(generateNumbers,new ExecutionDataflowBlockOptions()
      { BoundedCapacity = Int32.MaxValue });
}

第三个测试限制为 BoundedCapacity generateNumbers / 1000，ergo100000。它需要60 秒钟才能完成，并消耗450MB的RAM，平均CPU利用率约为60％。

[Test]
[Sequential]
public void TPLPerformaceBounedCapacityTenthTest()
{
    var generateNumbers = 100000000;
    this.TPLPerformaceTest(generateNumbers,new ExecutionDataflowBlockOptions()
      { BoundedCapacity = generateNumbers / 1000 });
}

第四个测试将MaxDegreeOfParallelism限制为-1（根据手册无限制）。它消耗了27GB的RAM，平均CPU利用率约为85％，并且在5分钟内还没有完成。

[Test]
[Sequential]
public void TPLPerformaceMaxDegreeOfParallelismTest()
{
    var generateNumbers = 100000000;
    this.TPLPerformaceTest(generateNumbers, new ExecutionDataflowBlockOptions()
      { MaxDegreeOfParallelism = -1 });
}

由于我的合理期望，所有方法似乎都会非常严重地影响性能，并且无法正常运行。

Answer 1

您会因为以下原因而降低性能：

transformBlock.SendAsync(i).Wait(); //To ensure delivery

这将在交付完成之前阻止当前线程。您应该切换到await释放线程以继续执行其他任务：

await transformBlock.SendAsync(i); //To ensure delivery

更新：

我对你的话感到困惑

根据手册，BoundedCapacity应int.MaxValue默认设置为

因为这不是真的，所以来自官方文档：

BoundedCapacity
包含的大多数数据流块都System.Threading.Tasks.Dataflow.dll支持规范容量。
这是该块可以存储并在任何时间飞行的项目数量的限制。
默认情况下，此值初始化为DataflowBlockOptions.Unbounded（-1），表示没有限制。

运行此代码后，您可以在此处查看所有默认值：

var options = new ExecutionDataflowBlockOptions();

因此BoundedCapacity将set设置为的第二个测试确实增加了一个限制，这增加了对块缓冲区中的位置可用性的一些检查。int.MaxValue

您可以在第三项测试中看到类似的行为，该行为比第二项消耗更少的内存，但是对缓冲区进行更多的检查和等待时间以释放空间，因此它的运行速度较慢，但​​分配的内存很少。

另外，您可以在屏幕截图上看到MaxDegreeOfParallelism等于1：

MaxDegreeOfParallelism
默认情况下，单个数据流块一次仅处理一条消息，对所有尚未处理的消息进行排队，以便在当前处理的消息完成时可以对其进行处理。

将此参数设置为之后-1，您将打开潘多拉魔盒，因为所有消息都由同一任务计划程序同时执行，再次根据文档：

如果设置为DataflowBlockOptions.Unbounded（-1），则可以同时处理任意数量的消息，而最大消息数量则由数据流块针对的基础调度程序自动管理。

从内存消耗中可以看出，任务调度程序决定为每条消息启动新线程，因为线程池中没有可用的线程，1MB每条消息占用大约一个27000线程，因此您有一个大约与CPU相互竞争的线程时间。如您所见，他们并不擅长此事。
推荐的并行度通常为Environment.ProcessorCount，因此，如果要加快一个块的速度，可以将设置MaxDegreeOfParallelism为此属性。但是，在更复杂的情况下，这并非总是最佳选择，因为其他块将停止等待CPU时间。

那么，您reasonable expectations的目的是什么？