Build a Bounded Processing Pipeline With System.Threading.Channels

A bounded C# pipeline with Channels connects producers and consumers through an in-memory queue with a fixed positive capacity. When that queue is full, a lossless design waits for space instead of dropping accepted work. The interesting part is not creating Channel<T>. The interesting part is defining who owns messages, completion, cancellation, draining, failures, and every task.

This article builds one complete producer/consumer pipeline with System.Threading.Channels. It uses BoundedChannelFullMode.Wait, one producer, several consumers, one completion coordinator, and explicit cleanup for resource-owning messages. The example is intentionally in-process. It does not claim durable delivery, exactly-once processing, or universal throughput gains.

The goal is a lifecycle you can explain from admission to shutdown. Once those semantics are visible, the code becomes much easier to reason about.

A Bounded C# Pipeline Starts With an Explicit Capacity

Microsoft describes Channels as asynchronous producer/consumer synchronization structures built around a FIFO queue. A channel exposes a ChannelWriter<T> to producers and a ChannelReader<T> to consumers.

A bounded channel adds a maximum number of queued items, and the bounded channel factory requires positive capacity. That sounds like a small configuration detail, but it changes the admission contract:

• Below capacity, WriteAsync can admit another item.
• At capacity, the configured bounded full mode determines whether writing waits or an item is dropped.
• When a consumer removes an item, a waiting writer can continue.
• Capacity counts queued items, not items currently being processed by consumers.

Capacity therefore limits the buffer between producer and consumer work. It does not directly limit the total number of messages in the application. A producer may own one item that is waiting to be admitted, the channel may hold up to its capacity, and each consumer may own one item in active processing.

There is no universally correct capacity. It depends on item size, acceptable memory retention, burst tolerance, producer behavior, and how quickly consumers release capacity. Choose a positive value as an explicit operational limit, then validate it under the workload that matters to the application.

The total in-flight ownership is larger than the configured capacity. With capacity 2 and three consumers, the channel can hold two queued messages while three more are actively owned by workers. The producer may also hold one message whose WriteAsync is waiting. That makes capacity a queue bound, not a complete memory formula.

Resource size matters as much as item count. Two small identifiers and two pooled megabyte buffers consume very different amounts of memory even though both occupy two channel slots. If message sizes vary significantly, validate retained memory under realistic distributions and consider whether large payloads should be represented by owned handles rather than copied objects.

Worker count is a separate limit. Increasing consumers can increase the number of messages in processing without changing the number buffered in the channel. That may be appropriate for independent work, but it also changes downstream concurrency, external-service pressure, and processing-completion order. Capacity and worker count should therefore be configured and reasoned about independently.

Wait Mode Provides Lossless In-Memory Backpressure

BoundedChannelFullMode.Wait tells the writer to wait when the channel is full. WriteAsync completes after the item is admitted or the operation ends through cancellation or channel completion, while TryWrite reports immediately whether it could write. Wait does not intentionally discard an item to make room.

That is the lossless part of this example: an item is either successfully written or the producer receives an exception or cancellation before ownership transfers. It does not mean the whole application can never lose work. A process crash still destroys in-memory state, and a consumer can fail after performing only part of its work.

Waiting is also backpressure, not parallelism. The producer yields while capacity is unavailable. Consumers free capacity by reading. The relationship is asynchronous, so a waiting writer does not need to block a thread.

If async and await are still new territory, my async and await beginner tips provide prerequisite context. In this pipeline, every channel operation and processing delay is awaited, and every cancellable operation receives the pipeline token.

Define Producer, Channel, Consumer, and Message Ownership

Ownership should move in one direction:

This contract prevents double disposal and abandoned resources. It also gives a precise meaning to a successful write: the producer no longer owns that message.

The example uses a pooled byte buffer so cleanup is visible in code. That is not a requirement for Channels. A small immutable record containing only strings and numbers may need no disposal. The same ownership transfer still applies, but cleanup is then logical rather than resource-based.

The channel itself is owned by the method that creates and supervises it. Producers receive only the writer. Consumers receive only the reader. Neither side decides independently that the entire pipeline has completed.

Complete Bounded Channel Pipeline Example

The sample targets .NET 10 and C# 14 with nullable reference types enabled. System.Threading.Channels is part of the shared framework, so no external package is required.

