C# Channels Explained: Producer-Consumer Concurrency Done Right

using System;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;

// Demonstrates the split Writer/Reader ownership model and
// shows that a reader parked on an empty channel costs zero threads.
class ChannelInternalsDemo
{
    static async Task Main()
    {
        // Create an unbounded channel — no capacity ceiling.
        // UnboundedChannelOptions lets you tune single-writer / single-reader
        // for extra lock-free optimisation when you know your topology.
        var options = new UnboundedChannelOptions
        {
            SingleWriter = false,  // multiple producers allowed
            SingleReader = true,   // only one consumer — enables faster single-reader path
            AllowSynchronousContinuations = false // keeps continuations off the writer's call stack
        };

        Channel<string> logChannel = Channel.CreateUnbounded<string>(options);

        // Hand ONLY the writer to producer logic — it cannot call ReadAsync.
        ChannelWriter<string> writer = logChannel.Writer;

        // Hand ONLY the reader to consumer logic — it cannot call WriteAsync.
        ChannelReader<string> reader = logChannel.Reader;

        // --- Consumer: runs on its own task, drains the channel ---
        Task consumerTask = Task.Run(async () =>
        {
            // ReadAllAsync returns an IAsyncEnumerable<T>.
            // It yields each item as it arrives and exits cleanly
            // when the writer signals completion (writer.Complete()).
            await foreach (string logEntry in reader.ReadAllAsync())
            {
                Console.WriteLine($"[Consumer] Processed: {logEntry}  | Thread {Thread.CurrentThread.ManagedThreadId}");
                await Task.Delay(50); // simulate processing time
            }
            Console.WriteLine("[Consumer] Channel closed — exiting.");
        });

        // --- Producers: two tasks writing concurrently ---
        Task producerA = Task.Run(async () =>
        {
            for (int i = 1; i <= 5; i++)
            {
                string entry = $"ProducerA-Event-{i}";
                // TryWrite is synchronous and allocation-free when the channel has capacity.
                // Use WriteAsync when the channel might be full (bounded channels).
                bool accepted = writer.TryWrite(entry);
                Console.WriteLine($"[ProducerA] Wrote '{entry}': accepted={accepted}");
                await Task.Delay(30);
            }
        });

        Task producerB = Task.Run(async () =>
        {
            for (int i = 1; i <= 5; i++)
            {
                string entry = $"ProducerB-Event-{i}";
                await writer.WriteAsync(entry); // async overload — awaits if channel is full
                Console.WriteLine($"[ProducerB] Wrote '{entry}'");
                await Task.Delay(45);
            }
        });

        // Wait for both producers to finish writing.
        await Task.WhenAll(producerA, producerB);

        // CRITICAL: signal that no more items will be written.
        // Without this, ReadAllAsync loops forever waiting for more data.
        writer.Complete();
        Console.WriteLine("[Main] Writer completed.");

        // Wait for the consumer to drain everything before exiting.
        await consumerTask;
        Console.WriteLine("[Main] Done.");
    }
}

using System;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;

// Real-world scenario: an order processing pipeline where
// consumers are slow (DB writes) and producers are fast (HTTP intake).
// We use BoundedChannel with Wait mode to apply natural backpressure.
class BoundedChannelBackpressure
{
    static async Task Main()
    {
        // Capacity of 5 simulates a small in-memory buffer.
        // In production this might be 500–5000 depending on item size and latency budget.
        var boundedOptions = new BoundedChannelOptions(capacity: 5)
        {
            FullMode = BoundedChannelFullMode.Wait,   // producers wait — no data loss
            SingleWriter = false,
            SingleReader = false
        };

        Channel<OrderRequest> orderChannel = Channel.CreateBounded<OrderRequest>(boundedOptions);

        using var cancellationSource = new CancellationTokenSource();
        CancellationToken shutdownToken = cancellationSource.Token;

        // --- Two consumers simulating slow DB writers ---
        Task[] consumerTasks = new Task[2];
        for (int consumerId = 1; consumerId <= 2; consumerId++)
        {
            int id = consumerId; // capture for closure
            consumerTasks[id - 1] = Task.Run(async () =>
            {
                try
                {
                    // ReadAllAsync respects cancellation AND channel completion.
                    // It stops when either the token fires or writer.Complete() is called.
                    await foreach (OrderRequest order in
                        orderChannel.Reader.ReadAllAsync(shutdownToken))
                    {
                        Console.WriteLine($"[Consumer-{id}] Processing order #{order.OrderId}");
                        await Task.Delay(200, shutdownToken); // simulate slow DB write
                        Console.WriteLine($"[Consumer-{id}] Saved order #{order.OrderId}");
                    }
                }
                catch (OperationCanceledException)
                {
                    Console.WriteLine($"[Consumer-{id}] Shutdown signal received.");
                }
            });
        }

        // --- Fast producer simulating HTTP request intake ---
        Task producerTask = Task.Run(async () =>
        {
            for (int orderId = 1; orderId <= 15; orderId++)
            {
                var order = new OrderRequest(orderId, $"Item-{orderId}");
                Console.WriteLine($"[Producer] Attempting to queue order #{orderId}" +
                                  $" (channel count: {orderChannel.Reader.Count})");

                // WriteAsync will AWAIT here if channel is full (5 items).
                // This is backpressure in action — the HTTP handler would naturally
                // slow down, preventing memory explosion.
                await orderChannel.Writer.WriteAsync(order, shutdownToken);
                Console.WriteLine($"[Producer] Queued order #{orderId}");

                await Task.Delay(30); // producer is faster than consumers (30ms vs 200ms)
            }

            // Signal no more orders are coming.
            // TryComplete returns false if already completed — safe to call.
            orderChannel.Writer.TryComplete();
            Console.WriteLine("[Producer] All orders submitted. Channel closed.");
        });

        await Task.WhenAll(producerTask);
        await Task.WhenAll(consumerTasks);
        Console.WriteLine("[Main] All orders processed.");
    }
}

