Channels
A thread-safe queue for async producers and consumers: creating a channel, writing to it, reading from it and completing it.
Channel<T> channel = Channel.CreateUnbounded<T>(); // then channel.Writer and channel.ReaderWhat a channel is
A channel is a queue that two pieces of code can use at the same time. A queue is a line: items come out in the order they went in. The code that puts items in is called the producer. The code that takes them out is the consumer. Code that is safe to use from two pieces of work running at once is called thread-safe, and a channel is thread-safe by design, so you do not guard it yourself.
Channels live in the System.Threading.Channels namespace. You create one with a static method rather than new: Channel.CreateUnbounded<string>(). Unbounded means the channel will hold as many items as memory allows.
The two halves
A Channel<T> has two properties. Writer is a ChannelWriter<T>, and Reader is a ChannelReader<T>. Give the writer to the producer and the reader to the consumer, and neither one can do the other's job by accident.
await writer.WriteAsync(item) adds an item. writer.Complete() says no more items will ever be written. reader.ReadAllAsync() gives you something you can walk with await foreach, which is a foreach loop allowed to wait between items. That loop ends when the channel is complete and empty.
Shorthand and longhand
await foreach (var x in reader.ReadAllAsync()) is shorthand. The long form is while (await reader.WaitToReadAsync()) wrapped around while (reader.TryRead(out var x)). They do the same work; the short form is what you write today.
What trips people up
A channel is not a broadcast. Each item goes to exactly one reader. Two consumers reading one channel split the items between them.
Example
using System.Threading.Channels;
// A channel is a queue with two ends. The two sides share no variable directly.
Channel<string> channel = Channel.CreateUnbounded<string>();
// Shorthand for new string[] { "alpha", "beta", "gamma" }.
string[] words = ["alpha", "beta", "gamma"];
// The producer runs on its own. It writes three items, then closes the channel.
Task producer = Task.Run(async () =>
{
foreach (string word in words)
{
await channel.Writer.WriteAsync(word);
}
// Complete means: no more items will ever be written.
channel.Writer.Complete();
});
// The consumer. This loop ends when the channel is complete and empty.
int count = 0;
await foreach (string word in channel.Reader.ReadAllAsync())
{
count++;
Console.WriteLine($"read {count}: {word}");
}
// Awaiting the producer surfaces any exception it threw.
await producer;
Console.WriteLine($"finished after {count} items");
// Output:
// read 1: alpha
// read 2: beta
// read 3: gamma
// finished after 3 items
When to use it
- Hand log lines from a request handler to a single background writer without making the request wait for disk.
- Feed file paths from a directory scan to a worker that opens and parses each one.
- Collect progress messages from several running tasks into one place that prints them in arrival order.
- Replace a lock around a Queue<T> with a queue the consumer can await instead of poll.
More examples
The longhand read loop
Two nested loops are needed in the long form because WaitToReadAsync only tells you that at least one item is ready, and TryRead only takes one item. Read this version once so you can recognise it in older code, then use ReadAllAsync. The long form is still worth writing by hand when you want to drain several items per wake-up and batch them.
using System.Threading.Channels;
Channel<int> channel = Channel.CreateUnbounded<int>();
// TryWrite never waits. On an unbounded channel it always succeeds.
channel.Writer.TryWrite(1);
channel.Writer.TryWrite(2);
channel.Writer.Complete();
// Longhand. WaitToReadAsync returns false once the channel is complete and empty.
while (await channel.Reader.WaitToReadAsync())
{
while (channel.Reader.TryRead(out int value))
{
Console.WriteLine($"got {value}");
}
}
// Shorthand for exactly the same loop:
// await foreach (int value in channel.Reader.ReadAllAsync())
// Console.WriteLine($"got {value}");
// Output:
// got 1
// got 2
Passing the failure to the reader
Complete takes an optional exception. Items already in the channel are still read, and only when the channel is empty does the reader see the error. Use this when the producer fails: the consumer learns that the stream ended badly instead of treating a short stream as a complete one. The same exception is also available on reader.Completion, a Task that finishes when no more data will arrive.
using System.Threading.Channels;
Channel<int> channel = Channel.CreateUnbounded<int>();
channel.Writer.TryWrite(10);
// Complete can carry the exception that stopped the producer.
channel.Writer.Complete(new InvalidOperationException("source went away"));
try
{
await foreach (int value in channel.Reader.ReadAllAsync())
{
Console.WriteLine($"value {value}");
}
}
catch (InvalidOperationException ex)
{
Console.WriteLine($"reader saw: {ex.Message}");
}
// Output:
// value 10
// reader saw: source went away
Handing out one end only
ChannelWriter<T> and ChannelReader<T> are ordinary parameter types, so a method can take exactly the half it needs. This is how a channel is normally passed around a program. Note that the producer here finishes before the consumer starts: that only works because the channel is unbounded. With a capacity limit, the writes would wait for a reader that has not started yet and the program would stop.
using System.Threading.Channels;
Channel<int> channel = Channel.CreateUnbounded<int>();
// The producer receives only the writer, so it cannot read by mistake.
await ProduceAsync(channel.Writer, 3);
// The consumer receives only the reader, so it cannot write.
Console.WriteLine($"sum: {await SumAsync(channel.Reader)}");
static async Task ProduceAsync(ChannelWriter<int> writer, int count)
{
for (int i = 1; i <= count; i++)
{
await writer.WriteAsync(i);
}
writer.Complete();
}
static async Task<int> SumAsync(ChannelReader<int> reader)
{
int total = 0;
await foreach (int value in reader.ReadAllAsync())
{
total += value;
}
return total;
}
// Output:
// sum: 6
Discussion