<Project Sdk="Microsoft.NET.Sdk">
  <PropertyGroup>
    <OutputType>Exe</OutputType>
    <TargetFramework>net10.0</TargetFramework>
    <LangVersion>14.0</LangVersion>
    <Nullable>enable</Nullable>
    <ImplicitUsings>enable</ImplicitUsings>
    <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
  </PropertyGroup>
</Project>

The following Program.cs creates one producer and three consumers. Set failOnSequence to a message sequence if you want to exercise the worker-fault path.

using System.Buffers;
using System.Collections.Concurrent;
using System.Runtime.ExceptionServices;
using System.Text;
using System.Threading.Channels;

namespace BoundedChannelsPipeline;

public sealed record InputMessage(long Sequence, string Text);

public sealed class WorkItem : IDisposable
{
    private static int _activeCount;

    private readonly int _payloadLength;
    private IMemoryOwner<byte>? _payloadOwner;

    private WorkItem(
        long sequence,
        IMemoryOwner<byte> payloadOwner,
        int payloadLength)
    {
        Sequence = sequence;
        _payloadOwner = payloadOwner;
        _payloadLength = payloadLength;
        Interlocked.Increment(ref _activeCount);
    }

    public static int ActiveCount => Volatile.Read(ref _activeCount);

    public long Sequence { get; }

    public ReadOnlyMemory<byte> Payload
    {
        get
        {
            IMemoryOwner<byte> owner = _payloadOwner ??
                throw new ObjectDisposedException(nameof(WorkItem));
            return owner.Memory[.._payloadLength];
        }
    }

    public static WorkItem Create(InputMessage input)
    {
        int byteCount = Encoding.UTF8.GetByteCount(input.Text);
        IMemoryOwner<byte> owner = MemoryPool<byte>.Shared.Rent(byteCount);

        try
        {
            int bytesWritten = Encoding.UTF8.GetBytes(
                input.Text.AsSpan(),
                owner.Memory.Span);
            return new WorkItem(input.Sequence, owner, bytesWritten);
        }
        catch
        {
            owner.Dispose();
            throw;
        }
    }

    public void Dispose()
    {
        IMemoryOwner<byte>? owner =
            Interlocked.Exchange(ref _payloadOwner, null);

        if (owner is not null)
        {
            owner.Dispose();
            Interlocked.Decrement(ref _activeCount);
        }
    }
}

public sealed record PipelineRunResult(
    IReadOnlyList<long> ProcessedSequences);

public static class Program
{
    public static async Task Main()
    {
        InputMessage[] input =
        [
            new(0, "alpha"),
            new(1, "bravo"),
            new(2, "charlie"),
            new(3, "delta"),
            new(4, "echo"),
            new(5, "foxtrot")
        ];

        PipelineRunResult result = await RunPipelineAsync(
            input,
            capacity: 2,
            workerCount: 3,
            failOnSequence: null,
            CancellationToken.None);

        Console.WriteLine(
            $"Processed {result.ProcessedSequences.Count} messages.");
    }

    public static async Task<PipelineRunResult> RunPipelineAsync(
        IReadOnlyList<InputMessage> input,
        int capacity,
        int workerCount,
        long? failOnSequence,
        CancellationToken cancellationToken)
    {
        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(capacity);
        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(workerCount);

        Channel<WorkItem> channel = Channel.CreateBounded<WorkItem>(
            new BoundedChannelOptions(capacity)
            {
                FullMode = BoundedChannelFullMode.Wait,
                SingleWriter = true,
                SingleReader = workerCount == 1,
                AllowSynchronousContinuations = false
            });

        using CancellationTokenSource abort =
            CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);

        var consumerFailure = new TaskCompletionSource<Exception>(
            TaskCreationOptions.RunContinuationsAsynchronously);

        var processedSequences = new ConcurrentQueue<long>();

        Task producer = ProduceAsync(
            input,
            channel.Writer,
            abort.Token);

        var consumers = new Task[workerCount];
        for (int workerId = 0; workerId < workerCount; workerId++)
        {
            consumers[workerId] = ConsumeAsync(
                workerId,
                channel.Reader,
                consumerFailure,
                processedSequences,
                failOnSequence,
                abort.Token);
        }

        Task[] allTasks = [producer, .. consumers];
        bool writerCompleted = false;

