C#
Channels and concurrent
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.
Next lesson: Source generator
