如何使用 BatchBlock 正确完成分支管道？

Question

考虑以下管道：

public static void Main()
{
    var firstBlock = new TransformBlock<string, string>(s =>
    {
         Console.WriteLine("FirstBlock");
         return s + "-FirstBlock";
    });
    var batchBlock = new BatchBlock<string>(100, new GroupingDataflowBlockOptions { Greedy = true });
    var afterBatchBlock = new TransformManyBlock<string[], string>(strings =>
    {
        Console.WriteLine("AfterBatchBlock");
        return new[] { strings[0] + "-AfterBatchBlock" };
    });
    var lastBlock = new ActionBlock<string>(s =>
    {
        Console.WriteLine($"LastBlock {s}");
    });
    firstBlock.LinkTo(batchBlock, new DataflowLinkOptions { PropagateCompletion = true }, x => x.Contains("0"));
    batchBlock.LinkTo(afterBatchBlock, new DataflowLinkOptions { PropagateCompletion = true });
        
    afterBatchBlock.LinkTo(lastBlock, new DataflowLinkOptions { PropagateCompletion = true });
        firstBlock.LinkTo(lastBlock, new DataflowLinkOptions { PropagateCompletion = true });

    firstBlock.Post("0");

    firstBlock.Complete();
    firstBlock.Completion.GetAwaiter().GetResult();
    batchBlock.Completion.GetAwaiter().GetResult();
    afterBatchBlock.Completion.GetAwaiter().GetResult();
    lastBlock.Completion.GetAwaiter().GetResult();
}

运行此代码会被以下输出阻止：

第一块
AfterBatchBlock

我认为幕后发生的事情如下：

• FirstBlock 向其所有链接目标（BatchBlock、AfterBatchBlock、LastBlock）发送完成信号
• BatchBlock 和 LastBlock 完成
• AfterBatchBlock 然后尝试将其输出传递给 LastBlock，但它已经完成并且永远卡住了。

我的问题是：

1. 这是一个错误吗？
2. 假设它是设计使然，克服我的管道达到的不良状态的推荐方法是什么

Answer 1

不，这不是错误。这是预期的行为。您必须首先意识到您处理的不是简单、直接的数据流管道，而是数据流网格。它不是一个复杂的网格，但仍然不是我所说的管道。处理形成网格的数据流块的完成可能很棘手。在您的情况下，您有一个块 lastBlock，它是两个源块 firstBlock 和 afterBatchBlock 的目标。此块应在两个源块完成时完成，而不是一个。因此，当将此块与其源链接时，您不能使用PropagateCompletion = true 选项。

没有内置的 API 可以做到这一点，但是实现一个二对一的完成传播器并不是很困难。事实上，你可以复制粘贴this answer 中的PropagateCompletion 方法，然后像这样使用它：

afterBatchBlock.LinkTo(lastBlock);
firstBlock.LinkTo(lastBlock);

PropagateCompletion(new IDataflowBlock[] { afterBatchBlock, firstBlock },
    new[] { lastBlock });