A channel is simply a data structure that is used to store the data for the consumer to retrieve an appropriate synchronization to enable that to happen safely, while also enabling appropriate notification in both directions. There are multiple possible design decisions involved.

Lets take a simple example of the channel.

Example 1

public sealed class Channel
{
    public void Write (T value);
    public ValueTask ReadAsync(CancellationToken cancellationToken = default);
}
				

Here our Write method gives us a method we can use to produce data into the channel. And our ReadAsync method gives us a method to consume from that method.

Data production will always be accomplished successfully and simultaneously since we manifested our channel is unconditional. Hence, we tried to make it non-asynchronous and void-returning. In contrast, our method for consuming is ReadAsync, which is asynchronous because the data we want to consume may not be available yet.

For that we will need to wait for it to arrive if nothing is available to consume at the time we try. And while in our getting-started design we are not overly concerned with performance, we also don’t want to have lots of unnecessary overheads.

Since we expect to be reading frequently, and for us to often be reading when data is already available to be consumed, our ReadAsync method returns a ValueTask rather than a Task, so that we can make it allocation-free when it completes synchronously.

Now we need to implement these two methods. We will add two fields to our type.One will be used to serve as the storage mechanism and the other will be used to coordinate between the producer and consumer.

private readonly ConcurrentQueue _concurrentQueue = new ConcurrentQueue();
private readonly SemaphoreSlim _slimSemaphore = new SemaphoreSlim(0);
				

We use a ConcurrentQueue to store the data, ConcurrentQueue is already thread-safe for any number of producers and any number of consumers to access concurrently and freeing us from needing to do our locking process to protect the buffering data structure.

And we have used a SempahoreSlim to help coordinate between producers and consumers. And to notify the consumers that might be waiting for additional data to arrive.

Our write method is simple, we just need to store the data into the queue and increment the semaphores count by releasing it.

public void Write (T value)
{
_queue.Enqueue(value);
    _semaphore.Release(); 
}
			

Enqueue is used to store data, whereas semaphore will notify any consumer that more data is available.

public async ValueTask ReadAsync(CancellationToken cancellationToken = default)
{
    await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false); 
bool gotOne = _queue.TryDequeue(out T item); 
Debug.Assert(gotOne);
    return item;
}
			

In the above method, we need to wait for the data to be available.

System.Threading.Channels

public abstract class ChannelWriterClass
{
    public abstract bool TryWrite(T item);
    public virtual ValueTask WriteAsync(T item, CancellationToken cancellationToken = default);
    public abstract ValueTask WaitToWriteAsync(CancellationToken cancellationToken = default);
    public void Complete(Exception error);
    public virtual bool TryComplete(Exception error);
}
			

public abstract class ChannelReaderClass
{
    public abstract bool TryRead(out T item);
    public virtual ValueTask ReadAsync(CancellationToken cancellationToken = default)
    public abstract ValueTask WaitToReadAsync(CancellationToken cancellationToken = default);
    public virtual IAsyncEnumerable ReadAllAsync([EnumeratorCancellation] CancellationToken cancellationToken = default);
    public virtual Task CompletionTask { get; }
}
				

However, ChannelWriter also provides the WriteAsync method, in such a case where the channel is full and writing would need to wait, WriteAsync can be used, with the producer awaiting the result of WriteAsync and when the room becomes available then it will be allowed to continue.

There are some situations where code may not want to produce a value immediately. If a value represents an expensive resource or producing a value is expensive and if there is a reasonable chance the producer is running faster than the consumer, then it is because of the producer may want to delay the producing a value until it knows that write will be immediately successful.

For that, there is WaitToWriteAsync. A producer can wait for WaitToWriteAsync to return true, and after that choose to produce a value, then WriteAsyncs or TryWrites to the channel.

Note here that WriteAsync is virtual. Some implementations may provide a more optimized implementation but some base type can provide a reasonable implementation with abstract TryWrite and WaitToWriteAsync.

public async ValueTask WriteAsync(T item, CancellationToken cancellationToken)
{
    while (await WaitToWriteAsync(cancellationToken).ConfigureAwait(false))
        if (TryWrite(item))
            return;

    throw new ChannelCompletedException();
}
				

Here first, the while loop is present because channels by default can be used by any number of consumers and any number of producers concurrently.

There are a variety of typical patterns for how one consumes from a ChannelReader.We can apply this ReadAsync() method for infinite stream of values as:

while (true)
{
    T item = await channelReader.ReadAsync();
    Use(item);
}	
				

Up to this, we know how to read from readers and write to writers. But they come from the built-in channel Channel.

public abstract class Channel
{
    public ChannelReader Reader { get;  }
    public ChannelWriter Writer { get; }
}