定义 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 = 100_000_000;
    this.TPLPerformaceTest(generateNumbers,new ExecutionDataflowBlockOptions());
}

第二个测试，只是将BoundedCapacity设置为int.MaxValue，因此完全没有限制，需要20-30s完成，消耗2.1GB strong> 的 RAM 和平均 CPU 利用率约为 50%。根据手册，BoundedCapacity 应该默认设置为 int.MaxValue，所以我看不出性能下降的原因。

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

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

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

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

[Test]
[Sequential]
public void TPLPerformaceMaxDegreeOfParallelismTest()
{
    var generateNumbers = 100_000_000;
    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

因为这不是真的，from official documentation:

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

在这里你可以看到运行这段代码后的所有默认值：

var options = new ExecutionDataflowBlockOptions();

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

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

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

MaxDegreeOfParallelism
默认情况下，单个数据流块一次只处理一条消息，将所有尚未处理的消息排队，以便在当前处理的消息完成时处理它们。

将此参数设置为-1后，你打开了潘多拉魔盒，因为所有消息同时被同一个任务调度器执行，再次according to the documentation：

如果设置为DataflowBlockOptions.Unbounded (-1)，则可以同时处理任意数量的消息，最大值由数据流块所针对的底层调度程序自动管理。

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

那么你的reasonable expectations 是什么？