Channels and concurrent

C# course · Expert · Lesson 45/47

Write a parallel pipeline with channels and concurrent collections.

Channels and concurrent

In this lesson we pipe logs through a Channel and count them in a ConcurrentDictionary. The producer writes each log with WriteAsync and closes with Complete; the consumer reads each log with ReadAllAsync and counts with AddOrUpdate.

This also uses ParseWithSpan from the previous lesson: each log's level is extracted without copy. Results are sorted alphabetically and come out deterministic.

Run method design

The Run method is declared as async Task. Inside, an unbounded channel and a ConcurrentDictionary are created. Logs are written one by one, the channel is completed, then consumed with await foreach.

Each log's level is extracted with ParseWithSpan and counted with AddOrUpdate(level, 1, (_, n) => n + 1). Results are printed alphabetically with OrderBy(k => k.Key).

Exercise Add the Run method

Add public static async System.Threading.Tasks.Task Run(string[] logs) to LogProcessor. Inside, create a channel with System.Threading.Channels.Channel.CreateUnbounded<string>() and a dictionary with new System.Collections.Concurrent.ConcurrentDictionary<string, int>(). Write logs with await channel.Writer.WriteAsync(log), call channel.Writer.Complete(). With await foreach (var log in channel.Reader.ReadAllAsync()), extract the level with ParseWithSpan(log.AsSpan()) and count with counts.AddOrUpdate(level, 1, (_, n) => n + 1). Print with OrderBy(k => k.Key). In Main, call LogProcessor.Run(logs).Wait().

Show example solution

LogProcessor.Run(logs).Wait();
Example solution:
using System;
using System.Linq;

static class LogProcessor
{
    public static int ParseLevel(string level)
    {
        return level switch
        {
            "INFO" => 0,
            "WARN" => 1,
            "ERROR" => 2,
            _ => -1
        };
    }

    public static int CountByLevel(string[] logs, int targetLevel)
    {
        int count = 0;
        foreach (var log in logs)
        {
            if (ParseLevel(log.Split(':')[0]) == targetLevel)
            {
                count++;
            }
        }
        return count;
    }

    public static string ParseWithSpan(ReadOnlySpan<char> log)
    {
        int colon = log.IndexOf(':');
        if (colon < 0) return "";
        return log.Slice(0, colon).ToString();
    }

    public static unsafe int UnsafeSum(int a, int b)
    {
        int* pa = &a;
        int* pb = &b;
        return *pa + *pb;
    }

    public static async System.Threading.Tasks.Task Run(string[] logs)
    {
        var channel = System.Threading.Channels.Channel.CreateUnbounded<string>();
        var counts = new System.Collections.Concurrent.ConcurrentDictionary<string, int>();
        foreach (var log in logs)
        {
            await channel.Writer.WriteAsync(log);
        }
        channel.Writer.Complete();
        await foreach (var log in channel.Reader.ReadAllAsync())
        {
            var level = ParseWithSpan(log.AsSpan());
            counts.AddOrUpdate(level, 1, (_, n) => n + 1);
        }
        foreach (var kvp in counts.OrderBy(k => k.Key))
        {
            Console.WriteLine($"{kvp.Key}: {kvp.Value}");
        }
    }
}

class Program
{
    static void Main()
    {
        Console.WriteLine("LogProcessor skeleton ready");
        var logs = new[] { "INFO: a", "WARN: b", "INFO: c", "ERROR: d" };
        Console.WriteLine(LogProcessor.CountByLevel(logs, 0));
        Console.WriteLine(LogProcessor.ParseWithSpan("ERROR: disk".AsSpan()));
        Console.WriteLine(LogProcessor.UnsafeSum(3, 4));
        LogProcessor.Run(logs).Wait();
    }
}

Test yourself

Match each Channel flow step

CreateUnbounded

Create the channel
WriteAsync

Produce a log
ReadAllAsync

Consume logs

Complete the AddOrUpdate call

counts.____(level, 1, (_, n) => n + 1);

Show answer

counts.AddOrUpdate(level, 1, (_, n) => n + 1);

Which method closes the channel when production is done?
- Writer.Close()
- Writer.Finish()
- Writer.End()
- Writer.Complete()
Show answer

Writer.Complete() — Correct. Writer.Complete() ends writing.

Channels and concurrent