TPL 数据流块永远不会在 PropagateCompletion 上完成

Posted 2023-03-31

【中文标题】TPL 数据流块永远不会在 PropagateCompletion 上完成

【英文标题】：TPL Dataflow block never completes on PropagateCompletion

【发布时间】：2015-09-05 07:10:34

【问题描述】：

自从上次更改我的传播完成管道以来，我的一个缓冲区块永远不会完成。让我总结一下什么有效，什么不再有效：

以前的工作：

A.LinkTo(B, PropagateCompletion);
B.LinkTo(C, PropagateCompletion);
C.LinkTo(D, PropagateCompletion);
D.Receive();

// everything completes

不再工作：

A.LinkTo(B, PropagateCompletion);
C.LinkTo(D, PropagateCompletion);

await A.Completion;
someWriteOnceBlock.Post(B.Count);
// B.Complete(); commented on purpose
B.LinkTo(C, PropagateCompletion);

D.Receive();

// Only A reaches completion
// B remains in 'waiting for activation'
// C executes but obviously never completes since B doesn't either

如果我取消注释注释行，一切正常，但显然该行不应该是必需的。

不知何故，我的 BufferBlock B 永远不会完成，即使链接 to 的块已完成并传播其完成，并且链接的块 from 它接收所有缓冲项.

【问题讨论】：

你应该显示你的实际代码。

@i3arnon 我不能放整个代码，因为管道非常复杂，清理它以保留相关部分需要很长时间，但更重要的是，我管道中的所有内容都是很标准，我所有的积木都是链接的，一件事进去，一件事出来。 await 周围的那些行是唯一不寻常的事情，它们正是我拥有它们的方式，我相信在我完美工作的管道的其余部分中没有任何内容可以解释取消注释注释行使其工作，并且注释它会阻止 B永远完成。

你最后有运气吗？a

【参考方案1】：

通过等待 A 的完成，在 A 完成之前，不会执行任何剩余的代码。这就是 await 的工作方式 - 将代码包装在一个延续中，准备完成等待的代码。因此，在这种情况下，B 在 A 完成后链接到 A，因此我认为不会传播完成。

【讨论】：

首先，您可能是说 A 与 B 相关联（而不是 B 与 A 相关联），其次，您的解释仅证实了等待 A.Completion 是问题所在，显然我已经知道了，但事实并非如此没有说明传播完成发生了什么，它只是被忽略了吗？

【参考方案2】：

我遇到了同样的问题 - 我的最后一个块是一个 TransformBlock，它只是坐在那里等待完成。

只有在没有剩余要处理的输入、所有输出都已处理且已完成的情况下，一个块才会完成，方法是调用 block.Complete 或完成沿链传播。

我通过将最后一个块设为 ActionBlock 来解决此问题，因为这不会产生任何输出。这是我的代码，它需要一系列文本文件并反序列化它们的内容：

Dim maxDop = Environment.ProcessorCount
Dim executionOptions = New ExecutionDataflowBlockOptions With .MaxDegreeOfParallelism = maxDop
Dim linkOptions = New DataflowLinkOptions With .PropagateCompletion = True

Dim inputBlock = New BufferBlock(Of String)
Dim transformBlock = New TransformBlock(Of String, IEnumerable(of JsonObject))(Function(fileName) DeserialiseFromFileAsync(fileName)
Dim outputBlock As New BufferBlock(Of IEnumerable(of JsonObject))
Dim combineBlock = New ActionBlock(Of IEnumerable(of JsonObject))(Sub(col) ' Do something to add all the collections together)

inputBlock.LinkTo(transformBlock, linkOptions)
transformBlock.LinkTo(outputBlock, linkOptions)
outputBlock.LinkTo(combineBlock, linkOptions)

For fileNo = 1 To 10
    inputBlock.Post(String.Concat("JsonFile", fileNo, ".txt"))
Next

inputBlock.Complete() 'Complete the first block, propogation will handle the rest
Await combineBlock.Completion 'Await the last block completing