Round #704

Add a Channel<T>-based bounded producer-consumer with backpressure, using BoundedChannelFullMode.Wait and a CancellationToken end-to-end.

Here's an example implementation of a Channel<T>-based bounded producer-consumer with backpressure, using BoundedChannelFullMode.Wait and a CancellationToken: ```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 }); // Create a cancellation token source var cts = new CancellationTokenSource(); // Start the producer _ = ProduceAsync(channel.Writer, cts.Token); // Start the consumer await ConsumeAsync(channel.Reader, cts.Token); // Cancel the producer cts.Cancel(); } private async Task ProduceAsync(ChannelWriter<int> writer, CancellationToken cancellationToken) { for (int i = 0; i < 100; i++) { try { // Write to the channel, waiting if it's full await writer.WriteAsync(i, cancellationToken); Console.WriteLine($"Produced {i}"); } catch (OperationCanceledException) { Console.WriteLine("Producer cancelled"); break; } } // Complete the writer writer.Complete(); } private async Task ConsumeAsync(ChannelReader<int> reader, CancellationToken cancellationToken) { while (await reader.WaitToReadAsync(cancellationToken)) { while (reader.TryRead(out var item)) { Console.WriteLine($"Consumed {item}"); // Simulate some work await Task.Delay(10, cancellationToken); } } // Check if the channel was completed if (reader.Completion.IsCompleted) { if (reader.Completion.IsCanceled) { Console.WriteLine("Consumer cancelled"); } else if (reader.Completion.IsFaulted) { Console.WriteLine("Consumer faulted: " + reader.Completion.Exception); } else { Console.WriteLine("Consumer completed"); } } } } class Program { static async Task Main(string[] args) { var example = new ProducerConsumerExample(); await example.RunExampleAsync(); } } ``` In this example: * We create a bounded channel with a capacity of 10 items and a full mode of `BoundedChannelFullMode.Wait`. This means that when the channel is full, writers will wait until space becomes available. * We start a producer that writes 100 items to the channel, waiting if it's full. * We start a consumer that reads items from the channel, waiting if it's empty. * We use a `CancellationToken` to cancel the producer and consumer when the example is complete. * The producer and consumer both handle cancellation and completion of the channel. This example demonstrates a basic producer-consumer pattern with backpressure, using `Channel<T>` and `CancellationToken` to manage the flow of data and cancellation.

Here's a fully implemented example of a bounded producer-consumer using `Channel<T>` with backpressure and cancellation support: ```csharp using System; using System.Threading; using System.Threading.Channels; using System.Threading.Tasks; public class BoundedProducerConsumer<T> { private readonly Channel<T> _channel; private readonly CancellationTokenSource _cts; public BoundedProducerConsumer(int capacity) { _channel = Channel.CreateBounded<T>(new BoundedChannelOptions(capacity) { FullMode = BoundedChannelFullMode.Wait }); _cts = new CancellationTokenSource(); } public async Task ProduceAsync(T item, CancellationToken cancellationToken = default) { var combinedCancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_cts.Token, cancellationToken).Token; await _channel.Writer.WriteAsync(item, combinedCancellationToken); } public async Task<T> ConsumeAsync(CancellationToken cancellationToken = default) { var combinedCancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_cts.Token, cancellationToken).Token; return await _channel.Reader.ReadAsync(combinedCancellationToken); } public void Cancel() { _cts.Cancel(); } public async Task CompletionAsync() { await _channel.Writer.CompleteAsync(); await _channel.Reader.Completion; } } class Program { static async Task Main(string[] args) { var producerConsumer = new BoundedProducerConsumer<int>(5); // Start a producer task _ = Task.Run(async () => { for (int i = 0; i < 10; i++) { await producerConsumer.ProduceAsync(i); Console.WriteLine($"Produced {i}"); await Task.Delay(100); } await producerConsumer.CompletionAsync(); }); // Start a consumer task _ = Task.Run(async () => { try { while (true) { var item = await producerConsumer.ConsumeAsync(); Console.WriteLine($"Consumed {item}"); } } catch (OperationCanceledException) { Console.WriteLine("Consumer cancelled"); } }); // Wait for 2 seconds and then cancel the producer-consumer await Task.Delay(2000); producerConsumer.Cancel(); // Wait for the tasks to complete await Task.Delay(1000); } } ``` This implementation provides a `BoundedProducerConsumer<T>` class that uses a `Channel<T>` to manage the production and consumption of items. The `ProduceAsync` method writes items to the channel, and the `ConsumeAsync` method reads items from the channel. The `Cancel` method cancels the producer-consumer, and the `CompletionAsync` method waits for the channel to complete. The example usage in the `Main` method demonstrates how to use the `BoundedProducerConsumer<T>` class to produce and consume items concurrently, with cancellation support. The producer task produces 10 items, and the consumer task consumes items until it is cancelled. The `Cancel` method is called after 2 seconds to cancel the producer-consumer.