Build a Channel Pipeline

Connect producers and consumers with an explicit capacity, completion protocol, and native message lifetime.

AdvancedThreading.ChannelsBuilt and run

Goal and component

Move values between producers and consumers without sharing a mutable container directly. Threading.Channels supplies bounded, unbounded, and priority-ordered channels backed by mutexes and condition variables.

set(SHARP_RUNTIME_COMPONENTS Threading.Channels)
target_link_libraries(app PRIVATE SharpRuntime::Threading.Channels)

Create a channel and keep both ends

#include <System/Threading/Channels/Channel.hpp>

using System::Threading::Channels::Channel;

auto channel = Channel<int>::CreateBounded(2);
const bool first = channel.Writer->TryWrite(10);
const bool second = channel.Writer->TryWrite(20);

int value{};
const bool read = channel.Reader->TryRead(value);
(void)first;
(void)second;
(void)read;

Channel<T> exposes shared reader and writer endpoints. The factory constructs shared backing state; copies of the endpoint shared_ptrs keep that state alive. The implementation is thread-safe but not lock-free.

Choose bounded or unbounded capacity

Factory	Use when	Risk to control
`CreateBounded(capacity)`	A producer must observe consumer pressure.	Blocking/waiting can deadlock if the consumer shares a required lock.
`CreateUnbounded()`	The workload is externally bounded and writes should proceed immediately.	A stalled consumer can turn queue growth into memory exhaustion.
`CreateUnboundedPrioritized(options)`	Ascending priority rather than FIFO is part of the protocol.	Low-priority work can starve; comparer correctness becomes shared-state correctness.

The bounded default is BoundedChannelFullMode::Wait. The other full modes drop the incoming, newest, or oldest item according to the selected enum. A drop policy is an application-level data-loss decision, not a tuning flag.

Wait without busy-spinning

while (channel.Reader->WaitToReadAsync().Wait()) {
    int item{};
    while (channel.Reader->TryRead(item)) {
        // Process item. Do not hold an unrelated lock while doing slow work.
    }
}

This is the current replacement for .NET’s await foreach (ReadAllAsync()), which is not reproduced. A waiting task uses background work; retain all state captured by processing code and decide where blocking is acceptable.

Complete the producer side

channel.Writer->Complete();

// Existing buffered items can still be drained. Once closed and drained,
// WaitToReadAsync returns false and ReadAsync throws ChannelClosedException.

TryComplete reports a duplicate close as false; Complete throws ChannelClosedException for it. Completion may carry an exception_ptr. ReadAsync/WriteAsync wrap a producer failure as the closed-channel cause, whereas the wait/completion surfaces expose the completion error directly. Test the surface your application actually observes.

Async endpoint lifetime

The default ReadAsync and WriteAsync implementations call shared_from_this and retain their endpoint for the background task. Factory-created endpoints satisfy that contract. A custom subclass must also live in a shared_ptr before calling these methods or shared_from_this fails.

Channel ownership does not retain payload dependencies

The queue owns each stored T according to normal C++ copy/move rules. A raw pointer, span, or reference wrapper inside T can still dangle while waiting in the channel. Prefer values or explicit smart ownership for cross-thread messages.

C# comparison

C# / .NET	Sharp Runtime	Difference
`Channel.CreateBounded<T>`	`Channel<T>::CreateBounded`	Factory result holds public shared Reader/Writer endpoints.
`await reader.ReadAsync()`	`reader->ReadAsync().Wait()`	No language-level await; blocking and scheduling are explicit.
`ReadAllAsync`	Wait/try-read loop	No async-enumerable integration.
GC-managed message graph	Native value/smart-pointer rules	Borrowed payload members need separate lifetime proof.

Test the protocol, not just the container

Test zero/one/saturated capacity and every selected full mode.
Test completion both before and after buffered items are drained.
Test producer failure, duplicate completion, and writes after close.
Prove shutdown cannot leave a producer waiting for a consumer that has exited.
Use a bounded message count and fixed worker count in stress tests to keep memory use predictable.