Skip to content
wannacode

Producer and consumer

In short: Producer-consumer is a pattern in which some parts of a program put tasks into a bounded queue and others take them out; in JavaScript they are async functions on one thread, and a full queue makes the producer wait.

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.
JavaScript has no such queue ready-made. In one thread with async functions it takes fewer than 40 lines with promises: put and take return a promise that is fulfilled when there is a place or an order. No lock is needed, because the code between two awaits cannot be interrupted.
BoundedQueue
class BoundedQueue {
  #items = [];
  #putters = []; // those waiting to put
  #takers = []; // those waiting to take


  constructor(capacity) {
    this.capacity = capacity;
  }


  get size() {
    return this.#items.length;
  }


  async put(item) {
    while (this.#items.length >= this.capacity) {
      await new Promise((resolve) => this.#putters.push(resolve));
    }
    this.#items.push(item);
    this.#takers.shift()?.();
  }


  tryPut(item) {
    if (this.#items.length >= this.capacity) return false;
    this.#items.push(item);
    this.#takers.shift()?.();
    return true;
  }


  async take() {
    while (this.#items.length === 0) {
      await new Promise((resolve) => this.#takers.push(resolve));
    }
    const item = this.#items.shift();
    this.#putters.shift()?.();
    return item;
  }
}

#putters and #takers are queues of resolve functions: who waits for a place and who waits for an order. take wakes one waiting producer, put wakes one waiting consumer. The one that wakes up checks again (while), because somebody else may have been faster.

See it

The site and the packers

const DONE = null;


async function front(orders, queue, packers) {
  for (const order of orders) {
    await queue.put(order); // waits if full
  }
  for (let i = 0; i < packers; i++) {
    await queue.put(DONE); // one pill per packer
  }
}


async function pack(queue, packed) {
  while (true) {
    const order = await queue.take(); // waits if empty
    if (order === DONE) return; // the pill: stop
    packed.push(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 await take.
  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: await 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 await take 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:

// an order is { id, euro, express }

The site and the packers:

const DONE = null;


async function front(orders, queue, packers) {
  for (const order of orders) {
    await queue.put(order); // waits if full
  }
  for (let i = 0; i < packers; i++) {
    await queue.put(DONE); // one pill per packer
  }
}


async function pack(queue, packed) {
  while (true) {
    const order = await queue.take(); // waits if empty
    if (order === DONE) return; // the pill: stop
    packed.push(order);
  }
}
front and the two packs are three async functions in one thread. await is the place where each can wait and let the others run. This is taking turns, not working in parallel: for packing that computes a lot you need Workers (see below).
const orders = [
  { id: 1001, euro: 25, express: false },
  { id: 1002, euro: 40, express: true },
  { id: 1003, euro: 15, express: false },
  { id: 1004, euro: 60, express: false },
  { id: 1005, euro: 30, express: true },
  { id: 1006, euro: 45, express: false },
  { id: 1007, euro: 20, express: true },
  { id: 1008, euro: 35, express: false },
];
const queue = new BoundedQueue(3);
const packed = [];
await Promise.all([
  front(orders, queue, 2),
  pack(queue, packed),
  pack(queue, packed),
]);
report(orders, packed);

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

function report(orders, packed) {
  const distinct = new Set(packed.map((order) => order.id)).size;
  console.log(`Packed: ${packed.length} of ${orders.length}`);
  console.log(`Each order once: ${distinct === orders.length ? "yes" : "no"}`);
  for (const express of [true, false]) {
    const group = packed.filter((order) => order.express === express);
    const euro = group.reduce((sum, order) => sum + order.euro, 0);
    const kind = express ? "Express" : "Standard";
    console.log(`${kind}: ${group.length} orders, ${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 await queue.take() 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 = null.

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:

tryPut(order) in our queue returns false if there is no place.

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?

const queue = new BoundedQueue(2);
console.log(queue.tryPut("1001"));
console.log(queue.tryPut("1002"));
console.log(queue.tryPut("1003"));
console.log(await queue.take());
console.log(queue.tryPut("1003"));
console.log(queue.size);
Answer

Try it

Fill in

How many pills does the site put?
for (let i = 0; i < ; i++) {
  await 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

When packing computes a lot, you need real threads: Workers. There is no queue of shared objects between them. The main thread sends each order to a Worker with postMessage, and the Worker receives it in on("message"):

import { Worker, isMainThread, parentPort } from "node:worker_threads";
import { once } from "node:events";


if (isMainThread) {
  const orders = [
    { id: 1001, euro: 25, express: false },
    { id: 1002, euro: 40, express: true },
    { id: 1003, euro: 15, express: false },
    { id: 1004, euro: 60, express: false },
    { id: 1005, euro: 30, express: true },
    { id: 1006, euro: 45, express: false },
    { id: 1007, euro: 20, express: true },
    { id: 1008, euro: 35, express: false },
  ];
  const packers = [0, 1].map(() => new Worker(new URL(import.meta.url)));
  const packed = [];
  const reports = packers.map((packer) => once(packer, "message"));
  orders.forEach((order, i) => packers[i % 2].postMessage(order)); // never waits
  packers.forEach((packer) => packer.postMessage(null));
  for (const [done] of await Promise.all(reports)) packed.push(...done);
  report(orders, packed);
} else {
  const done = [];
  parentPort.on("message", (order) => {
    if (order === null) {
      parentPort.postMessage(done);
      parentPort.close();
    } else {
      done.push(order);
    }
  });
}
Output
Packed: 8 of 8
Each order once: yes
Express: 3 orders, 90 €
Standard: 5 orders, 180 €

This is producer and consumer too, with two differences. First, there is no limit: postMessage never waits, the messages pile up in the Worker’s memory, and you have to build backpressure yourself (the Worker answers "ready" and only then gets the next one). Second, the orders are not taken from a shared queue; the main thread decides which order goes to which Worker (here in turn). The order is copied when it is sent, so there is no race condition on it.

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 tryPut. Count the refused orders and print "accepted + refused = 12".
  • By kind. Express orders must be packed first. Does two queues 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 ·