What is TPL Dataflow, and when would you choose it over Channels or Parallel.ForEachAsync?

1. Why per-stage parallelism wins

Sequential:   [download][resize][upload][download][resize][upload]...
               network    CPU    network

Pipeline:     download x16 ──► resize xN(cores) ──► batch 50 ──► upload x8
              (I/O bound)       (CPU bound)                       (I/O bound, batched)
              all stages run at the same time on different items

Throughput is limited by the slowest stage; you size that stage's parallelism and let bounded capacities stop faster stages from running ahead.

2. How backpressure works between linked blocks

When a target block is at BoundedCapacity, it postpones offers from the source. The source keeps the item and offers it again when the target frees space. The source's own buffer then fills, so it postpones its upstream, and eventually SendAsync at the head waits. Without BoundedCapacity the whole chain degenerates into unbounded queues.

3. Error handling and dead-lettering

var deadLetter = new ActionBlock<(int Id, Exception Error)>(f => log.LogError(f.Error, "Item {Id} failed", f.Id));

var resize = new TransformManyBlock<Download, Resized>(async d =>
{
    try { return [await ResizeAsync(d)]; }
    catch (Exception ex)
    {
        await deadLetter.SendAsync((d.Id, ex));
        return [];                     // swallow this item, keep the pipeline alive
    }
}, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = Environment.ProcessorCount, BoundedCapacity = 100 });

An unhandled exception faults the block: it stops accepting input, drops buffered items, and (with PropagateCompletion) faults downstream completions, which is right for fatal errors and wrong for one bad item.

4. Linking with predicates: a trap

source.LinkTo(evenTarget, x => x % 2 == 0);
// odd items match no link -> they stay in source forever -> pipeline never completes
source.LinkTo(DataflowBlock.NullTarget<int>());   // or a target for the rest

Always provide a link for items that match no predicate.

5. Ordering cost

With MaxDegreeOfParallelism > 1 and EnsureOrdered = true, a slow item holds back faster ones behind it in the output buffer. If order doesn't matter, set EnsureOrdered = false for better throughput and lower memory.

6. Observability

• InputCount / OutputCount per block → queue lengths (export periodically as metrics).
• Processed/failed counters inside delegates.
• Completion status per block for diagnosing where a pipeline stopped.

7. Version notes

8. Try it yourself

1. Run the code example and change the parallelism per stage; observe total time.
2. Remove all BoundedCapacity values and feed 1,000,000 items; watch memory.
3. Replace SendAsync with Post and count false returns.
4. Remove PropagateCompletion from one link and observe await upload.Completion hanging.
5. Throw in the resize block for one item; compare unhandled vs dead-letter handling.
6. Add a predicate link without a fallback and see the pipeline never complete.

using System.Threading.Tasks.Dataflow; // NuGet: System.Threading.Tasks.Dataflow

var link = new DataflowLinkOptions { PropagateCompletion = true };

var download = new TransformBlock<int, byte[]>(async id =>
{
    await Task.Delay(50); // simulate an HTTP download
    return new byte[1024 + id];
}, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 16, BoundedCapacity = 100 });

var resize = new TransformBlock<byte[], int>(bytes =>
{
    return bytes.Length / 2; // simulate CPU work
}, new ExecutionDataflowBlockOptions
{
    MaxDegreeOfParallelism = Environment.ProcessorCount,
    BoundedCapacity = 100
});

var batch = new BatchBlock<int>(50, new GroupingDataflowBlockOptions { BoundedCapacity = 200 });

int written = 0;
var upload = new ActionBlock<int[]>(async items =>
{
    await Task.Delay(20); // simulate a bulk upload
    Interlocked.Add(ref written, items.Length);
}, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 2, BoundedCapacity = 4 });

download.LinkTo(resize, link);
resize.LinkTo(batch, link);
batch.LinkTo(upload, link);

for (int id = 0; id < 1_000; id++)
    await download.SendAsync(id); // respects BoundedCapacity

download.Complete();
await upload.Completion;
Console.WriteLine($"Uploaded {written} images");