        try
        {
            Task firstOutcome = await Task.WhenAny(
                producer,
                consumerFailure.Task);

            if (ReferenceEquals(firstOutcome, consumerFailure.Task))
            {
                Exception consumerError = await consumerFailure.Task;
                ExceptionDispatchInfo.Capture(consumerError).Throw();
                throw new InvalidOperationException("Unreachable.");
            }

            await producer;
            channel.Writer.Complete();
            writerCompleted = true;

            Task allConsumers = Task.WhenAll(consumers);
            Task drainOutcome = await Task.WhenAny(
                allConsumers,
                consumerFailure.Task);

            if (ReferenceEquals(drainOutcome, consumerFailure.Task))
            {
                Exception consumerError = await consumerFailure.Task;
                ExceptionDispatchInfo.Capture(consumerError).Throw();
                throw new InvalidOperationException("Unreachable.");
            }

            await allConsumers;
            await channel.Reader.Completion;

            return new PipelineRunResult(
                processedSequences.ToArray());
        }
        catch (Exception ex)
        {
            await abort.CancelAsync();

            if (!writerCompleted)
            {
                channel.Writer.Complete(ex);
            }

            await ObserveAllAsync(allTasks);
            DisposeBufferedMessages(channel.Reader);
            await ObserveCompletionAsync(channel.Reader.Completion);
            throw;
        }
    }

    private static async Task ProduceAsync(
        IEnumerable<InputMessage> input,
        ChannelWriter<WorkItem> writer,
        CancellationToken cancellationToken)
    {
        foreach (InputMessage source in input)
        {
            WorkItem item = WorkItem.Create(source);
            bool ownershipTransferred = false;

            try
            {
                await writer.WriteAsync(item, cancellationToken);
                ownershipTransferred = true;
            }
            finally
            {
                if (!ownershipTransferred)
                {
                    item.Dispose();
                }
            }
        }
    }

    private static async Task ConsumeAsync(
        int workerId,
        ChannelReader<WorkItem> reader,
        TaskCompletionSource<Exception> failureSignal,
        ConcurrentQueue<long> processedSequences,
        long? failOnSequence,
        CancellationToken cancellationToken)
    {
        try
        {
            await foreach (WorkItem item in
                reader.ReadAllAsync(cancellationToken))
            {
                using (item)
                {
                    await ProcessAsync(
                        workerId,
                        item,
                        processedSequences,
                        failOnSequence,
                        cancellationToken);
                }
            }
        }
        catch (Exception ex)
        {
            failureSignal.TrySetResult(ex);
            throw;
        }
    }

    private static async Task ProcessAsync(
        int workerId,
        WorkItem item,
        ConcurrentQueue<long> processedSequences,
        long? failOnSequence,
        CancellationToken cancellationToken)
    {
        int delayMilliseconds = (int)(item.Sequence % 3) switch
        {
            0 => 45,
            1 => 15,
            _ => 30
        };

        await Task.Delay(
            TimeSpan.FromMilliseconds(delayMilliseconds),
            cancellationToken);

        if (item.Sequence == failOnSequence)
        {
            throw new InvalidOperationException(
                $"Processing failed for sequence {item.Sequence}.");
        }

        string text = Encoding.UTF8.GetString(item.Payload.Span);
        processedSequences.Enqueue(item.Sequence);
        Console.WriteLine(
            $"Worker {workerId} completed {item.Sequence}: {text}");
    }

    private static async Task ObserveAllAsync(
        IReadOnlyCollection<Task> tasks)
    {
        try
        {
            await Task.WhenAll(tasks);
        }
        catch
        {
            foreach (Task task in tasks)
            {
                _ = task.Exception;
            }
        }
    }

    private static void DisposeBufferedMessages(
        ChannelReader<WorkItem> reader)
    {
        while (reader.TryRead(out WorkItem? item))
        {
            item.Dispose();
        }
    }

    private static async Task ObserveCompletionAsync(Task completion)
    {
        await completion.ConfigureAwait(
            ConfigureAwaitOptions.SuppressThrowing);
    }
}

The local lifecycle validation artifact records four executed checks against this sample: normal drain, original consumer-fault surfacing, cancellation of a producer blocked by capacity, and zero active resource-owning WorkItem instances after cleanup. These are intended guarantees of this sample implementation, not guarantees supplied automatically by every Channels design.

One possible run shows why FIFO admission and processing completion are separate guarantees. The exact worker assignment and completion order can vary:

Worker 1 completed 1: bravo
Worker 2 completed 2: charlie
Worker 0 completed 0: alpha
Worker 2 completed 4: echo
Worker 1 completed 3: delta
Worker 0 completed 5: foxtrot

