Concurrent collections and channels

Eight threads counted 800,000 page views into one ConcurrentDictionary, and the code that read a count and then wrote it back recorded only 680,274 of them. The C# concurrent collections make each call safe, not two calls in a row.

You will update shared data with one call per change, pass work from producers to consumers through channels, and bound a channel so that a fast producer waits instead of filling memory.

Concurrent collections make each call atomic, not a sequence of calls

System.Collections.Concurrent holds collections that many threads can use at once without a lock of your own. counts.cs lets eight threads count 100,000 page views each, spread over four pages, into two dictionaries: one updated with TryGetValue and then an indexer set, the other with AddOrUpdate.

using System.Collections.Concurrent;

// Eight threads each count 100,000 page views, spread over four pages, into one ConcurrentDictionary.
string[] pages = ["/home", "/cart", "/pay", "/help"];
const int Threads = 8;
const int ViewsPerThread = 100_000;

var checkThenSet = new ConcurrentDictionary<string, int>();
var addOrUpdate = new ConcurrentDictionary<string, int>();

RunThreads(i =>
{
    string page = pages[i % pages.Length];

    // Two safe calls in a row are not one safe operation: another thread can update the page in between.
    if (checkThenSet.TryGetValue(page, out int views))
    {
        checkThenSet[page] = views + 1;
    }
    else
    {
        checkThenSet[page] = 1;
    }

    // One call that reads, adds and stores for its key, and tries again if another thread got there first.
    addOrUpdate.AddOrUpdate(page, 1, (_, current) => current + 1);
});

Console.WriteLine($"expected:                {Threads * ViewsPerThread:N0} views");
Console.WriteLine($"TryGetValue, then set:   {checkThenSet.Values.Sum():N0} views");
Console.WriteLine($"AddOrUpdate:             {addOrUpdate.Values.Sum():N0} views");
foreach (string page in pages)
{
    Console.WriteLine($"  {page,-6} {addOrUpdate[page],7:N0}");
}

void RunThreads(Action<int> view)
{
    using var start = new Barrier(Threads);
    var threads = new Thread[Threads];
    for (int t = 0; t < Threads; t++)
    {
        threads[t] = new Thread(() =>
        {
            start.SignalAndWait();
            for (int i = 0; i < ViewsPerThread; i++)
            {
                view(i);
            }
        });
        threads[t].Start();
    }

    foreach (Thread thread in threads)
    {
        thread.Join();
    }
}

expected:                800,000 views
TryGetValue, then set:   680,274 views
AddOrUpdate:             800,000 views
  /home  200,000
  /cart  200,000
  /pay   200,000
  /help  200,000

Both dictionaries stayed intact: no exception, no corrupted table. Yet the first lost 119,726 views, because another thread could change a page’s count between the read and the write, the lost update of the lesson on locks. AddOrUpdate reads, computes and stores in one call, and tries again when another thread got there first, so it counted all 800,000. The C# concurrent collections give you one atomic call per operation; an update that takes two calls is still a race. The collections, and what each one is for:

Microsoft’s guide to choosing a thread-safe collection adds the trade-offs: a Dictionary that is only read needs no synchronisation at all, and ConcurrentBag does best when the threads that add items are also the ones that take them. ConcurrentQueue<T> is the plain hand-over in order: Enqueue adds an item at the back, and TryDequeue takes the one at the front. When the queue is empty, TryDequeue returns false at once, so a consumer that must wait for work needs something to wait on, which is what a channel adds. One trap stays inside ConcurrentDictionary itself. GetOrAdd runs its value factory outside its locks, so under contention the factory can run more than once, though only one value is stored.

A bounded channel passes items along and makes a fast producer wait

A channelchannel: A Channel<T> from System.Threading.Channels: a first-in, first-out queue that producers write to and consumers read from, both with await. A bounded channel holds at most its capacity; an unbounded one accepts every write., from System.Threading.Channels, is a first-in, first-out queue with two ends: producers write to channel.Writer, consumers read from channel.Reader, and both sides wait with await, without blocking a thread. Channel.CreateUnbounded<T>() accepts every write at once. Channel.CreateBounded<T>(capacity) holds at most capacity items, and its full modefull mode: The setting of a bounded channel that decides what a write does when the channel is full. Wait, the default, makes WriteAsync wait for room; DropOldest, DropNewest and DropWrite throw an item away instead. decides what a write does when it is full. bounded.cs fills a channel of capacity 2 and reads it out:

using System.Threading.Channels;

// A channel that holds at most 2 orders. The writer is the producer's end, the reader the consumer's.
Channel<int> orders = Channel.CreateBounded<int>(2);