// A meaningful domain object — never use primitives for domain concepts in pipelines.
record OrderRequest(int OrderId, string ItemName);

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;

// Three-stage pipeline:
//   Stage 1: Parse raw sensor strings into SensorReading structs (CPU-bound)
//   Stage 2: Enrich with location metadata via simulated DB lookup (I/O-bound)
//   Stage 3: Batch and 'flush' enriched readings (I/O-bound)
class ThreeStagePipeline
{
    static async Task Main()
    {
        using var linkedCts = new CancellationTokenSource();
        CancellationToken pipelineToken = linkedCts.Token;

        // Stage 1 → Stage 2 channel: bounded to apply backpressure if DB is slow
        Channel<SensorReading> parsedReadingsChannel =
            Channel.CreateBounded<SensorReading>(new BoundedChannelOptions(50)
            {
                FullMode = BoundedChannelFullMode.Wait,
                SingleWriter = true,  // only one parser
                SingleReader = false  // multiple enrichers
            });

        // Stage 2 → Stage 3 channel: enriched readings awaiting batching
        Channel<EnrichedReading> enrichedReadingsChannel =
            Channel.CreateBounded<EnrichedReading>(new BoundedChannelOptions(20)
            {
                FullMode = BoundedChannelFullMode.Wait,
                SingleWriter = false, // multiple enrichers write here
                SingleReader = true   // single batcher reads here
            });

        // ── Stage 1: Parser (single task, CPU-bound) ──────────────────────────
        Task parserTask = Task.Run(async () =>
        {
            try
            {
                string[] rawSensorData = GenerateRawSensorData(count: 30);
                foreach (string rawLine in rawSensorData)
                {
                    SensorReading reading = ParseSensorLine(rawLine);
                    await parsedReadingsChannel.Writer.WriteAsync(reading, pipelineToken);
                }
            }
            catch (OperationCanceledException) { /* pipeline cancelled upstream */ }
            finally
            {
                // Always complete in finally — even if an exception occurs.
                // This unblocks downstream stages instead of hanging forever.
                parsedReadingsChannel.Writer.TryComplete();
                Console.WriteLine("[Stage1-Parser] Complete.");
            }
        });

        // ── Stage 2: Enrichers (fan-out — 3 parallel DB lookup workers) ───────
        const int enricherCount = 3;
        Task[] enricherTasks = new Task[enricherCount];
        for (int workerId = 0; workerId < enricherCount; workerId++)
        {
            int id = workerId;
            enricherTasks[id] = Task.Run(async () =>
            {
                try
                {
                    // All enrichers read from the SAME reader — competitive distribution.
                    await foreach (SensorReading reading in
                        parsedReadingsChannel.Reader.ReadAllAsync(pipelineToken))
                    {
                        // Simulate async DB lookup
                        string location = await LookupSensorLocation(reading.SensorId, pipelineToken);
                        var enriched = new EnrichedReading(reading, location);
                        await enrichedReadingsChannel.Writer.WriteAsync(enriched, pipelineToken);
                        Console.WriteLine($"[Stage2-Enricher-{id}] Enriched sensor {reading.SensorId}");
                    }
                }
                catch (OperationCanceledException) { }
            });
        }

        // Close the enriched channel only after ALL enrichers finish.
        // Task.WhenAll ensures we don't close too early.
        Task enricherCompletionTask = Task.Run(async () =>
        {
            await Task.WhenAll(enricherTasks);
            enrichedReadingsChannel.Writer.TryComplete();
            Console.WriteLine("[Stage2] All enrichers done. Enriched channel closed.");
        });

        // ── Stage 3: Batcher (single task — collects into batches of 5) ────────
        Task batcherTask = Task.Run(async () =>
        {
            var batch = new List<EnrichedReading>(capacity: 5);
            try
            {
                await foreach (EnrichedReading enriched in
                    enrichedReadingsChannel.Reader.ReadAllAsync(pipelineToken))
                {
                    batch.Add(enriched);
                    if (batch.Count >= 5)
                    {
                        await FlushBatch(batch, pipelineToken);
                        batch.Clear();
                    }
                }
                // Flush any remaining items after channel closes
                if (batch.Count > 0)
                    await FlushBatch(batch, pipelineToken);
            }
            catch (OperationCanceledException) { }
            Console.WriteLine("[Stage3-Batcher] Complete.");
        });

        await Task.WhenAll(parserTask, enricherCompletionTask, batcherTask);
        Console.WriteLine("[Pipeline] All stages finished.");
    }

