Putting it together

C# course · Expert · Lesson 47/47

Combine all expert topics into a working mini app.

Putting it together

In the final lesson we combine all pieces: zero-copy level parsing with Span, a producer-consumer flow with Channel, a source generator simulation for DTOs, and now multi-threaded counting with Parallel.ForEach.

The ProcessWithThreads method processes all logs in parallel and keeps an atomic counter with Interlocked.Increment. Main now calls every piece in sequence.

The ProcessWithThreads method

ProcessWithThreads resets a static int processed field, processes each log in parallel with Parallel.ForEach, and counts atomically with Interlocked.Increment(ref processed). Then it prints the total.

Because processed is a static field, ref can be used safely from inside a lambda. The output is deterministic: with four logs the result is four.

Exercise Add ProcessWithThreads and integrate

Add a private static int processed; field and a public static void ProcessWithThreads(string[] logs) method to LogProcessor. In the body, set processed = 0;, call System.Threading.Tasks.Parallel.ForEach(logs, _ => { System.Threading.Interlocked.Increment(ref processed); }), then Console.WriteLine($"processed: {processed}"). In Main, call LogProcessor.ProcessWithThreads(logs) at the end.

Show example solution

LogProcessor.ProcessWithThreads(logs);
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}");
        }
        LogProcessor.ProcessWithThreads(logs);
    }
}

static class DtoGenerator
{
    public static System.Collections.Generic.List<(string Hint, string Source)> Sources { get; } = new();

    public static void AddSource(string hintName, string source)
    {
        Sources.Add((hintName, source));
    }

    public static void Generate(string[] typeNames)
    {
        foreach (var name in typeNames)
        {
            AddSource(name + ".g.cs", $"public class {name}Dto {{ }}");
        }
    }

    private static int processed;

    public static void ProcessWithThreads(string[] logs)
    {
        processed = 0;
        System.Threading.Tasks.Parallel.ForEach(logs, _ =>
        {
            System.Threading.Interlocked.Increment(ref processed);
        });
        Console.WriteLine($"processed: {processed}");
    }
}

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();
        DtoGenerator.Generate(new[] { "Info", "Warn", "Error" });
        foreach (var entry in DtoGenerator.Sources)
        {
            Console.WriteLine(entry.Hint);
        }
    }
}

Test yourself

Match each expert topic

Span<T>

Memory-efficient parsing
Channel<T>

Parallel flow
Parallel.ForEach

Multi-threading
Which class is used to count all processed logs in the project?
- Interlocked (atomic counter)
- StringBuilder
- DateTime
- Random
Show answer

Interlocked (atomic counter) — Correct. Interlocked does atomic increment.

Complete the Parallel.ForEach call

Parallel.____(logs, _ => Interlocked.Increment(ref processed));

Show answer

Parallel.ForEach(logs, _ => Interlocked.Increment(ref processed));

Putting it together