Stopping dataflow sequence

Question

Hello.

I have 3 connected dataflow blocks:
TransformBlock b1, b2;
ActionBlock b3

b1.LinkTo(b2); b2.LinkTo(b3);

b1 - receives a lot of incoming messages, and execution of each takes long time (web downloads).
b2, b3 processes and persists results and this blocks requires much less time to execute one operation.

After some time I want to stop all this sequence and I call:

b1.Compleete();
b2.Compleete();
b3.Compleete();

And what I see:

b2, b3 stops process incoming messages, but b1 (it have a lot of messages in incoming queue) contionue to process.

How to terminate all this sequence and drop all incoming data?

Answer 1

It seems to me you don't want to “complete” the processing, you want to “cancel” it. The way to do that in TPL is using CancellationTokenSource and CancellationToken. You can pass one to a block using its options. In the case of a TransformBlock, it would look like this:

var cts = new CancellationTokenSource();
var b1 = new TransformBlock<InputType, OutputType>(/* whatever */, new ExecutionDataflowBlockOptions { CancellationToken = cts.Token });

To actually cancel it, you can then call cts.Cancel(). Note that cancellation is basically treated as an exceptional case, so for example b1.Completion.Wait() will throw an exception after you canceled it.