Every task in this program has a named owner. RunPipelineAsync stores the producer and consumer tasks, awaits them during normal completion, and observes them during an abort. There are no discarded worker tasks and no fire-and-forget loops.

SingleReader and SingleWriter Must Tell the Truth

SingleWriter and SingleReader are promises supplied by the channel creator. They are not worker-count settings. They tell the implementation whether concurrent write or read operations can occur over the channel's lifetime.

The example has exactly one producer task, so SingleWriter = true is honest. The coordinator calls Complete, but completion is not a concurrent message write. If the design later adds several producers that call WriteAsync, the flag must become false unless those producers are serialized behind one writer owner.

The example has a configurable consumer count. With three consumers, several reads can be pending, so SingleReader = false. If workerCount is one, setting it to true becomes honest.

These options do not create workers, schedule work, or guarantee ordering. They describe access that the surrounding code already established.

AllowSynchronousContinuations remains false; enabling it can allow an operation that completes pending async work to run the continuation inline. That may be useful in a measured design, but it also couples producer execution to consumer continuation behavior. The conservative setting keeps that coupling out of this educational example.

One Completion Coordinator Owns the Writer

Only RunPipelineAsync completes the writer. The producer writes messages but never calls Complete, whose contract announces that no more data will be written. Consumers read messages but never complete the writer. This prevents several participants from racing to announce the final channel state.

The normal sequence is:

1. Start the producer and all consumers.
2. Wait for either producer completion or an early consumer failure.
3. Await the producer so its exception or cancellation is observed.
4. Complete the writer once, because no more messages can be admitted.
5. Let consumers read everything already buffered.
6. Await every consumer and then the reader's completion task.

If there were multiple producers, the same rule would apply. The coordinator would await all producer tasks and complete the writer only after every producer finished successfully. Individual producers would still not compete to complete the channel.

The completion coordinator is not merely a convenient place to put Complete. It is the component that knows whether all writers are finished, whether the pipeline should drain, and which exception should represent an abnormal end.

Cancellation Stops Pending Operations, Not the Channel

Channel reads and writes accept cancellation tokens. Canceling a pending WriteAsync stops that admission attempt. Canceling ReadAllAsync stops that consumer's enumeration, although immediately available data may still be yielded after cancellation has been requested.

Neither operation cancellation invokes ChannelWriter.Complete or independently declares that no producer will ever write again. Cancellation and channel completion answer different questions:

• Cancellation asks a pending operation to stop waiting or processing.
• Writer completion states that no more items will be written.

The sample connects them through policy rather than confusing them as the same mechanism. A producer failure, consumer failure, or caller cancellation reaches the coordinator's catch block. The coordinator cancels the linked abort token to unblock sibling operations. If the writer is still open, the coordinator then completes it with the triggering exception.

Cancellation remains cooperative under the .NET cancellation model. ProcessAsync, WriteAsync, and ReadAllAsync all receive the linked token. A real stage must pass that token to every cancellable dependency it owns. Cancellation does not undo a message that already finished processing or reverse an external side effect.

FIFO Admission Does Not Guarantee FIFO Processing Completion

The Channels FIFO queue contract preserves item admission and removal order at the channel boundary. With the single producer in this example, message sequence also describes admission order, and consumers remove earlier queued items before later queued items.

Three consumers can still finish in another order because the FIFO channel contract governs queue removal, not completion of processing performed after a read. Once consumer 0 reads sequence 0 and consumer 1 reads sequence 1, those messages are no longer in the channel. Their processing durations are independent. Sequence 1 may finish before sequence 0 even though it was admitted later.

This distinction matters because "ordered" can refer to several contracts:

• Producer call order.
• Successful admission order.
• Consumer read order.
• Processing completion order.
• Result publication order.

The sample guarantees FIFO admission from its one producer and FIFO removal from the channel. It does not reorder completed work. If final output must follow sequence order, add a deliberate reorder stage keyed by sequence or use one consumer for the order-sensitive stage. Do not assume the channel will serialize work that happens after a read.

Normal Completion Drains Accepted Work

The sample intends normal shutdown to drain accepted work rather than cancel consumers, and the local normal-drain test verifies that behavior. After the producer finishes, the coordinator calls channel.Writer.Complete(). That closes admission while preserving buffered items.

ReadAllAsync continues yielding available items until no more data can become available. Each consumer disposes its current WorkItem after ProcessAsync returns. When the buffer is empty and no future writes are possible, the enumerations finish. The coordinator awaits Task.WhenAll(consumers) and then channel.Reader.Completion.

