Round #236

Implement async producer-consumer pattern using Channel<T>

**Async Producer-Consumer Pattern using Channel<T>** The producer-consumer pattern is a classic design pattern that allows for asynchronous communication between two or more threads. In this implementation, we will use the `Channel<T>` class in C# to create an async producer-consumer pattern. ### Overview * The producer will write items to the channel. * The consumer will read items from the channel. ### Implementation ```csharp using System; using System.Threading; using System.Threading.Channels; using System.Threading.Tasks; public class ProducerConsumerExample { private readonly Channel<int> _channel; public ProducerConsumerExample(int capacity = 10) { _channel = Channel.CreateBounded<int>(new BoundedChannelOptions(capacity) { FullMode = BoundedChannelFullMode.Wait }); } public async Task ProduceAsync(int itemId) { await _channel.Writer.WriteAsync(itemId); Console.WriteLine($"Produced item {itemId}"); } public async Task ConsumeAsync(CancellationToken cancellationToken) { while (await _channel.Reader.WaitToReadAsync(cancellationToken)) { while (_channel.Reader.TryRead(out int item)) { Console.WriteLine($"Consumed item {item}"); } } } public async Task StartAsync(int numItems, CancellationToken cancellationToken) { // Start the consumer _ = ConsumeAsync(cancellationToken); // Start the producer for (int i = 0; i < numItems; i++) { await ProduceAsync(i); await Task.Delay(100, cancellationToken); // Simulate production delay } // Complete the channel _channel.Writer.Complete(); // Wait for the consumer to finish await _channel.Reader.Completion; } } class Program { static async Task Main(string[] args) { var example = new ProducerConsumerExample(); var cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(10)); await example.StartAsync(10, cancellationTokenSource.Token); } } ``` ### Explanation * We create a `Channel<int>` with a bounded capacity of 10 items. * The `ProduceAsync` method writes an item to the channel. * The `ConsumeAsync` method reads items from the channel and continues to do so until the channel is closed. * The `StartAsync` method starts the consumer and producer, and waits for the consumer to finish. * In the `Main` method, we create an instance of `ProducerConsumerExample` and start the async producer-consumer pattern. ### Advice * Use `Channel<T>` for async producer-consumer patterns when you need to handle a stream of data. * Always complete the channel writer when you're done producing items. * Use `WaitToReadAsync` to wait for items to be available in the channel. * Use `TryRead` to read items from the channel without blocking. * Use `CancellationToken` to handle cancellation and timeouts.

```csharp using System; using System.Threading; using System.Threading.Tasks; using System.Threading.Channels; public class ProducerConsumerExample { public async Task RunExampleAsync() { // Create a bounded channel with a capacity of 10 items var channel = Channel.CreateBounded<int>(new BoundedChannelOptions(10) { FullMode = BoundedChannelFullMode.Wait }); // Start the producer _ = ProduceAsync(channel.Writer, 10); // Start the consumer await ConsumeAsync(channel.Reader); } private async Task ProduceAsync(ChannelWriter<int> writer, int count) { try { for (int i = 0; i < count; i++) { await writer.WriteAsync(i); Console.WriteLine($"Produced: {i}"); await Task.Delay(100); } writer.Complete(); } catch (Exception ex) { Console.WriteLine($"Producer error: {ex.Message}"); writer.TryComplete(ex); } } private async Task ConsumeAsync(ChannelReader<int> reader) { try { while (await reader.WaitToReadAsync()) { while (reader.TryRead(out int item)) { Console.WriteLine($"Consumed: {item}"); } } } catch (Exception ex) { Console.WriteLine($"Consumer error: {ex.Message}"); } } } class Program { static async Task Main(string[] args) { var example = new ProducerConsumerExample(); await example.RunExampleAsync(); } } ``` This C# code implements the async producer-consumer pattern using `Channel<T>`. The `ProducerConsumerExample` class contains two main methods: `ProduceAsync` and `ConsumeAsync`. - The `ProduceAsync` method writes items to the channel. It uses a `for` loop to produce a specified number of items, writing each item to the channel using `await writer.WriteAsync(i)`. After producing all items, it calls `writer.Complete()` to signal the end of production. - The `ConsumeAsync` method reads items from the channel. It uses a `while` loop to continuously read items from the channel using `await reader.WaitToReadAsync()` and `reader.TryRead(out int item)`. This allows the consumer to wait for new items to be produced and then consume them. The `RunExampleAsync` method creates a bounded channel and starts both the producer and consumer. The `Main` method in the `Program` class demonstrates how to use the `ProducerConsumerExample` class. The code includes error handling to ensure that any exceptions that occur during production or consumption are caught and logged. The `TryComplete` method is used to complete the channel with an exception, allowing the consumer to handle the error.