c# - TransformBlock 永远不会完成

Question

我正试图围绕 TPL 数据流 block 中的“完成”进行思考。特别是，TransformBlock 似乎从未完成。为什么？

示例程序

我的代码计算从 1 到 1000 的所有整数的平方。为此我使用了一个 BufferBlock 和一个 TransformBlock。在我的代码后面，我等待 TransformBlock 完成。但是，该 block 从未真正完成，我不明白为什么。

static void Main(string[] args)
{
    var bufferBlock = new BufferBlock<int>();
    var calculatorBlock = new TransformBlock<int, int>(i =>
    {
        Console.WriteLine("Calculating {0}²", i);
        return (int)Math.Pow(i, 2);
    }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 8 });

    using (bufferBlock.LinkTo(calculatorBlock, new DataflowLinkOptions { PropagateCompletion = true }))
    {
        foreach (var number in Enumerable.Range(1, 1000))
        {
            bufferBlock.Post(number);
        }

        bufferBlock.Complete();

        // This line never completes
        calculatorBlock.Completion.Wait();

        // Unreachable code
        IList<int> results;
        if (calculatorBlock.TryReceiveAll(out results))
        {
            foreach (var result in results)
            {
                Console.WriteLine("x² = {0}", result);
            }
        }
    }
}

起初我以为我造成了僵局，但事实并非如此。当我在调试器中检查 calculatorBlock.Completion 任务时，它的 Status 属性设置为 WaitingForActivation。那是我大脑蓝屏的时刻。

Answer 1

您的管道挂起的原因是 BufferBlock 和 TransformBlock 显然在清空项目之前不会完成(我猜 所需的行为IPropagatorBlocks 尽管我还没有找到关于它的文档)。

这可以用一个更小的例子来验证:

var bufferBlock = new BufferBlock<int>();
bufferBlock.Post(0);
bufferBlock.Complete();
bufferBlock.Completion.Wait();

除非您在完成之前添加 bufferBlock.Receive();，否则这会无限期地阻塞。

如果您在被 TryReceiveAll 代码块阻止之前从管道中删除了项目，则将另一个 ActionBlock 连接到管道，转换您的 TransformBlock 到 ActionBlock 或任何其他不再阻止的方式。

---

关于您的具体解决方案，您似乎根本不需要 BufferBlock 或 TransformBlock，因为 block 本身有一个输入队列并且您不使用TransformBlock 的返回值。这可以通过 ActionBlock 实现:

var block = new ActionBlock<int>(
    i =>
    {
        Console.WriteLine("Calculating {0}²", i);
        Console.WriteLine("x² = {0}", (int)Math.Pow(i, 2));
    },
    new ExecutionDataflowBlockOptions {MaxDegreeOfParallelism = 8});
foreach (var number in Enumerable.Range(1, 1000))
{
    block.Post(number);
}
block.Complete();
block.Completion.Wait();