What are System.Threading.Channels, and how would you build a bounded producer/consumer pipeline with them?

1. Why channels instead of BlockingCollection<T>

2. The efficient read loop

while (await reader.WaitToReadAsync(ct))          // false when completed and empty
{
    while (reader.TryRead(out var item))           // drain what's available without awaiting
        Process(item);
}

ReadAllAsync is fine for most code; the loop above avoids an await per item in hot paths and makes batching easy:

var batch = new List<Msg>(500);
while (await reader.WaitToReadAsync(ct))
{
    while (batch.Count < 500 && reader.TryRead(out var m)) batch.Add(m);
    await InsertBatchAsync(batch, ct);
    batch.Clear();
}

3. Fan-out / fan-in with completion propagation

static Task RunStageAsync<TIn, TOut>(ChannelReader<TIn> input, ChannelWriter<TOut> output,
    int workers, Func<TIn, CancellationToken, ValueTask<TOut>> transform, CancellationToken ct)
{
    var tasks = Enumerable.Range(0, workers).Select(async _ =>
    {
        await foreach (var item in input.ReadAllAsync(ct))
            await output.WriteAsync(await transform(item, ct), ct);
    }).ToArray();

    return Task.WhenAll(tasks).ContinueWith(
        t => output.TryComplete(t.Exception?.GetBaseException()),
        CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
}

If any worker fails, the next stage's writer completes with the exception, its readers' Completion faults, and the failure surfaces at the end of the pipeline instead of everything hanging.

4. Preserving order per key

var partitions = Enumerable.Range(0, 8).Select(_ => Channel.CreateBounded<Evt>(1000)).ToArray();
int Partition(string key) => (int)((uint)key.GetHashCode() % (uint)partitions.Length);
await partitions[Partition(evt.AccountId)].Writer.WriteAsync(evt, ct);
// one consumer per partition -> per-account ordering preserved, 8-way parallelism overall

(Use a stable hash such as XxHash32 if partitioning must be consistent across processes.)

5. Internals in brief

• Bounded channels use a lock plus a queue of items and queues of blocked readers/writers.
• Waiters are represented by pooled IValueTaskSource objects, so waiting doesn't allocate a Task.
• Single-reader/single-writer unbounded channels can use lock-free paths.

6. Version notes

7. Try it yourself

1. Run the code example; confirm the sum and that all stages complete.
2. Make one parser throw on a specific input and confirm the final await foreach throws instead of hanging.
3. Bounded capacity 5, fast producer, slow consumer (100 ms/item): log when the producer waits.
4. Switch to DropOldest with an itemDropped callback and count drops.
5. Add batching with the WaitToReadAsync / TryRead loop and measure DB round trips.
6. Partition by key and verify per-key ordering with multiple consumers.

using System.Threading.Channels;

var raw = Channel.CreateBounded<string>(new BoundedChannelOptions(100)
{
    FullMode = BoundedChannelFullMode.Wait,
    SingleWriter = true
});
var parsed = Channel.CreateBounded<int>(new BoundedChannelOptions(100) { SingleReader = true });

// Stage 1: producer
var producer = Task.Run(async () =>
{
    for (int i = 0; i < 1_000; i++)
        await raw.Writer.WriteAsync(i.ToString()); // waits when the channel is full
    raw.Writer.Complete();
});

// Stage 2: four parsers in parallel
var parsers = Enumerable.Range(0, 4).Select(_ => Task.Run(async () =>
{
    await foreach (var line in raw.Reader.ReadAllAsync())
        await parsed.Writer.WriteAsync(int.Parse(line));
})).ToArray();

// Complete stage 3's input when all parsers finish (propagating failure if any).
_ = Task.WhenAll(parsers).ContinueWith(
    t => parsed.Writer.Complete(t.Exception),
    TaskScheduler.Default);

// Stage 3: single aggregator
long sum = 0;
await foreach (var n in parsed.Reader.ReadAllAsync())
    sum += n;

await producer;
Console.WriteLine($"Sum {sum}"); // 499500