Skip to content
wannacode

Producer and consumer

In short: Producer-consumer is a pattern in which some threads put tasks into a bounded queue and others take them out; a full queue makes the producer wait (backpressure).

Problem

Black Friday

At 10:00 the sale starts and the site takes 50 orders a second. The warehouse packs 10. There are three bad ways to deal with that.

The site packs by itself. The customer clicks "Order" and waits until somebody packs the box. The site is as fast as the warehouse.

The site puts the orders in a plain list and the warehouse takes them from there. Two packers can take the same order and another one can get lost: that is a race condition. If you add a lock, a packer with an empty list has to check again and again whether something has come.

The list has no limit. The site is happy: it never waits. But every second the list grows by 40 orders. After an hour it holds 144 000, the last one has waited an hour, and memory runs out.

The idea

A queue with limited room

A queue goes between the two sides. The producer puts tasks at its back, the consumers take them from the front. This is the producer and consumer pattern. The two sides do not know each other: the site does not know how many packers there are, and the packers do not know where the orders come from.

The queue is blocking and has limited room:

  • if it is empty, the consumer waits, without spinning the processor, until something comes;
  • if it is full, the producer waits until a consumer frees a place;
  • the lock is inside the queue: every order goes to exactly one consumer.
In C# this is BlockingCollection<T> with boundedCapacity: 3. Add waits if it is full. GetConsumingEnumerable() is a loop that takes while there is something and waits if it is empty. It lives in System.Collections.Concurrent, like ConcurrentBag, where we collect the packed orders.

See it

The site and the packers

static void Front(List<Order> orders,
                  BlockingCollection<Order> queue)
{
    foreach (var order in orders)
    {
        queue.Add(order); // waits if full
    }
    queue.CompleteAdding(); // no more: the packers stop
}


static void Pack(BlockingCollection<Order> queue,
                 ConcurrentBag<Order> packed)
{
    // waits if empty
    foreach (var order in queue.GetConsumingEnumerable())
    {
        packed.Add(order);
    }
}
Step by step, in words
  1. The site (producer) puts orders in a queue with 3 places; two packers (consumers) take them from the front. It is empty, so they wait in GetConsumingEnumerable.
  2. The site puts 1001 and 1002 at the back of the queue.
  3. Each packer takes the order at the front and packs it. Packing is slower than taking orders.
  4. The site puts 1003, 1004 and 1005. All 3 places are taken.
  5. 1006 does not fit: Add waits, and the site stops taking orders. This is backpressure: the slow side holds back the fast one.
  6. Packer 1 is done with 1001 and takes 1003. A place frees up, and the waiting put lets 1006 in.
  7. And so on: the packers take from the front, the site adds at the back, and the queue never holds more than 3.
  8. No more orders. CompleteAdding tells the queue that nothing else will come.
  9. Empty and completed: GetConsumingEnumerable ends and each foreach stops. Without CompleteAdding the packers would wait forever.
  10. 8 of 8, each once: 3 express for 90 €, 5 standard for 180 €. Who packs which changes from run to run; the totals do not.

The order:

record Order(int Id, int Euro, bool Express);

The site and the packers:

static void Front(List<Order> orders,
                  BlockingCollection<Order> queue)
{
    foreach (var order in orders)
    {
        queue.Add(order); // waits if full
    }
    queue.CompleteAdding(); // no more: the packers stop
}


static void Pack(BlockingCollection<Order> queue,
                 ConcurrentBag<Order> packed)
{
    // waits if empty
    foreach (var order in queue.GetConsumingEnumerable())
    {
        packed.Add(order);
    }
}
front and pack run in three threads: one site and two packers. The site puts the eight orders and is done, and the packers take until they get a sign to stop.
var orders = new List<Order>
{
    new(1001, 25, false),
    new(1002, 40, true),
    new(1003, 15, false),
    new(1004, 60, false),
    new(1005, 30, true),
    new(1006, 45, false),
    new(1007, 20, true),
    new(1008, 35, false),
};
var queue = new BlockingCollection<Order>(boundedCapacity: 3);
var packed = new ConcurrentBag<Order>();
var threads = new List<Thread>
{
    new Thread(() => Front(orders, queue)),
    new Thread(() => Pack(queue, packed)),
    new Thread(() => Pack(queue, packed)),
};
foreach (var thread in threads) thread.Start();
foreach (var thread in threads) thread.Join();
Report(orders, packed);

