Блок потока данных TPL никогда не завершается на PropagateCompletion

Question

Со времени последнего изменения моего конвейера с распространенным концом один из моих блоков буфера никогда не завершается. Пусть мне подвести итог, что работает, а что нет больше:

Ранее работал:

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 никогда не достигает завершения, даже если блок связан с его завершением и распространяет свое завершение, и блок связан от он получает все буферизованные элементы.

Answer 1

К ожиданию завершения A ни один из оставшихся кодов не выполняется до тех пор, пока A не завершится. Вот как работает работа - код после того, как он завернут в продолжение, готовое к завершению ожидаемого кода. Таким образом, в этом сценарии B связан с A после того, как A завершает, так что пополнение не распространяется, я думаю.

Answer 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