Round #861

Build a producer-consumer pair using Channel<T>: producer emits work items at 100/sec; consumer processes them with bounded backpressure (max 50 in-flight).

**Producer-Consumer Pair using Channel<T>** ============================================= Below is an example implementation of a producer-consumer pair using `Channel<T>` in C#. ### Producer-Consumer Code ```csharp using System; using System.Threading; using System.Threading.Channels; using System.Threading.Tasks; class Program { static async Task Main(string[] args) { // Create a bounded channel with a capacity of 50 var channel = Channel.CreateBounded<int>(new BoundedChannelOptions(50) { FullMode = BoundedChannelFullMode.Wait }); // Start the producer _ = ProduceWorkItems(channel.Writer, 100); // Start the consumer _ = ConsumeWorkItems(channel.Reader); // Keep the program running await Task.Delay(TimeSpan.FromMinutes(1)); } static async Task ProduceWorkItems(ChannelWriter<int> writer, int rate) { var random = new Random(); while (true) { // Write a work item to the channel await writer.WriteAsync(random.Next(0, 100)); // Wait for the specified rate await Task.Delay(TimeSpan.FromSeconds(1.0 / rate)); } } static async Task ConsumeWorkItems(ChannelReader<int> reader) { while (await reader.WaitToReadAsync()) { while (reader.TryRead(out var workItem)) { // Process the work item Console.WriteLine($"Processing work item: {workItem}"); // Simulate processing time await Task.Delay(TimeSpan.FromMilliseconds(10)); } } } } ``` ### Explanation * We create a bounded channel with a capacity of 50 using `Channel.CreateBounded`. * The `ProduceWorkItems` method writes work items to the channel at a rate of 100 per second. * The `ConsumeWorkItems` method reads work items from the channel and processes them. * The `BoundedChannelFullMode.Wait` option ensures that the producer will wait if the channel is full, preventing it from overflowing. * The `WaitToReadAsync` method is used to wait for work items to become available in the channel. * The `TryRead` method is used to read work items from the channel without blocking. ### Advice * Adjust the capacity of the channel and the rate of the producer to suit your specific use case. * Consider using a different `BoundedChannelFullMode` option if you need to handle full channels differently. * Use `await Task.Delay` to introduce delays in your producer and consumer, simulating real-world processing times. * Monitor the performance of your producer-consumer pair and adjust the configuration as needed to achieve optimal throughput and latency.

```csharp using System; using System.Threading; using System.Threading.Channels; class ProducerConsumer { public static async Task RunProducerConsumer() { // Create a channel with bounded capacity var channel = Channel.CreateBounded<int>(new BoundedChannelOptions(50) { FullMode = BoundedChannelFullMode.Wait }); // Start the producer _ = ProduceWorkItems(channel.Writer, 100); // Start the consumer await ConsumeWorkItems(channel.Reader); } private static async Task ProduceWorkItems(ChannelWriter<int> writer, int rate) { int item = 0; while (true) { await writer.WriteAsync(item); item++; await Task.Delay(TimeSpan.FromSeconds(1.0 / rate)); } } private static async Task ConsumeWorkItems(ChannelReader<int> reader) { while (await reader.WaitToReadAsync()) { while (reader.TryRead(out int item)) { Console.WriteLine($"Processing item {item}"); // Simulate processing time await Task.Delay(10); } } } public static void Main(string[] args) { RunProducerConsumer().Wait(); } } ``` This implementation uses `Channel<T>` to create a producer-consumer pair. The producer emits work items at a rate of 100 per second, and the consumer processes them with bounded backpressure, allowing a maximum of 50 in-flight items. The `ProduceWorkItems` method writes items to the channel, and the `ConsumeWorkItems` method reads and processes items from the channel. The `BoundedChannelOptions` class is used to configure the channel with a bounded capacity and a full mode of `Wait`, which causes the producer to wait when the channel is full. This implementation requires additional explanation due to the complexity of the producer-consumer pattern and the use of asynchronous programming.