Threads, Locks and Parallel Work

async and parallelism are different problems. async is about not holding a thread while waiting for something else; parallelism is about using several cores at once. This lesson is the second one.

Two threads writing the same field with nothing arranged between them is a data race, and the result is not merely a wrong number: without a memory barrier one thread may never see the other's write at all.

int counter = 0;
var threads = new Thread[4];

for (int i = 0; i < threads.Length; i++)
{
    threads[i] = new Thread(() =>
    {
        for (int n = 0; n < 100_000; n++) counter++;    // read, add, write
    });
    threads[i].Start();
}
foreach (Thread thread in threads) thread.Join();

Console.WriteLine($"counter = {counter} (expected 400000)");
Console.WriteLine("usually short, and never reliably right");

The three ways to make it correct

int guarded = 0;
int atomic = 0;
object gate = new();

var threads = new Thread[4];
for (int i = 0; i < threads.Length; i++)
{
    threads[i] = new Thread(() =>
    {
        for (int n = 0; n < 50_000; n++)
        {
            lock (gate) { guarded++; }            // mutual exclusion + barriers
            Interlocked.Increment(ref atomic);    // one hardware instruction
        }
    });
    threads[i].Start();
}
foreach (Thread thread in threads) thread.Join();

Console.WriteLine($"lock:        {guarded}");
Console.WriteLine($"Interlocked: {atomic}");
Console.WriteLine("and the third way: share nothing that changes");

Lock on a private readonly object, never on this or on a type - anything another piece of code can also lock is a deadlock waiting to be assembled.

Do not create threads by hand

Task.Run queues work on the thread pool, which already knows how many cores there are. Parallel.ForEach and AsParallel go further and split a collection across them.

var clock = System.Diagnostics.Stopwatch.StartNew();

int[] input = Enumerable.Range(1, 12).ToArray();

// Sequential.
long sequential = input.Select(Work).Sum();
long sequentialMs = clock.ElapsedMilliseconds;

clock.Restart();

// Parallel LINQ: the same query, spread across cores.
long parallel = input.AsParallel().Select(Work).Sum();

Console.WriteLine($"sequential {sequentialMs}ms -> {sequential}");
Console.WriteLine($"parallel   {clock.ElapsedMilliseconds}ms -> {parallel}");
Console.WriteLine($"same answer: {sequential == parallel}");

static long Work(int n)
{
    long total = 0;
    for (int i = 0; i < 3_000_000; i++) total += (i % 7) * n;
    return total;
}

Two rules for AsParallel and Parallel.ForEach: the work must be CPU-bound (for I/O, use async - a parked thread is not a busy core), and the body must not touch shared mutable state. The second is where people come unstuck.

// Wrong: a List<T> is not thread-safe, and this silently loses or corrupts items.
var unsafeList = new List<int>();
Parallel.For(0, 1000, i => { lock (unsafeList) { unsafeList.Add(i); } });
Console.WriteLine($"with a lock around every add: {unsafeList.Count} (correct, and slow)");

// Right: let the query build the result, with no shared state at all.
List<int> collected = Enumerable.Range(0, 1000).AsParallel().Select(i => i).ToList();
Console.WriteLine($"no sharing: {collected.Count}");

// Or use a collection designed for it.
var bag = new System.Collections.Concurrent.ConcurrentBag<int>();
Parallel.For(0, 1000, bag.Add);
Console.WriteLine($"ConcurrentBag: {bag.Count}");

Channels: a queue between producers and consumers

System.Threading.Channels is the modern way to hand work from one part of a program to another without sharing a collection.

using System.Threading.Channels;

Channel<int> channel = Channel.CreateBounded<int>(capacity: 4);

Task producer = Task.Run(async () =>
{
    for (int i = 1; i <= 10; i++)
    {
        await channel.Writer.WriteAsync(i);      // blocks when the buffer is full,
    }                                            // which is backpressure, free
    channel.Writer.Complete();                   // without this, the reader waits forever
});

int total = 0;
await foreach (int value in channel.Reader.ReadAllAsync())
{
    total += value;
}

await producer;
Console.WriteLine($"consumed, total {total}");

The bounded capacity is the part worth noticing: a producer faster than its consumer will wait rather than filling memory.

What to take away

• async is about not holding threads; parallelism is about using cores. Different problems.
• An unsynchronised shared field is a data race - wrong values, and stale reads.
• lock on a private readonly object; Interlocked for a single counter.
• Prefer Task.Run, Parallel.ForEach and AsParallel to creating threads.
• Parallel work must be CPU-bound and must not touch shared mutable state.
• A bounded Channel<T> gives you a producer/consumer queue with backpressure.