Round #287

Implement async producer-consumer pattern using Channel<T>

Below is a **complete, self‑contained example** that shows how to build an **asynchronous producer‑consumer pipeline** with the modern `System.Threading.Channels.Channel<T>` type. The code demonstrates: * Creating a bounded (or unbounded) channel. * One or many **producer** tasks that write to the channel without blocking the caller. * One or many **consumer** tasks that read from the channel asynchronously. * Graceful shutdown (cancellation, completion, and error handling). * A simple console UI that you can run directly (`dotnet run`) to see the pattern in action. --- ## 1. Project layout ```text AsyncProducerConsumer/ │ ├─ Program.cs ← entry point (contains the demo) ├─ Producer.cs ← static helper that produces items ├─ Consumer.cs ← static helper that consumes items └─ Models/ └─ WorkItem.cs ← the payload type (you can replace it with any T) ``` All files are shown below, but you can also copy‑paste the whole thing into a single `Program.cs` file if you prefer a single‑file demo. --- ## 2. The payload – `WorkItem` ```csharp // Models/WorkItem.cs namespace AsyncProducerConsumer.Models; /// <summary> /// Simple data that travels through the channel. /// Replace or extend this with whatever your real domain model is. /// </summary> public sealed class WorkItem { public int Id { get; } public DateTime CreatedAt { get; } public WorkItem(int id) => (Id, CreatedAt) = (id, DateTime.UtcNow); public override string ToString() => $"WorkItem {{ Id = {Id}, CreatedAt = {CreatedAt:O} }}"; } ``` --- ## 3. Producer helper ```csharp // Producer.cs using System.Threading.Channels; using AsyncProducerConsumer.Models; namespace AsyncProducerConsumer; /// <summary> /// Produces <see cref="WorkItem"/> instances and writes them into the supplied channel. /// </summary> public static class Producer { /// <summary> /// Starts a background task that generates items at the given interval. /// </summary> /// <param name="channelWriter">The writer side of the channel.</param> /// <param name="delay">Delay between items (simulates work).</param> /// <param name="cancellationToken">Token to stop production.</param> /// <returns>A task that completes when production stops or the channel is completed.</returns> public static async Task RunAsync( ChannelWriter<WorkItem> channelWriter, TimeSpan delay, CancellationToken cancellationToken) { int nextId = 0; try { while (!cancellationToken.IsCancellationRequested) { var item = new WorkItem(Interlocked.Increment(ref nextId)); // The WriteAsync call respects back‑pressure when the channel is bounded. await channelWriter.WriteAsync(item, cancellationToken); Console.WriteLine($"[Producer] Enqueued {item}"); await Task.Delay(delay, cancellationToken); } } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { // Normal shutdown – just fall through. } finally { // Signal to consumers that no more items will be written. channelWriter.Complete(); Console.WriteLine("[Producer] Completed channel."); } } } ``` --- ## 4. Consumer helper ```csharp // Consumer.cs using System.Threading.Channels; using AsyncProducerConsumer.Models; namespace AsyncProducerConsumer; /// <summary> /// Consumes <see cref="WorkItem"/> instances from the supplied channel. /// </summary> public static class Consumer { /// <summary> /// Starts a background task that reads items until the channel is completed. /// </summary> /// <param name="channelReader">The reader side of the channel.</param> /// <param name="consumerId">An identifier used for logging.</param> /// <param name="cancellationToken">Token to stop consumption early.</param> /// <returns>A task that completes when the channel is closed or cancelled.</returns> public static async Task RunAsync( ChannelReader<WorkItem> channelReader, int consumerId, CancellationToken cancellationToken) { try { await foreach (var item in channelReader.ReadAllAsync(cancellationToken)) { // Simulate processing time. await ProcessAsync(item, consumerId, cancellationToken); } } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { // Normal early exit. } finally { Console.WriteLine($"[Consumer {consumerId}] Finished."); } } private static async Task ProcessAsync(WorkItem item, int consumerId, CancellationToken ct) { // Here you would put your real business logic. // For demo purposes we just wait a random short time. var rnd = new Random(); int processingMs = rnd.Next(50, 200); await Task.Delay(processingMs, ct); Console.WriteLine( $"[Consumer {consumerId}] Processed {item} (took {processingMs} ms)"); } } ``` --- ## 5. Program – wiring everything together ```csharp // Program.cs using System.Threading.Channels; using AsyncProducerConsumer; using AsyncProducerConsumer.Models; namespace AsyncProducerConsumerDemo; class Program { static async Task Main(string[] args) { // -------------------------------------------------------------- // 1️⃣ Choose the channel type // -------------------------------------------------------------- // Bounded channel – back‑pressure, max 100 items in the buffer. var options = new BoundedChannelOptions(100) { FullMode = BoundedChannelFullMode.Wait, // producers wait when full SingleWriter = false, // we may have many producers SingleReader = false // we may have many consumers }; var channel = Channel.CreateBounded<WorkItem>(options); // If you prefer an unbounded channel (no back‑pressure) use: // var channel = Channel.CreateUnbounded<WorkItem>(); // -------------------------------------------------------------- // 2️⃣ Cancellation token – press Ctrl+C to stop the demo // -------------------------------------------------------------- using var cts = new CancellationTokenSource(); Console.CancelKeyPress += (_, e) => { e.Cancel = true; // prevent the process from terminating immediately cts.Cancel(); // signal cancellation to all tasks Console.WriteLine("\nCancellation requested…"); }; // -------------------------------------------------------------- // 3️⃣ Start producers // -------------------------------------------------------------- var producerTasks = new[] { // Producer 1 – fast Producer.RunAsync(channel.Writer, TimeSpan.FromMilliseconds(100), cts.Token), // Producer 2 – slower Producer.RunAsync(channel.Writer, TimeSpan.FromMilliseconds(250), cts.Token) }; // -------------------------------------------------------------- // 4️⃣ Start consumers // -------------------------------------------------------------- var consumerCount = 3; var consumerTasks = Enumerable.Range(1, consumerCount) .Select(id => Consumer.RunAsync(channel.Reader, id, cts.Token)) .ToArray(); // -------------------------------------------------------------- // 5️⃣ Wait for everything to finish // -------------------------------------------------------------- // When cancellation is requested the producers will complete the writer, // which causes the readers to finish after draining the queue. await Task.WhenAll(producerTasks); await Task.WhenAll(consumerTasks); Console.WriteLine("\nAll producers and consumers have shut down gracefully."); } } ``` --- ## 6. How it works (quick walkthrough) | Step | What happens | |------|--------------| | **Channel creation** | `Channel.CreateBounded<T>(options)` builds a thread‑safe queue with a capacity (`100` in the example). When the buffer is full, `WriteAsync` **awaits** until a consumer removes an item. | | **Producer** | Loops until cancelled, creates a `WorkItem`, and calls `await writer.WriteAsync(item, ct)`. If the channel is full the call asynchronously waits, providing natural back‑pressure. When the loop ends the producer calls `writer.Complete()` to signal “no more items”. | | **Consumer** | Uses `await foreach (var item in reader.ReadAllAsync(ct))`. `ReadAllAsync` completes only when the writer has called `Complete()` **and** the buffer is empty, so the consumer automatically drains the remaining items before exiting. | | **Cancellation** | Pressing **Ctrl+C** triggers `cts.Cancel()`. All `Task.Delay` and `WriteAsync/ReadAllAsync` calls observe the token and throw `OperationCanceledException`, which we catch to exit cleanly. | | **Multiple producers / consumers** | The channel is safe for concurrent writers and readers (`SingleWriter = false`, `SingleReader = false`). Adding more producers or consumers is just a matter of launching more tasks. | --- ## 7. Running the demo ```bash # From the folder that contains the .csproj (or create a new console project) dotnet new console -n AsyncProducerConsumerDemo # Replace the generated Program.cs with the files above (or copy‑paste them). dotnet run ``` You’ll see output similar to: ``` [Producer] Enqueued WorkItem { Id = 1, CreatedAt = 2026-05-08T12:34:56.789Z } [Consumer 1] Processed WorkItem { Id = 1, CreatedAt = 2026-05-08T12:34:56.789Z } (took 73 ms) [Producer] Enqueued WorkItem { Id = 2, CreatedAt = 2026-05-08T12:34:56.889Z } [Producer] Enqueued WorkItem { Id = 3, CreatedAt = 2026-05-08T12:34:57.139Z } [Consumer 2] Processed WorkItem { Id = 2, CreatedAt = 2026-05-08T12:34:56.889Z } (took 112 ms) ... ^C Cancellation requested… [Producer] Completed channel. [Consumer 1] Finished. [Consumer 2] Finished. [Consumer 3] Finished. All producers and consumers have shut down gracefully. ``` The program stops cleanly, drains the queue, and disposes of all resources. --- ## 8. Extending the pattern * **Error handling** – If a producer encounters a non‑recoverable error, call `writer.Complete(exception)`; the exception will be re‑thrown from `ReadAllAsync` for the consumers to handle. * **Transform pipelines** – Chain multiple channels (e.g., `Channel<A>` → `Channel<B>` → `Channel<C>`) to build a processing pipeline. * **Parallel processing** – Inside the consumer you can `Task.Run` the actual work or use `Parallel.ForEachAsync` if you need more concurrency per item. * **Back‑pressure tuning** – Adjust `BoundedChannelOptions.FullMode` (`Wait`, `DropOldest`, `DropNewest`, `DropWrite`) depending on the semantics you need. --- ### TL;DR *Create a `Channel<T>` (bounded or unbounded). *Producers `await writer.WriteAsync(item, ct)`. *Consumers `await foreach (var item in reader.ReadAllAsync(ct))`. *Signal completion with `writer.Complete()`. *Use a `CancellationToken` for graceful shutdown.* That’s the idiomatic, fully‑async producer‑consumer pattern in modern C#. Happy coding!

