c# - 定义 BoundedCapacity 会降低性能

Question

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

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

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

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 的 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

<罢工>

更新:

我对你的话感到困惑

According to manual, BoundedCapacity should be set to int.MaxValue by default

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

BoundedCapacity
The majority of the dataflow blocks included in System.Threading.Tasks.Dataflow.dll support the specification of a bounded capacity.
This is the limit on the number of items the block may be storing and have in flight at any one time.
By default, this value is initialized to DataflowBlockOptions.Unbounded (-1), meaning that there is no limit.

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

var options = new ExecutionDataflowBlockOptions();

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

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

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

MaxDegreeOfParallelism
By default, an individual dataflow block processes only one message at a time, queueing all not-yet-processed messages so that they may be processed when the currently processing message is completed.

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

If set to DataflowBlockOptions.Unbounded (-1), any number of messages may be processed concurrently, with the maximum automatically managed by the underlying scheduler targeted by the dataflow block.

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

那你的reasonable expectations是什么？为此？