A Producer-Consumer Pipeline
Several stages joined by bounded channels, each with its own worker count, and the rules about completing and draining them.
ChannelReader<TOut> Stage(ChannelReader<TIn> input) // each stage owns and completes its output channelWhat a pipeline is
A pipeline is a chain of stages. Each stage does one job and passes its result to the next. Between any two stages sits a channel: the earlier stage writes into it, the later stage reads from it.
The point of the shape is that each stage gets its own number of workers, where a worker is one loop reading from that stage's input channel. Fetching web pages might want eight workers, because each spends its time waiting on the network. Writing to a single database connection wants one.
Who completes what
Every channel needs exactly one call to Complete, made by whoever owns the writing side. When a stage has several workers, no single worker may call it. The first worker to finish would close the channel while the others were still writing, and their next write would throw ChannelClosedException. Wait for all the workers with Task.WhenAll, then complete.
Where it ends up bounded
Give every channel a capacity. The pipeline then runs no faster than its slowest stage, and the number of items in memory is capped at the sum of the capacities plus the items being worked on right now. With unbounded channels a slow last stage is invisible until the process runs out of memory.
What trips people up
You must drain the last channel while the earlier stages are still running. If you await the producer first and the channels are bounded, the producer is waiting for room that nobody is making and the program stops forever. Start the stages, read the end of the pipeline, and await the stage tasks afterwards to surface their errors.
Example
using System.Threading.Channels;
Channel<int> ids = Channel.CreateBounded<int>(2);
Channel<string> results = Channel.CreateBounded<string>(2);
// Stage 1: one producer. It owns ids, so it is the one that completes ids.
Task produce = Task.Run(async () =>
{
for (int id = 1; id <= 6; id++)
{
await ids.Writer.WriteAsync(id);
}
ids.Writer.Complete();
});
// Stage 2: two workers share one reader, so each id is handled exactly once.
async Task WorkAsync()
{
await foreach (int id in ids.Reader.ReadAllAsync())
{
await results.Writer.WriteAsync($"item-{id}");
}
}
Task[] workers = [WorkAsync(), WorkAsync()];
// Close stage 2's output only once every worker has stopped writing to it.
Task close = Task.Run(async () =>
{
await Task.WhenAll(workers);
results.Writer.Complete();
});
// Stage 3: one consumer, draining while the stages above are still running.
List<string> collected = [];
await foreach (string item in results.Reader.ReadAllAsync())
{
collected.Add(item);
}
await Task.WhenAll(produce, close);
// Two workers finish in any order, so sort before printing.
collected.Sort(StringComparer.Ordinal);
Console.WriteLine(string.Join(", ", collected));
// Output:
// item-1, item-2, item-3, item-4, item-5, item-6
When to use it
- Download, parse and store thousands of URLs with a different worker count at each step.
- Move messages off a cloud queue through validation and into a database with a fixed cap on items in flight.
- Resize a folder of images with several workers while one final stage writes the files out.
- Give a long-running import job a memory ceiling that does not depend on how fast the source produces rows.
More examples
A stage that returns its reader
Returning ChannelReader<T> makes a stage composable: Label(Parse(Fetch(reader))) builds a whole pipeline, and each stage owns exactly one channel. The catch block is not optional. A stage that dies without completing its output leaves every later stage waiting on a channel nothing will ever close. The discarded task from Task.Run is fine only in this short illustration; in a real service, keep that Task so shutdown can await it.
using System.Threading.Channels;
Channel<int> source = Channel.CreateBounded<int>(2);
await source.Writer.WriteAsync(2);
await source.Writer.WriteAsync(3);
source.Writer.Complete();
await foreach (string item in Stages.Label(source.Reader).ReadAllAsync())
{
Console.WriteLine(item);
}
static class Stages
{
// The stage creates its output channel, so the stage completes it.
// Callers only ever see the reader and cannot write to it.
public static ChannelReader<string> Label(ChannelReader<int> input)
{
Channel<string> output = Channel.CreateBounded<string>(4);
_ = Task.Run(async () =>
{
try
{
await foreach (int value in input.ReadAllAsync())
{
await output.Writer.WriteAsync($"n={value}");
}
output.Writer.Complete();
}
catch (Exception ex)
{
// Pass the failure on. Otherwise the next stage waits forever.
output.Writer.Complete(ex);
}
});
return output.Reader;
}
}
// Output:
// n=2
// n=3
Promises the options make
Every channel between two stages needs these flags set to match the worker counts on either side. A fan-out stage with four workers writing into one channel must have SingleWriter false; a single final consumer can have SingleReader true, which lets the channel take a faster internal path. Getting them wrong is not caught by the compiler and does not throw. It corrupts the channel's internal state, which shows up later as lost or duplicated items.
using System.Threading.Channels;
// These two flags are promises about how the channel is used, not requests.
BoundedChannelOptions options = new BoundedChannelOptions(16)
{
SingleWriter = false, // several stage-2 workers write here
SingleReader = true // exactly one stage-3 consumer reads here
};
Channel<int> channel = Channel.CreateBounded<int>(options);
for (int i = 1; i <= 3; i++)
{
channel.Writer.TryWrite(i * 10);
}
channel.Writer.Complete();
int total = 0;
await foreach (int value in channel.Reader.ReadAllAsync())
{
total += value;
}
Console.WriteLine($"total: {total}");
// Output:
// total: 60
Discussion