Channels and Advanced Concurrency

A channel is an async-friendly, thread-safe queue connecting producers and consumers:

Concept

A channel is an async-friendly, thread-safe queue connecting producers and consumers:

Never blocks writers (memory can grow) Bounded buffer; FullMode decides what happens when full FullMode = Wait

Writers await WriteAsync until there is space — backpressure

FullMode = DropOldest / DropNewest / DropWrite Keep the latest data (telemetry, UI updates)

Writers: TryWrite, WriteAsync, Complete(). Readers: TryRead, ReadAsync, ReadAllAsync() (an IAsyncEnumerable<T> that ends when the writer completes). Options SingleReader/SingleWriter enable faster implementations.

Lock-Free and Atomic Techniques Atomic operations: Interlocked.CompareExchange(ref x, newValue, expected) — the building block of lock-free algorithms (compare-and-swap loops).

Memory ordering: CPUs and compilers may reorder memory operations; Volatile.Read/Write, Interlocked and lock insert the necessary barriers.

Immutable state: publish a new immutable snapshot atomically (Interlocked.Exchange(ref config, newConfig)) instead of mutating shared objects — readers always see a consistent version.

Actor-style concurrency: each "actor" owns its state and processes messages from its own channel one at a time — no locks needed.

Why It Matters

Channels are the idiomatic .NET way to build pipelines: CAN frame capture → decode → store, request queues in services, and background processing — with natural backpressure so a slow stage cannot exhaust memory.

Syntax

var ch = Channel.CreateBounded<T>(new BoundedChannelOptions(100) { FullMode = BoundedChannelFullMode.Wait });
await ch.Writer.WriteAsync(item, ct);
ch.Writer.Complete();
await foreach (var item in ch.Reader.ReadAllAsync(ct)) { ... }

Example

using System.Threading.Channels;

// Bounded channel = producer/consumer with backpressure
var channel = Channel.CreateBounded<Frame>(new BoundedChannelOptions(capacity: 4)
{
    FullMode = BoundedChannelFullMode.Wait,      // producers wait when the buffer is full (backpressure)
    SingleReader = false,
    SingleWriter = true,
});

int producerWaits = 0;
var producer = Task.Run(async () =>
{
    for (int i = 0; i < 20; i++)
    {
        var frame = new Frame(0x100 + i % 4, i);
        if (!channel.Writer.TryWrite(frame))    // buffer full -> would block
        {
            producerWaits++;
            await channel.Writer.WriteAsync(frame);
        }
    }
    channel.Writer.Complete();                   // no more items
});

// Two consumers read concurrently until the channel completes
var totals = new int[2];
var consumers = Enumerable.Range(0, 2).Select(c => Task.Run(async () =>
{
    await foreach (Frame f in channel.Reader.ReadAllAsync())
    {
        await Task.Delay(2);                     // slower than the producer
        totals[c]++;
    }
})).ToArray();

await producer;
await Task.WhenAll(consumers);
Console.WriteLine($"frames consumed: {totals.Sum()} (by 2 cons
…

Expected Result

With a capacity of 4 and slow consumers, the producer had to wait — that is backpressure. DropOldest with capacity 3 kept only the newest three values.

frames consumed: 20 (by 2 consumers), producer had to wait: True
unbounded channel accepted all: 1000
DropOldest keeps the newest: 8,9,10
lock-free max: 9999
old snapshot still consistent: Config { Timeout = 100, Retries = 3 }, new: Config { Timeout = 100, Retries = 5 }

Common Mistakes

Forgetting Writer.Complete() — consumers wait forever.

Unbounded channels between a fast producer and a slow consumer (memory grows without limit).

Hand-written lock-free code without a proven algorithm — prefer channels, concurrent collections and lock.

Mutating objects after passing them through a channel.

Best Practices

Prefer bounded channels with Wait for work queues; drop modes for "latest value wins" data.

Pass immutable messages (records) through channels.

Propagate cancellation tokens to WriteAsync/ReadAllAsync.

Use Interlocked/immutable snapshots for simple shared state; channels for workflows.

Interview Relevance

Medium–high: "channels vs BlockingCollection", "what is backpressure?", "explain compare-and-swap", "what is a memory barrier?", "actor model".

Practice Questions

Build a three-stage pipeline (read lines → parse frames → aggregate statistics) with two channels.

Implement an actor that owns a counter and processes increment/read messages from a channel.

Use DropOldest to keep the latest 10 telemetry samples for a dashboard.

Write a lock-free AtomicMax and test it with Parallel.For.

Quick Quiz

Q1. (Multiple choice) Which FullMode provides backpressure? A) DropOldest B) DropNewest C) Wait D) DropWrite Answer: C.

Q2. (Predict the output) A bounded channel of capacity 3 in DropOldest mode receives 1..10. What remains?

Answer: 8,9,10 (verified).

Q3. (True/False) ReadAllAsync() ends only when the writer calls Complete().

Answer: True. Q4. (Find the bug) A consumer loop await foreach (var f in ch.Reader.ReadAllAsync()) never finishes although all frames were written.

Answer: The producer never called ch.Writer.Complete().

Coding Challenge

Build a two-stage pipeline: stage 1 writes the numbers 1..10 to a bounded channel; stage 2 squares them into a second channel; the main method sums the results.

using System.Threading.Channels;

var numbers = Channel.CreateBounded<int>(2);
var squares = Channel.CreateBounded<int>(2);

var stage1 = Task.Run(async () =>
{
    for (int i = 1; i <= 10; i++) await numbers.Writer.WriteAsync(i);
    numbers.Writer.Complete();
});
var stage2 = Task.Run(async () =>
{
    await foreach (int n in numbers.Reader.ReadAllAsync()) await squares.Writer.WriteAsync(n * n);
    squares.Writer.Complete();
});

int sum = 0;
await foreach (int sq in squares.Reader.ReadAllAsync()) sum += sq;
await Task.WhenAll(stage1, stage2);
Console.WriteLine($"sum of squares 1..10 = {sum}");