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 Python this is queue.Queue(maxsize=3). put waits if it is full, and get waits if it is empty. (The queue module is for threads. There is also asyncio.Queue with the same methods for async code.)

See it

The site and the packers

DONE = None




def front(orders, queue, packers):
    for order in orders:
        queue.put(order)  # waits if full
    for _ in range(packers):
        queue.put(DONE)  # one pill per packer




def pack(queue, packed):
    while True:
        order = queue.get()  # waits if empty
        if order is DONE:
            return  # the pill: stop
        packed.append(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 get.
  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: put 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. The site puts one pill per packer, after the orders, so every order is packed first.
  9. Each packer finishes its order, takes a pill and stops. Without the pill it would wait in get 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:

@dataclass(frozen=True)
class Order:
    id: int
    euro: int
    express: bool

The site and the packers:

DONE = None




def front(orders, queue, packers):
    for order in orders:
        queue.put(order)  # waits if full
    for _ in range(packers):
        queue.put(DONE)  # one pill per packer




def pack(queue, packed):
    while True:
        order = queue.get()  # waits if empty
        if order is DONE:
            return  # the pill: stop
        packed.append(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.
orders = [
    Order(1001, 25, False),
    Order(1002, 40, True),
    Order(1003, 15, False),
    Order(1004, 60, False),
    Order(1005, 30, True),
    Order(1006, 45, False),
    Order(1007, 20, True),
    Order(1008, 35, False),
]
queue = Queue(maxsize=3)
packed = []  # list.append is safe from many threads
threads = [threading.Thread(target=front, args=(orders, queue, 2)),
           threading.Thread(target=pack, args=(queue, packed)),
           threading.Thread(target=pack, args=(queue, packed))]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()
report(orders, packed)

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

def report(orders, packed):
    distinct = len({order.id for order in packed})
    print(f"Packed: {len(packed)} of {len(orders)}")
    print("Each order once:", "yes" if distinct == len(orders) else "no")
    for express in (True, False):
        group = [order for order in packed if order.express == express]
        kind = "Express" if express else "Standard"
        print(f"{kind}: {len(group)} orders, {sum(o.euro for o in group)} €")
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 get() until an order comes. If the site just finishes, the packers wait forever and the program never ends.

So at the end the site puts a poison pill: a special value that is not an order. A packer that takes the pill stops. There is one pill per packer: each takes only one and does not put it back.

The pills go into the queue after the orders, so every order is packed before them.

Here the pill is DONE = None, and is DONE compares it by identity.

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:

put_nowait(order) raises queue.Full if there is no place. put(order, timeout=1) waits at most a second and then raises Full.

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?

def offer(queue, item):
    try:
        queue.put_nowait(item)
        return True
    except Full:
        return False




queue = Queue(maxsize=2)
print(offer(queue, "1001"))
print(offer(queue, "1002"))
print(offer(queue, "1003"))
print(queue.get())
print(offer(queue, "1003"))
print(queue.qsize())
Answer

Try it

Fill in

How many pills does the site put?
for _ in range():
    queue.put(DONE)

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

Queue() without maxsize has no limit: put never waits and the problem from the start of the lesson is back. In async code the queue is asyncio.Queue(maxsize=3) with await queue.put(...) and await queue.get(): the same thing, but in one thread.

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 (one per packer) 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 put(order, timeout=0.01). Count the refused orders and print "accepted + refused = 12".
  • By kind. Express orders must be packed first. Does queue.PriorityQueue 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 ·