Console.WriteLine($"TryWrite(1): {orders.Writer.TryWrite(1)}");
Console.WriteLine($"TryWrite(2): {orders.Writer.TryWrite(2)}");
Console.WriteLine($"TryWrite(3): {orders.Writer.TryWrite(3)}, the channel is full");

ValueTask write3 = orders.Writer.WriteAsync(3);
Console.WriteLine($"WriteAsync(3) completed: {write3.IsCompleted}, it waits for room");

int first = await orders.Reader.ReadAsync();
Console.WriteLine($"ReadAsync() took order {first}; WriteAsync(3) completed: {write3.IsCompleted}");
await write3;

orders.Writer.Complete(); // no more orders: readers finish once the channel is empty
await foreach (int order in orders.Reader.ReadAllAsync())
{
    Console.WriteLine($"ReadAllAsync() took order {order}");
}

Console.WriteLine($"ReadAllAsync() ended; Reader.Completion is {orders.Reader.Completion.Status}");

// A producer and a slower consumer: the producer can never get more than 2 orders ahead.
Channel<int> pipe = Channel.CreateBounded<int>(2);
int maxWaiting = 0;
Task producer = Task.Run(async () =>
{
    for (int order = 1; order <= 5; order++)
    {
        await pipe.Writer.WriteAsync(order);
        maxWaiting = Math.Max(maxWaiting, pipe.Reader.Count);
    }

    pipe.Writer.Complete();
});

int total = 0;
await foreach (int order in pipe.Reader.ReadAllAsync())
{
    await Task.Delay(20); // the consumer takes 20 ms an order
    total += order;
}

await producer;
Console.WriteLine($"orders 1 to 5 added up to {total}; at most {maxWaiting} orders ever waited in the channel");

TryWrite(1): True
TryWrite(2): True
TryWrite(3): False, the channel is full
WriteAsync(3) completed: False, it waits for room
ReadAsync() took order 1; WriteAsync(3) completed: True
ReadAllAsync() took order 2
ReadAllAsync() took order 3
ReadAllAsync() ended; Reader.Completion is RanToCompletion
orders 1 to 5 added up to 15; at most 2 orders ever waited in the channel

The third TryWrite returned false, and WriteAsync(3) did not complete until ReadAsync had taken order 1: the producer waited for room. That is back-pressureback-pressure: A consumer setting the pace of its producer. An async stream has it built in: the producer runs only when the consumer asks for the next item, so items never pile up between them. More in “Async streams with IAsyncEnumerable”, the consumer setting the pace of its producer. Step through it:

Writer.Complete() says that no more items will come. Items written before it are still read, and ReadAllAsync ends only when the channel is complete and empty. In the second half of bounded.cs, a producer of five orders met a consumer that takes 20 ms an order: the orders added up to 15, and the channel never held more than 2.

The default full mode, Wait, loses nothing. The other three drop an item instead of waiting. DropOldest removes the oldest item to make room, DropNewest removes the newest item already in the channel, and DropWrite drops the item being written. SingleReader and SingleWriter in a channel’s options promise one reader or one writer, which Microsoft’s channels article says lets the channel work more efficiently.

GetOrAdd and AddOrUpdate under contention, measured Pro: opens with a Pro, Premium or Ultimate pass

Full modes and back-pressure, measured Pro: opens with a Pro, Premium or Ultimate pass

A lab: a log pipeline that fails and stops cleanly Pro: opens with a Pro, Premium or Ultimate pass

Practise

The self-check covers the first two sections: one atomic call per update, and bounded channels with their completion. It starts where your answers so far were least sure:

Interview questions

Fresher · What does a bounded channel do when it is full?

That depends on its full mode, chosen when you create it. In the default mode, Wait, nothing is lost: TryWrite returns false at once, and WriteAsync waits until a consumer reads an item and makes room, without blocking a thread. That wait is the channel’s back-pressure: a producer can never get more than the capacity ahead of its consumer, so memory stays bounded. The three drop modes never make the producer wait and throw an item away instead: DropOldest the oldest item in the channel, DropNewest the newest item in it, and DropWrite the item being written. Choose Wait for data that must all be processed, such as orders, and a drop mode only where losing some is acceptable, such as live readings.

Key takeaways

• A concurrent collection makes each call atomic, so a read and a write in two calls still lose updates: counts.cs lost 119,726 of 800,000 views.
• Change shared values with one call that does the whole update, such as AddOrUpdate, GetOrAdd or TryUpdate.
• A bounded channel holds at most its capacity, and in the default Wait mode a full channel makes WriteAsync wait.
• Writer.Complete ends the readers once the channel is empty, and a producer that never completes leaves them waiting.