**Solution Overview** The .NET `System.Threading.Channels` library provides a high‑performance, thread‑safe queue that can be used to implement the classic producer‑consumer pattern without the pitfalls of `BlockingCollection` or manual lock handling. The implementation below demonstrates: * A **bounded** channel (`Channel<T>.CreateBounded`) that limits the number of buffered items. * An **asynchronous producer** that writes items to the channel (`ChannelWriter<T>.WriteAsync`). * An **asynchronous consumer** that reads items from the channel (`ChannelReader<T>.ReadAllAsync`). * Graceful shutdown via `CancellationToken`. * A small console‑app that wires everything together and shows how the pattern can be registered in a DI container for real‑world services. --- ## 1. Project structure ``` src/ └─ AsyncProducerConsumer/ ├─ Program.cs ├─ IProducer.cs ├─ Producer.cs ├─ IConsumer.cs └─ Consumer.cs ``` All code targets **.NET 8** (or any .NET 6+ runtime) and uses the built‑in `System.Threading.Channels` package (no extra NuGet required). --- ## 2. Core contracts ```csharp // IProducer.cs public interface IProducer<T> { /// <summary> /// Starts producing items until the supplied cancellation token is triggered. /// </summary> Task ProduceAsync(CancellationToken cancellationToken); } // IConsumer.cs public interface IConsumer<T> { /// <summary> /// Starts consuming items until the channel is completed or the cancellation token is triggered. /// </summary> Task ConsumeAsync(CancellationToken cancellationToken); } ``` These interfaces keep the producer and consumer loosely coupled and make them easy to register in DI. --- ## 3. Implementation ### 3.1 Producer ```csharp // Producer.cs using System.Threading.Channels; public class Producer<T> : IProducer<T> { private readonly ChannelWriter<T> _writer; private readonly Func<CancellationToken, Task<T>> _itemFactory; private readonly TimeSpan _delayBetweenItems; /// <param name="writer">The channel writer that the producer will write to.</param> /// <param name="itemFactory"> /// A delegate that creates the next item. It receives the cancellation token so it can abort early. /// </param> /// <param name="delayBetweenItems"> /// Optional delay to simulate work between items (default = 0). /// </param> public Producer(ChannelWriter<T> writer, Func<CancellationToken, Task<T>> itemFactory, TimeSpan? delayBetweenItems = null) { _writer = writer ?? throw new ArgumentNullException(nameof(writer)); _itemFactory = itemFactory ?? throw new ArgumentNullException(nameof(itemFactory)); _delayBetweenItems = delayBetweenItems ?? TimeSpan.Zero; } public async Task ProduceAsync(CancellationToken cancellationToken) { try { while (!cancellationToken.IsCancellationRequested) { // Create the next item (could be I/O bound, hence async) T item = await _itemFactory(cancellationToken).ConfigureAwait(false); // WriteAsync respects back‑pressure: it will await when the channel is full. await _writer.WriteAsync(item, cancellationToken).ConfigureAwait(false); if (_delayBetweenItems > TimeSpan.Zero) await Task.Delay(_delayBetweenItems, cancellationToken).ConfigureAwait(false); } } finally { // Signal that no more items will be written. _writer.Complete(); } } } ``` ### 3.2 Consumer ```csharp // Consumer.cs using System.Threading.Channels; public class Consumer<T> : IConsumer<T> { private readonly ChannelReader<T> _reader; private readonly Func<T, CancellationToken, Task> _processItemAsync; /// <param name="reader">The channel reader that the consumer will read from.</param> /// <param name="processItemAsync"> /// Asynchronous delegate that processes a single item. /// </param> public Consumer(ChannelReader<T> reader, Func<T, CancellationToken, Task> processItemAsync) { _reader = reader ?? throw new ArgumentNullException(nameof(reader)); _processItemAsync = processItemAsync ?? throw new ArgumentNullException(nameof(processItemAsync)); } public async Task ConsumeAsync(CancellationToken cancellationToken) { // ReadAllAsync automatically completes when the writer calls Complete(). await foreach (var item in _reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) { await _processItemAsync(item, cancellationToken).ConfigureAwait(false); } } } ``` --- ## 4. Wiring everything together (Console app) ```csharp // Program.cs using System.Threading.Channels; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; await Host.CreateDefaultBuilder(args) .ConfigureServices((context, services) => { // 1️⃣ Create a bounded channel that can hold up to 100 items. var channel = Channel.CreateBounded<int>(new BoundedChannelOptions(100) { FullMode = BoundedChannelFullMode.Wait, // back‑pressure: producers wait when full SingleWriter = false, SingleReader = false }); // 2️⃣ Register the channel (both reader & writer) as singletons. services.AddSingleton(channel); services.AddSingleton(channel.Reader); services.AddSingleton(channel.Writer); // 3️⃣ Register concrete producer & consumer. services.AddSingleton<IProducer<int>>(sp => new Producer<int>( sp.GetRequiredService<ChannelWriter<int>>(), async ct => { // Simulate work that produces a number. await Task.Delay(50, ct); // pretend we fetched data return Random.Shared.Next(0, 1000); }, TimeSpan.FromMilliseconds(10))); services.AddSingleton<IConsumer<int>>(sp => new Consumer<int>( sp.GetRequiredService<ChannelReader<int>>(), async (value, ct) => { // Simulate processing. await Task.Delay(30, ct); Console.WriteLine($"Consumed: {value}"); })); // 4️⃣ Register a hosted service that runs both producer & consumer. services.AddHostedService<ProducerConsumerHostedService>(); }) .RunAsync(); /// <summary> /// Hosted service that starts the producer and consumer tasks and /// coordinates graceful shutdown. /// </summary> public sealed class ProducerConsumerHostedService : IHostedService { private readonly IProducer<int> _producer; private readonly IConsumer<int> _consumer; private readonly IHostApplicationLifetime _appLifetime; private readonly CancellationTokenSource _cts = new(); private Task? _producerTask; private Task? _consumerTask; public ProducerConsumerHostedService(IProducer<int> producer, IConsumer<int> consumer, IHostApplicationLifetime appLifetime) { _producer = producer; _consumer = consumer; _appLifetime = appLifetime; } public Task StartAsync(CancellationToken cancellationToken) { // Run producer & consumer in the background. _producerTask = _producer.ProduceAsync(_cts.Token); _consumerTask = _consumer.ConsumeAsync(_cts.Token); // When either side faults, stop the whole host. Task.WhenAny(_producerTask, _consumerTask).ContinueWith(t => { // Log the exception (in a real app use ILogger) if (t.IsFaulted) Console.Error.WriteLine(t.Exception?.Flatten().Message); // Trigger shutdown. _appLifetime.StopApplication(); }); return Task.CompletedTask; } public async Task StopAsync(CancellationToken cancellationToken) { // Signal cancellation and wait for both tasks to finish. _cts.Cancel(); var tasks = new[] { _producerTask, _consumerTask }; await Task.WhenAll(tasks.Where(t => t != null)).ConfigureAwait(false); } } ``` ### How it works 1. **Channel creation** – a bounded channel with a capacity of 100 items is created. When the buffer is full, `WriteAsync` will await until the consumer drains items, providing natural back‑pressure. 2. **Producer** – repeatedly generates a random integer, simulates a small delay, and writes it to the channel. When the host is stopping (cancellation token triggered) the producer exits the loop and calls `Complete()` on the writer, signalling the consumer that no more data will arrive. 3. **Consumer** – uses `ReadAllAsync` to enumerate items until the channel is completed. Each item is processed (here we just print it after a short delay). 4. **Hosted service** – starts both tasks when the host starts, monitors for failures, and ensures a graceful shutdown when the application stops or an exception occurs. --- ## 5. Dependency‑Injection registration (summary) ```csharp services.AddSingleton(channel); // the whole Channel<T> services.AddSingleton(channel.Reader); // ChannelReader<T> services.AddSingleton(channel.Writer); // ChannelWriter<T> services.AddSingleton<IProducer<T>, Producer<T>>(); services.AddSingleton<IConsumer<T>, Consumer<T>>(); services.AddHostedService<ProducerConsumerHostedService>(); ``` These registrations make the pattern reusable in any ASP.NET Core, Worker Service, or generic host scenario. --- ## 6. Running the sample ```bash dotnet run --project src/AsyncProducerConsumer ``` You will see a continuous stream of lines like: ``` Consumed: 483 Consumed: 12 Consumed: 759 ... ``` Press **Ctrl+C** to stop; the host will cancel the token, the producer will complete the channel, and the consumer will finish processing any remaining items before the process exits. --- ### Key take‑aways * `Channel<T>` replaces older constructs (`BlockingCollection`, manual `ConcurrentQueue` + `SemaphoreSlim`) with a single, well‑tested abstraction. * Back‑pressure is automatic: a bounded channel makes the producer await when the buffer is full. * `ReadAllAsync` provides a clean, cancellation‑aware enumeration of items. * The pattern fits naturally into the generic host/DI model, enabling clean separation of concerns and testability.