Channels & Async Streams

Key Points

• Channel<T> (System.Threading.Channels) is the modern allocation-free, async-first producer/consumer queue. Replaces BlockingCollection<T> and BufferBlock<T> for new code.
• Bounded vs unbounded: bounded applies backpressure on the writer; unbounded never blocks writes (and grows indefinitely if the reader can't keep up).
• FullMode controls bounded behavior when full: Wait (default), DropOldest, DropNewest, DropWrite. Each is explicit — no silent buffering.
• SingleReader / SingleWriter are massive perf flags. If true, the implementation skips locks/interlocked ops on that side. Violating the constraint corrupts state.
• IAsyncEnumerable<T> is the language's answer to "async sequences". await foreach consumes them; yield return produces them.
• [EnumeratorCancellation] + WithCancellation() is how cancellation tokens flow into IAsyncEnumerable.
• Backpressure pattern: producer awaits WriteAsync on a bounded channel — natural backpressure when the consumer is slow.

Concepts (deep dive)

Channels — the shape

                     ┌─────────────────────────┐
                     │       Channel<T>         │
                     │                          │
   Producer  ──────► │  ChannelWriter<T>        │
   (multiple ok)     │   - WriteAsync           │
                     │   - TryWrite             │
                     │   - Complete             │
                     └─────────┬────────────────┘
                               │
                               ▼ (queue, bounded or unbounded)
                     ┌─────────────────────────┐
                     │  ChannelReader<T>        │ ◄────── Consumer
                     │   - ReadAsync             │         (multiple ok)
                     │   - TryRead              │
                     │   - WaitToReadAsync       │
                     │   - ReadAllAsync          │
                     │   - Completion (Task)     │
                     └─────────────────────────┘

Create:

// Unbounded: writers never wait, channel grows freely
Channel<Item> ch = Channel.CreateUnbounded<Item>();

// Bounded with backpressure: writer awaits when full
var ch2 = Channel.CreateBounded<Item>(new BoundedChannelOptions(capacity: 1024)
{
    FullMode = BoundedChannelFullMode.Wait,
    SingleReader = true,
    SingleWriter = false
});

FullMode semantics

When a bounded channel is full and a writer arrives:

var ch = Channel.CreateBounded<Sample>(new BoundedChannelOptions(100)
{
    FullMode = BoundedChannelFullMode.DropOldest   // sliding window of latest 100 samples
});

⚠️ Senior trap: Drop* modes silently lose data. Track drop counts (e.g., Interlocked.Increment on a counter) so you can alert when loss is occurring.

Reader patterns

// 1. ReadAllAsync — the modern idiom
await foreach (var item in ch.Reader.ReadAllAsync(ct))
    Process(item);

// 2. WaitToReadAsync + TryRead — when you need batching or fine-grained control
while (await ch.Reader.WaitToReadAsync(ct))
{
    while (ch.Reader.TryRead(out var item))
        Process(item);
}

ReadAllAsync is the cleanest for "process each item one at a time". WaitToReadAsync + TryRead lets you drain in batches — efficient when processing is faster done in bulk.

Producer completion

public async Task ProduceAsync(ChannelWriter<Item> writer, CancellationToken ct)
{
    try
    {
        foreach (var item in await SourceAsync(ct))
            await writer.WriteAsync(item, ct);
    }
    catch (Exception ex)
    {
        writer.Complete(ex);   // signals failure to readers
        throw;
    }
    writer.Complete();         // normal completion
}

Complete() signals "no more items". Readers see WaitToReadAsync return false (or ReadAllAsync exit cleanly). Complete(exception) signals failure — readers' WaitToReadAsync throws the exception, propagating it through.

writer.TryComplete() is the "complete only if not already" variant — useful in concurrent producers where any might finish first.

Single-reader / single-writer optimizations

var ch = Channel.CreateBounded<int>(new BoundedChannelOptions(100)
{
    SingleReader = true,
    SingleWriter = true
});

When set, the channel uses lock-free fast paths that assume only one thread enters each side at a time. 2-3x throughput improvement for the right workloads.

⚠️ The runtime does not enforce the constraint. Two readers on a SingleReader=true channel will silently corrupt state — race conditions on internal fields. If you can't guarantee single-threaded access on a side, leave it false.

Idiomatic pipeline

public async Task RunPipelineAsync(CancellationToken ct)
{
    var ch = Channel.CreateBounded<RawItem>(new(1024) { SingleWriter = true });
    var enriched = Channel.CreateBounded<EnrichedItem>(new(512) { SingleReader = true });

    var producer = Task.Run(async () =>
    {
        try
        {
            await foreach (var raw in _source.ReadAsync(ct))
                await ch.Writer.WriteAsync(raw, ct);
        }
        finally { ch.Writer.Complete(); }
    });

    var enricher = Task.Run(async () =>
    {
        try
        {
            await foreach (var raw in ch.Reader.ReadAllAsync(ct))
            {
                var e = await EnrichAsync(raw, ct);
                await enriched.Writer.WriteAsync(e, ct);
            }
        }
        finally { enriched.Writer.Complete(); }
    });

    var consumer = Task.Run(async () =>
    {
        await foreach (var e in enriched.Reader.ReadAllAsync(ct))
            await _sink.WriteAsync(e, ct);
    });

    await Task.WhenAll(producer, enricher, consumer);
}

Each stage has its own bounded channel — backpressure propagates upstream automatically. The slowest stage rate-limits the pipeline.

IAsyncEnumerable<T> — the language feature

public async IAsyncEnumerable<int> RangeAsync(
    int start, int count,
    [EnumeratorCancellation] CancellationToken ct = default)
{
    for (int i = 0; i < count; i++)
    {
        ct.ThrowIfCancellationRequested();
        await Task.Delay(100, ct);
        yield return start + i;
    }
}

// Consumer:
await foreach (var x in svc.RangeAsync(0, 10).WithCancellation(extCt))
    Console.WriteLine(x);

Compiler generates a state machine implementing IAsyncEnumerable<T> and IAsyncEnumerator<T>. Each yield return yields and suspends; each MoveNextAsync resumes.

[EnumeratorCancellation] is the magic that lets WithCancellation(token) flow into the iterator's ct parameter. Without the attribute, WithCancellation is silently a no-op.

IAsyncEnumerable performance

Each MoveNextAsync call returns a ValueTask<bool>. If the next item is available synchronously (e.g., already buffered), no allocation. If async, the runtime allocates a Task<bool> for the continuation.

In a tight pipeline, that allocation per item adds up. Mitigation:

• Batch: IAsyncEnumerable<T[]> instead of IAsyncEnumerable<T> — amortize the await cost.
• Skip iterator for sync-fast paths: if all items are buffered, a sync IEnumerable<T> is cheaper.

ConfigureAwait on IAsyncEnumerable

await foreach (var x in source.ConfigureAwait(false))
    Process(x);

Configures every internal MoveNextAsync and DisposeAsync await. Library code: yes. Application code in ASP.NET Core: rarely needed (no sync context). See Async/Await — SyncContext & Deadlocks.

Channels vs BlockingCollection<T>

How it works under the hood

Channel<T> is in System.Threading.Channels. Source: runtime/src/libraries/System.Threading.Channels. Implementations:

• Unbounded — backed by a ConcurrentQueue<T>. WriteAsync always completes synchronously.
• Bounded — backed by a fixed-capacity ring buffer (or single-segment for SingleReader/SingleWriter cases). WriteAsync awaits when full.

Both expose WaitToReadAsync returning ValueTask<bool>. The ValueTask is backed by an IValueTaskSource<bool> instance reused per channel — zero allocation per await on the hot path when the channel has items.

IAsyncEnumerable<T> is a Roslyn-generated state machine. The generated class implements: - IAsyncEnumerable<T>.GetAsyncEnumerator(CancellationToken) — returns the state machine itself. - IAsyncEnumerator<T>.MoveNextAsync() — returns ValueTask<bool> driving the state machine. - IAsyncEnumerator<T>.DisposeAsync() — cleanup.

WithCancellation(token) returns a wrapping struct that, on GetAsyncEnumerator, links the token. Inside the iterator, [EnumeratorCancellation] tells the compiler "merge the public CancellationToken parameter with the linked one passed by the consumer."

Code: correct vs wrong

❌ Wrong: forgetting [EnumeratorCancellation]

public async IAsyncEnumerable<int> ReadAsync(CancellationToken ct)
{
    while (!ct.IsCancellationRequested) yield return 1;
}

// Consumer:
await foreach (var x in svc.ReadAsync().WithCancellation(extCt))
{ /* ❌ extCt isn't linked; ct in iterator is default(CancellationToken) */ }

✅ Correct

public async IAsyncEnumerable<int> ReadAsync(
    [EnumeratorCancellation] CancellationToken ct = default)
{
    while (!ct.IsCancellationRequested) yield return 1;
}

❌ Wrong: producer doesn't complete the channel

public async Task ProduceAsync(ChannelWriter<Item> w, CancellationToken ct)
{
    foreach (var item in await SourceAsync(ct))
        await w.WriteAsync(item, ct);
    // ❌ No w.Complete() — readers wait forever
}

✅ Correct

public async Task ProduceAsync(ChannelWriter<Item> w, CancellationToken ct)
{
    try
    {
        foreach (var item in await SourceAsync(ct))
            await w.WriteAsync(item, ct);
    }
    finally { w.Complete(); }
}

❌ Wrong: SingleReader violation

var ch = Channel.CreateBounded<int>(new(100) { SingleReader = true });
_ = Task.Run(async () => { await foreach (var x in ch.Reader.ReadAllAsync()) Process(x); });
_ = Task.Run(async () => { await foreach (var x in ch.Reader.ReadAllAsync()) Process(x); });
// ❌ Two readers on SingleReader=true → silent corruption

✅ Correct: leave it false, OR use one reader fan-outing

var ch = Channel.CreateBounded<int>(new(100) { SingleReader = false });
// or single reader that hands off:
var ch2 = Channel.CreateBounded<int>(new(100) { SingleReader = true });
_ = Task.Run(async () =>
{
    await foreach (var x in ch2.Reader.ReadAllAsync())
        _ = Task.Run(() => Process(x));
});

❌ Wrong: drop modes without observability

var ch = Channel.CreateBounded<Sample>(new(100) { FullMode = BoundedChannelFullMode.DropWrite });
// Production: drops happen silently; nobody knows.

✅ Correct: track drops

private long _dropped;
private void Write(Sample s)
{
    if (!ch.Writer.TryWrite(s))
        Interlocked.Increment(ref _dropped);
}
// Expose _dropped via metrics.

Design patterns for this topic

Pattern 1 — "Bounded channel + backpressure"

• Intent: producer naturally slows when consumer is slow.
• When to use it: any pipeline where the consumer can't keep up sometimes.
• Code sketch: see "idiomatic pipeline" above.

Pattern 2 — "Drop-oldest for sensor/metric streams"

• Intent: consumers care about recent data; old samples are useless.
• Code sketch:

var ch = Channel.CreateBounded<Sample>(new(100) {
    FullMode = BoundedChannelFullMode.DropOldest
});

• Trade-offs: quietly loses old data; track drop count.

Pattern 3 — "Async iterator over a paged source"

• Intent: stream paged results lazily.
• Code sketch:

public async IAsyncEnumerable<Item> AllItemsAsync(
    [EnumeratorCancellation] CancellationToken ct = default)
{
    string? continuationToken = null;
    do
    {
        var page = await _api.GetPageAsync(continuationToken, ct);
        foreach (var item in page.Items) yield return item;
        continuationToken = page.NextToken;
    } while (continuationToken != null);
}

Pattern 4 — "Bounded buffer + Drain"

• Intent: finish in-flight work cleanly on shutdown.
• Code sketch:

public async Task ShutdownAsync()
{
    _writer.Complete();
    await _consumerTask;   // consumer drains remaining items, then exits
}

Pattern 5 — "Batch for throughput"

• Intent: amortize per-item overhead.
• Code sketch:

const int BatchSize = 100;
var batch = new List<Item>(BatchSize);
await foreach (var item in ch.Reader.ReadAllAsync(ct))
{
    batch.Add(item);
    if (batch.Count >= BatchSize)
    {
        await ProcessBatchAsync(batch, ct);
        batch.Clear();
    }
}
if (batch.Count > 0) await ProcessBatchAsync(batch, ct);

Pros & cons / trade-offs

When to use / when to avoid

• Use Channel<T> for in-process producer/consumer: pipelines, work queues, message processing.
• Use bounded channels by default — gives you backpressure for free. Reach for unbounded only when you can prove writers are inherently slower.
• Use IAsyncEnumerable<T> for paged or streaming results that can be consumed lazily.
• Avoid BlockingCollection<T> in new code. Channels are better in every dimension.
• Avoid drop modes without metrics — silent loss is a production debugging nightmare.
• Avoid SingleReader=true if there's any chance of multiple readers.

Interview Q&A

Q1. What's Channel<T> and how does it differ from BlockingCollection<T>? Channel<T> is an async-first producer/consumer queue introduced in .NET Core 2.1 / .NET Standard 2.1. It avoids thread blocking and allocations on the hot path (via ValueTask<bool> and IValueTaskSource). BlockingCollection<T> is older, thread-blocking, and allocation-heavy.

Q2. What's the difference between bounded and unbounded channels? Unbounded: writers never wait; channel grows. Bounded: writers either wait (Wait mode) or drop on full. Bounded channels enable backpressure.

Q3. What does FullMode = DropOldest mean? When the bounded channel is full and a writer arrives, the head (oldest) item is discarded to make room. Useful for sliding windows of recent data.

Q4. Why are SingleReader=true / SingleWriter=true so much faster? The implementation skips locking on that side. CAS-based or even simple writes — assumes only one thread enters at a time. The constraint is a contract; violating it silently corrupts state.

Q5. How do you signal "no more items" to readers? writer.Complete(). Readers' WaitToReadAsync returns false; ReadAllAsync exits cleanly. Complete(exception) signals fault — readers throw the exception.

Q6. What's [EnumeratorCancellation] for? The attribute on the CancellationToken parameter of an async IAsyncEnumerable<T> iterator. The compiler links the parameter to the token threaded by WithCancellation() at the consumer. Without it, WithCancellation is silently a no-op.

Q7. Why might IAsyncEnumerable<T> allocate per item? Each MoveNextAsync returns ValueTask<bool>. If the next item is available synchronously, no allocation. If async, the runtime may allocate a Task<bool> for the continuation. Batching (IAsyncEnumerable<T[]>) amortizes this.

Q8. How do you implement a producer that stops when the consumer can't keep up? Bounded channel with FullMode = Wait. WriteAsync awaits when full — natural backpressure.

Q9. How do you process items in bulk from a channel? await foreach with a buffer + flush logic, or WaitToReadAsync + a tight TryRead drain. Both achieve "wait for something, then drain everything available, then process the batch".

Q10. What happens if the consumer throws while reading? The exception propagates out of ReadAllAsync / await foreach. The channel itself remains open; if you want to signal upstream, the consumer should call writer.Complete(ex) before unwinding.

Q11. Compare Channel<T> to TPL Dataflow's BufferBlock<T>. Channel<T> is lighter, async-first, and the modern recommendation. Dataflow has richer composition (linking, propagation), but for simple producer/consumer needs, channels win on simplicity and perf.

Q12. Why are channel writes potentially zero-allocation? The bounded ring buffer plus a reused IValueTaskSource<bool> per channel. WriteAsync reuses the source for await chaining; only when waiters spike beyond a small pool does it fall back to allocating.

Q13. When would you reach for Parallel.ForEachAsync over channels? When work is independent and you want N concurrent operations against a fixed degree-of-parallelism. Parallel.ForEachAsync is simpler than channel-based fan-out for the typical "process N items in parallel" case. Channels shine for true pipelines with multiple stages.

Gotchas / common mistakes

• ⚠️ Forgetting [EnumeratorCancellation] — WithCancellation silently does nothing.
• ⚠️ Forgetting writer.Complete() — readers wait forever.
• ⚠️ SingleReader/Writer violations — silent corruption.
• ⚠️ Unbounded channels in production — easy to leak memory.
• ⚠️ Drop modes without metrics — silent data loss.
• ⚠️ Per-item allocation in tight pipelines — batch instead.
• ⚠️ await foreach without ConfigureAwait(false) in libraries — context capture overhead.
• ⚠️ yield return inside a lock — deadlock risk; the iterator suspends with the lock held.

Further reading

• Stephen Toub: An Introduction to System.Threading.Channels
• Channel<T>
• IAsyncEnumerable<T>
• [EnumeratorCancellation]