The report checks things that do not depend on which packer took which order:

static void Report(List<Order> orders, ConcurrentBag<Order> packed)
{
    int distinct = packed.Select(o => o.Id).Distinct().Count();
    Console.WriteLine($"Packed: {packed.Count} of {orders.Count}");
    Console.WriteLine($"Each order once: {(distinct == orders.Count ? "yes" : "no")}");
    foreach (var express in new[] { true, false })
    {
        var group = packed.Where(o => o.Express == express).ToList();
        string kind = express ? "Express" : "Standard";
        Console.WriteLine($"{kind}: {group.Count} orders, {group.Sum(o => o.Euro)} €");
    }
}
Output
Packed: 8 of 8
Each order once: yes
Express: 3 orders, 90 €
Standard: 5 orders, 180 €

The output is the same on every run. How many orders each packer took changes, so the program does not print it.

The idea

How the packers stop

The packer waits in GetConsumingEnumerable() until an order comes. If the site just finishes, the packers wait forever and the program never ends.

So at the end the site calls CompleteAdding(): "no more will come". After that Add throws an exception, and GetConsumingEnumerable() ends as soon as the queue is empty. Every foreach over it stops by itself, however many packers there are. The orders already inside are packed first.

The other way, which you will meet in other people’s code, is the poison pill: a special value that is not an order, one per consumer. In C# you do not need it.

The idea

Backpressure

A full queue makes the site wait. That is not a bug, it is the point: the slow side pushes back on the fast one. This is called backpressure. The site cannot get ahead of the warehouse by more than the size of the queue, memory does not grow, and the last order waits at most as long as it takes to pack the 3 in front of it.

Waiting is not the only answer to a full queue. There is also a method that does not wait, but says whether it worked:

TryAdd(order) returns false if there is no place. TryAdd(order, 1000) waits at most a second.

With it the site can show "High demand, try again in a minute" instead of hanging. What matters is that the decision is yours: the order does not disappear silently.

Try it

Predict the output

A queue with 2 places and no packers. What will it print?

var queue = new BlockingCollection<string>(boundedCapacity: 2);
Console.WriteLine(queue.TryAdd("1001"));
Console.WriteLine(queue.TryAdd("1002"));
Console.WriteLine(queue.TryAdd("1003"));
Console.WriteLine(queue.Take());
Console.WriteLine(queue.TryAdd("1003"));
Console.WriteLine(queue.Count);
Answer

Try it

Fill in

What does the site call at the end so that the packers stop?
queue.(); // no more: the packers stop

Try it

Find the bug

A colleague does not want the site to wait and changes one line. Now the program never hangs, but the report sometimes says "Packed: 5 of 8". Which line has the bug?

Pick the line with the bug.

See it

Threads, Workers and messages

In new C# code you will also meet Channel<T> (System.Threading.Channels). Channel.CreateBounded<Order>(3) is the same queue with 3 places, but for async code: await writer.WriteAsync(order) waits without blocking a thread, writer.Complete() is CompleteAdding, and await foreach (var o in reader.ReadAllAsync()) is the consumer. BlockingCollection blocks threads; Channel lets them go while it waits.

Exercise

Two entrances, three packers

Solve the task in your editor.

Orders come from two entrances: the site (2001–2006) and the mobile app (3001–3006). Each entrance is a separate producer, and there are three packers. The queue has 4 places. The pills (CompleteAdding) come only after both entrances are done. Print:

Output
Packed: 12 of 12
Each order once: yes
Total: 325 €

More tasks

  • Refusal. Make the queue 1 place and the site use TryAdd(order, 10). Count the refused orders and print "accepted + refused = 12".
  • By kind. Express orders must be packed first. Does two queues and a packer that looks at the express one first help? What happens to the pills?
  • A chain. Add a second queue after the packers: they put the packed boxes in it, and a courier takes them. How many pills are needed now, and who puts them?

Test

Check yourself

1. The queue between the site and the packers is full. What does put do?
2. Why does the queue have a limit at all? Without one the site never waits.
3. Two sites put orders, three packers take them. How many pills are needed, and when are they put?
4. How do you know the program with the queue is right, when the packing order is different on every run?

Made by Ivo Gergov as part of Evo Tech's campaign for accessible programming education ·