    static string[] GenerateRawSensorData(int count)
    {
        var data = new string[count];
        for (int i = 0; i < count; i++)
            data[i] = $"SENSOR_{i % 5}|{22.5 + i * 0.1:F1}|{DateTime.UtcNow:O}";
        return data;
    }

    static SensorReading ParseSensorLine(string raw)
    {
        string[] parts = raw.Split('|');
        return new SensorReading(parts[0], double.Parse(parts[1]), DateTimeOffset.Parse(parts[2]));
    }

    static async Task<string> LookupSensorLocation(string sensorId, CancellationToken ct)
    {
        await Task.Delay(40, ct); // simulate DB round trip
        return sensorId switch
        {
            "SENSOR_0" => "Warehouse-A",
            "SENSOR_1" => "Warehouse-B",
            "SENSOR_2" => "Loading-Dock",
            _          => "Unknown"
        };
    }

    static async Task FlushBatch(List<EnrichedReading> batch, CancellationToken ct)
    {
        await Task.Delay(60, ct); // simulate S3/DB write
        Console.WriteLine($"[Stage3-Batcher] Flushed batch of {batch.Count} readings.");
    }
}

record SensorReading(string SensorId, double Temperature, DateTimeOffset Timestamp);
record EnrichedReading(SensorReading Reading, string Location);

using System;
using System.Diagnostics;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;

// Demonstrates TryRead polling for zero-allocation hot paths,
// and shows how to measure actual throughput on a channel.
class ChannelPerformancePatterns
{
    static async Task Main()
    {
        await DemonstrateTryReadPolling();
        await MeasureChannelThroughput();
    }

    // TryRead pattern: useful when consumer is always faster than producer
    // and you want zero async overhead. Trades CPU for latency.
    static async Task DemonstrateTryReadPolling()
    {
        Channel<int> hotChannel = Channel.CreateUnbounded<int>(
            new UnboundedChannelOptions { SingleWriter = true, SingleReader = true });

        // Producer: writes 10_000 integers as fast as possible
        _ = Task.Run(() =>
        {
            for (int value = 0; value < 10_000; value++)
                hotChannel.Writer.TryWrite(value); // synchronous, allocation-free
            hotChannel.Writer.TryComplete();
        });

        // Consumer: polls with TryRead — no async, no allocation per item.
        // Falls back to async WaitToReadAsync only when channel appears empty.
        long sum = 0;
        int itemsRead = 0;
        while (await hotChannel.Reader.WaitToReadAsync()) // async only when empty
        {
            // Drain everything currently available without yielding.
            while (hotChannel.Reader.TryRead(out int value))
            {
                sum += value;  // actual work
                itemsRead++;
            }
        }
        Console.WriteLine($"[TryRead] Read {itemsRead} items. Sum={sum}");
    }

    // Throughput benchmark: how many messages/sec can a bounded channel sustain?
    static async Task MeasureChannelThroughput()
    {
        const int messageCount = 500_000;
        Channel<int> benchmarkChannel = Channel.CreateBounded<int>(
            new BoundedChannelOptions(1_000)
            {
                FullMode    = BoundedChannelFullMode.Wait,
                SingleWriter = true,
                SingleReader = true
            });

        var stopwatch = Stopwatch.StartNew();

        Task producerTask = Task.Run(async () =>
        {
            for (int i = 0; i < messageCount; i++)
                await benchmarkChannel.Writer.WriteAsync(i);
            benchmarkChannel.Writer.TryComplete();
        });

        Task consumerTask = Task.Run(async () =>
        {
            int consumed = 0;
            // WaitToReadAsync + TryRead combo: maximises throughput by batching
            // synchronous drains under the single async suspension point.
            while (await benchmarkChannel.Reader.WaitToReadAsync())
                while (benchmarkChannel.Reader.TryRead(out _))
                    consumed++;
            Console.WriteLine($"[Benchmark] Consumed {consumed} messages.");
        });

        await Task.WhenAll(producerTask, consumerTask);
        stopwatch.Stop();

        double throughput = messageCount / stopwatch.Elapsed.TotalSeconds;
        Console.WriteLine($"[Benchmark] Throughput: {throughput:N0} messages/sec");
        Console.WriteLine($"[Benchmark] Total time: {stopwatch.ElapsedMilliseconds}ms");
    }
}