TransformBlock never completes

Question

I am trying to wrap my head around “completion” in TPL data stream blocks. In particular, TransformBlock never seems to end. Why?

Program example

My code calculates the square of all integers from 1 to 1000. For this, I used BufferBlock and a TransformBlock . Later in my code, I wait for the TransformBlock to complete. However, the block never ends, and I do not understand why.

 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); } } } } 

At first I thought I created a deadlock, but this does not seem to be the case. When I inspected the calculatorBlock.Completion task in the debugger, its Status property was set to WaitingForActivation . This was the moment when my brain was blue.

Answer 1

The reason your pipeline freezes is because both BufferBlock and TransformBlock do not seem to fill until they empty the elements (I assume the desired behavior of IPropagatorBlock , although I did not find on it documentation).

This can be verified with a more minimal example:

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

This blocks indefinitely unless you add bufferBlock.Receive(); to completion.

If you remove elements from your pipeline before blocking your TryReceiveAll code TryReceiveAll , connecting another ActionBlock to the pipeline, converting your TransformBlock to ActionBlock or in any other way, it will no longer block.

---

As for your specific solution, it seems that you do not need BufferBlock or TransformBlock , since blocks have an input queue for themselves, and you are not using the return value of TransformBlock . This can only be achieved with 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(); 

Answer 2

I think I understand that now. A TransformBlock instance is not considered “complete” until the following conditions are met:

• TransformBlock.Complete() was called
• InputCount == 0 - the block applied its conversion to each input element
• OutputCount == 0 - all converted elements left the output buffer

My program does not have a target block that is associated with the original TransformBlock , so the original block never flushes its output buffer.

As a workaround, I added a second BufferBlock , which is used to store the converted elements.

 static void Main(string[] args) { var inputBufferBlock = 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 }); var outputBufferBlock = new BufferBlock<int>(); using (inputBufferBlock.LinkTo(calculatorBlock, new DataflowLinkOptions { PropagateCompletion = true })) using (calculatorBlock.LinkTo(outputBufferBlock, new DataflowLinkOptions { PropagateCompletion = true })) { foreach (var number in Enumerable.Range(1, 1000)) { inputBufferBlock.Post(number); } inputBufferBlock.Complete(); calculatorBlock.Completion.Wait(); IList<int> results; if (outputBufferBlock.TryReceiveAll(out results)) { foreach (var result in results) { Console.WriteLine("x² = {0}", result); } } } }