Round #211

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 * We will create a `Producer` class that produces items and writes them to a `Channel<T>`. * We will create a `Consumer` class that reads items from the `Channel<T>` and processes them. * We will use the `Channel<T>` class to handle the asynchronous communication between the producer and consumer. ### Implementation ```csharp using System; using System.Threading; using System.Threading.Channels; using System.Threading.Tasks; // Define a class to represent the item being produced and consumed public class Item { public int Id { get; set; } public string Name { get; set; } public Item(int id, string name) { Id = id; Name = name; } } // Define the producer class public class Producer { private readonly Channel<Item> _channel; private readonly CancellationTokenSource _cts; public Producer(Channel<Item> channel, CancellationTokenSource cts) { _channel = channel; _cts = cts; } public async Task ProduceItemsAsync(int count) { for (int i = 0; i < count; i++) { if (_cts.IsCancellationRequested) { break; } var item = new Item(i, $"Item {i}"); await _channel.Writer.WriteAsync(item); Console.WriteLine($"Produced item: {item.Name}"); // Simulate some work being done await Task.Delay(100); } // Complete the channel writer _channel.Writer.Complete(); } } // Define the consumer class public class Consumer { private readonly Channel<Item> _channel; private readonly CancellationTokenSource _cts; public Consumer(Channel<Item> channel, CancellationTokenSource cts) { _channel = channel; _cts = cts; } public async Task ConsumeItemsAsync() { while (await _channel.Reader.WaitToReadAsync(_cts.Token)) { while (_channel.Reader.TryRead(out var item)) { Console.WriteLine($"Consumed item: {item.Name}"); // Simulate some work being done await Task.Delay(100); } } } } class Program { static async Task Main(string[] args) { // Create a bounded channel with a capacity of 10 items var channel = Channel.CreateBounded<Item>(new BoundedChannelOptions(10) { FullMode = BoundedChannelFullMode.Wait }); // Create a cancellation token source var cts = new CancellationTokenSource(); // Create a producer and consumer var producer = new Producer(channel, cts); var consumer = new Consumer(channel, cts); // Start the producer and consumer var producerTask = producer.ProduceItemsAsync(20); var consumerTask = consumer.ConsumeItemsAsync(); // Wait for the producer to complete await producerTask; // Wait for the consumer to complete await consumerTask; } } ``` ### Explanation * We define a `Producer` class that takes a `Channel<T>` and a `CancellationTokenSource` in its constructor. * The `ProduceItemsAsync` method produces items and writes them to the channel. * We define a `Consumer` class that takes a `Channel<T>` and a `CancellationTokenSource` in its constructor. * The `ConsumeItemsAsync` method reads items from the channel and processes them. * In the `Main` method, we create a bounded channel with a capacity of 10 items and a cancellation token source. * We create a producer and consumer, and start them using the `ProduceItemsAsync` and `ConsumeItemsAsync` methods. * We wait for the producer to complete using the `await` keyword, and then wait for the consumer to complete. ### Advice * Use the `Channel<T>` class to handle asynchronous communication between threads. * Use the `CancellationTokenSource` class to handle cancellation tokens. * Use the `await` keyword to wait for asynchronous operations to complete. * Use the `TryRead` method to read items from the channel without blocking. * Use the `WaitToReadAsync` method to wait for items to be available in the channel without blocking.

```csharp using System; using System.Threading; using System.Threading.Tasks; using System.Threading.Channels; public class ProducerConsumerExample { public async Task RunExample() { // Create a bounded channel with a capacity of 10 items var channel = Channel.CreateBounded<int>(new BoundedChannelOptions(10) { FullMode = BoundedChannelFullMode.Wait }); // Start the producer _ = ProduceNumbers(channel.Writer, 1, 10); // Start the consumer await ConsumeNumbers(channel.Reader); } private async Task ProduceNumbers(ChannelWriter<int> writer, int start, int end) { for (int i = start; i <= end; i++) { Console.WriteLine($"Producing: {i}"); await writer.WriteAsync(i); await Task.Delay(100); // Simulate production delay } writer.Complete(); } private async Task ConsumeNumbers(ChannelReader<int> reader) { while (await reader.WaitToReadAsync()) { while (reader.TryRead(out int number)) { Console.WriteLine($"Consumed: {number}"); } } } } class Program { static async Task Main(string[] args) { var example = new ProducerConsumerExample(); await example.RunExample(); } } ``` This implementation uses `Channel<T>` to create a bounded channel, which allows for asynchronous producer-consumer communication. The `ProduceNumbers` method writes numbers to the channel, while the `ConsumeNumbers` method reads numbers from the channel. Here's a step-by-step explanation: 1. **Channel Creation**: A bounded channel is created with a capacity of 10 items. The `FullMode` is set to `BoundedChannelFullMode.Wait`, which means that if the channel is full, the producer will wait until space becomes available. 2. **Producer**: The `ProduceNumbers` method is started, which writes numbers from 1 to 10 to the channel. The `WriteAsync` method is used to write to the channel, and `Task.Delay` is used to simulate a production delay. 3. **Consumer**: The `ConsumeNumbers` method is started, which reads numbers from the channel. The `WaitToReadAsync` method is used to wait until data is available in the channel, and `TryRead` is used to read from the channel. 4. **Completion**: Once the producer has finished writing to the channel, it calls `Complete` to signal that no more data will be written. The consumer continues to read from the channel until it is empty. This implementation demonstrates the async producer-consumer pattern using `Channel<T>`, allowing for efficient and asynchronous communication between the producer and consumer.