That is a drain:

• No new items are accepted after writer completion.
• Every successfully admitted item remains available to consumers.
• The method does not return until every consumer task ends.
• Reader completion confirms that no additional data can ever be read.

Drain time is workload-dependent. A host can place a separate time budget around RunPipelineAsync and cancel when that budget expires, but that changes the path from graceful drain to abort. The application should name that distinction instead of calling every shutdown "graceful."

Fault and Abort Paths Need Cleanup Too

A consumer can fail while the producer is blocked on a full channel. Without supervision, the consumer task faults, the producer keeps waiting, and other workers may continue without anyone closing the lifecycle.

The sample intends the consumerFailure signal to let the coordinator observe the first consumer exception promptly, and the local fault and blocked-producer tests verify the exercised failure paths. On any producer or consumer failure, the catch path:

1. Cancels the linked token so pending writes, reads, and processing can stop.
2. Completes the writer with the original exception if it is still open.
3. Awaits or observes the producer and every consumer task.
4. Drains any messages left in the channel after workers have stopped.
5. Disposes each remaining resource-owning message.
6. Observes the reader completion task.
7. Rethrows the triggering exception.

The order is intentional. Buffered messages cannot be disposed safely while a consumer might still read them. The coordinator first waits for workers to stop using the reader, then takes ownership of what remains.

If messages do not own disposable resources, the cleanup loop may have nothing material to release. Keeping the ownership rule explicit still prevents a future change from adding pooled buffers, streams, or leases without an abort cleanup plan.

Know What This Bounded Channel Pipeline Does Not Promise

This design is lossless only at the bounded in-memory admission boundary while the process is running. Wait avoids intentional discarding. It does not turn Channels into a durable broker.

The example does not promise:

• Recovery after process or machine failure.
• Persistence across application restarts.
• Exactly-once side effects.
• Automatic retries.
• Transactional coordination with a database or remote service.
• FIFO processing completion with multiple consumers.
• A particular throughput or latency result.

Those requirements need additional mechanisms. Durable work usually belongs in a persistent queue or broker with an explicit acknowledgement and redelivery model. Exactly-once claims require concrete transactional or deduplication behavior around every relevant side effect.

Channels remain useful without those promises. They provide a clear in-process handoff with configurable capacity, asynchronous backpressure, and completion. The key is to describe that contract accurately.

Frequently Asked Questions About Bounded C# Pipeline Channels

Why not use an unbounded channel?

The Channels guidance distinguishes unbounded channels from bounded channels that apply full-mode behavior. This article owns the bounded, lossless design, so the primary example uses a positive capacity and Wait.

Does BoundedChannelFullMode.Wait block a thread?

No. WriteAsync returns an awaitable operation that can remain pending until capacity is available, cancellation occurs, or the channel closes.

Should each producer complete the writer?

No when several producers share a channel. One coordinator should await all producers and complete the writer exactly once. In this example there is one producer, but completion still belongs to the coordinator so the lifecycle remains explicit.

Does canceling WriteAsync complete the channel?

No. Canceling WriteAsync ends that pending write attempt, while ChannelWriter.Complete remains a separate coordinator decision.

Are messages processed in FIFO order?

The Channels FIFO contract covers admission and removal at the queue boundary. With multiple consumers, processing can finish out of order after removal, so state the exact ordering boundary the application requires.

What happens to buffered disposable messages during an abort?

The local abort-cleanup test verifies that this sample waits for producer and consumer tasks to stop before the coordinator drains and disposes remaining messages. Disposing earlier would race with consumers that still own the reader.

Is a bounded Channel a durable queue?

No. The Channels documentation defines an in-process producer/consumer synchronization primitive, not durable storage, so durable delivery requires persistent infrastructure and a separate delivery contract.

Make the Lifecycle Part of the Design

A bounded C# pipeline with Channels is more than a bounded factory call. Capacity and Wait define admission. Honest reader and writer flags describe concurrency. Ownership defines who disposes each message. One coordinator defines completion. Cancellation stops pending operations, while completion closes admission. FIFO describes the queue, not necessarily processing finish order.

The complete pattern is straightforward once every responsibility has one owner: producers create and transfer messages, consumers process and dispose them, and the coordinator supervises tasks, drains normally, aborts on faults, and cleans what remains. That is the foundation for a lossless in-memory pipeline without pretending that Channels provide durability or exactly-once behavior.