Ir al contenido
wannacode

Productor y consumidor

En resumen: Productor y consumidor es un patrón en el que unas partes de un programa ponen tareas en una cola limitada y otras las sacan; en JavaScript son funciones async en un solo hilo, y una cola llena hace esperar al productor.

Problema

Black Friday

A las 10:00 empieza la venta y el sitio recibe 50 pedidos por segundo. El almacén empaca 10. Hay tres malas maneras de manejarlo.

El sitio empaca por sí mismo. El cliente hace clic en «Pedir» y espera hasta que alguien empaca la caja. El sitio es tan rápido como el almacén.

El sitio pone los pedidos en una lista simple y el almacén los toma de ahí. Dos empacadores pueden tomar el mismo pedido y otro puede perderse: eso es una condición de carrera. Si añades un cerrojo, un empacador con la lista vacía tiene que comprobar una y otra vez si ha llegado algo.

La lista no tiene límite. El sitio está contento: nunca espera. Pero cada segundo la lista crece 40 pedidos. Después de una hora tiene 144 000, el último ha esperado una hora y se acaba la memoria.

La idea

Una cola con espacio limitado

Entre los dos lados va una cola. El productor pone tareas al final, los consumidores las toman del frente. Este es el patrón productor y consumidor. Los dos lados no se conocen: el sitio no sabe cuántos empacadores hay, y los empacadores no saben de dónde vienen los pedidos.

La cola es bloqueante y tiene espacio limitado:

  • si está vacía, el consumidor espera, sin hacer girar el procesador, hasta que llegue algo;
  • si está llena, el productor espera hasta que un consumidor libere un lugar;
  • el cerrojo está dentro de la cola: cada pedido va a exactamente un consumidor.
JavaScript no tiene una cola así ya hecha. En un solo hilo con funciones async son menos de 40 líneas con promesas: put y take devuelven una promesa que se cumple cuando hay un lugar o un pedido. No hace falta un cerrojo, porque el código entre dos await no puede ser interrumpido.
BoundedQueue
class BoundedQueue {
  #items = [];
  #putters = []; // los que esperan para poner
  #takers = []; // los que esperan para tomar


  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 y #takers son colas de funciones resolve: quién espera un lugar y quién espera un pedido. take despierta a un productor que espera, put despierta a un consumidor que espera. El que despierta comprueba otra vez (while), porque otro puede haber sido más rápido.

Míralo

El sitio y los empacadores

const DONE = null;


async function front(orders, queue, packers) {
  for (const order of orders) {
    await queue.put(order); // espera si está llena
  }
  for (let i = 0; i < packers; i++) {
    await queue.put(DONE); // una píldora por empacador
  }
}


async function pack(queue, packed) {
  while (true) {
    const order = await queue.take(); // espera si está vacía
    if (order === DONE) return; // la píldora: para
    packed.push(order);
  }
}
Paso a paso, en palabras
  1. El sitio (productor) pone pedidos en una cola con 3 lugares; dos empacadores (consumidores) los toman del frente. Está vacía, así que esperan en await take.
  2. El sitio pone 1001 y 1002 al final de la cola.
  3. Cada empacador toma el pedido del frente y lo empaca. Empacar es más lento que recibir pedidos.
  4. El sitio pone 1003, 1004 y 1005. Los 3 lugares están ocupados.
  5. 1006 no cabe: await put espera y el sitio deja de recibir pedidos. Esto es backpressure: el lado lento frena al rápido.
  6. El empacador 1 termina con 1001 y toma 1003. Se libera un lugar y el put que esperaba deja entrar 1006.
  7. Y así sigue: los empacadores toman del frente, el sitio añade al final y en la cola nunca hay más de 3.
  8. Se acabaron los pedidos. El sitio pone una píldora por empacador después de los pedidos, así cada pedido se empaca antes que ellas.
  9. Cada empacador termina su pedido, toma una píldora y se detiene. Sin la píldora esperaría en await take para siempre.
  10. 8 de 8, cada uno una vez: 3 exprés por 90 €, 5 normales por 180 €. Quién empaca cuál cambia de una ejecución a otra; los totales no.

El pedido:

// un pedido es { id, euro, express }

El sitio y los empacadores:

const DONE = null;


async function front(orders, queue, packers) {
  for (const order of orders) {
    await queue.put(order); // espera si está llena
  }
  for (let i = 0; i < packers; i++) {
    await queue.put(DONE); // una píldora por empacador
  }
}


async function pack(queue, packed) {
  while (true) {
    const order = await queue.take(); // espera si está vacía
    if (order === DONE) return; // la píldora: para
    packed.push(order);
  }
}
front y los dos pack son tres funciones async en un solo hilo. await es el lugar donde cada una puede esperar y dejar correr a las demás. Es turnarse, no trabajar en paralelo: para un empaque que calcula mucho hacen falta Workers (mira abajo).
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);

El informe comprueba cosas que no dependen de qué empacador tomó qué pedido:

function report(orders, packed) {
  const distinct = new Set(packed.map((order) => order.id)).size;
  console.log(`Empacados: ${packed.length} de ${orders.length}`);
  console.log(`Cada pedido una vez: ${distinct === orders.length ? "sí" : "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 ? "Exprés" : "Normales";
    console.log(`${kind}: ${group.length} pedidos, ${euro} €`);
  }
}
Salida
Empacados: 8 de 8
Cada pedido una vez: sí
Exprés: 3 pedidos, 90 €
Normales: 5 pedidos, 180 €

La salida es la misma en cada ejecución. Cuántos pedidos tomó cada empacador cambia, por eso el programa no lo imprime.

La idea

Cómo paran los empacadores

El empacador espera en await queue.take() hasta que llega un pedido. Si el sitio simplemente termina, los empacadores esperan para siempre y el programa nunca acaba.

Por eso, al final el sitio pone una píldora venenosa (poison pill): un valor especial que no es un pedido. Un empacador que toma la píldora se detiene. Hay una píldora por empacador: cada uno toma solo una y no la devuelve.

Las píldoras entran en la cola después de los pedidos, así cada pedido se empaca antes que ellas.

Aquí la píldora es DONE = null.

La idea

Backpressure

Una cola llena hace esperar al sitio. No es un error, es el objetivo: el lado lento empuja hacia atrás al rápido. Esto se llama backpressure. El sitio no puede adelantarse al almacén más que el tamaño de la cola, la memoria no crece y el último pedido espera como máximo lo que tarda en empacarse los 3 que tiene delante.

Esperar no es la única respuesta a una cola llena. También hay un método que no espera, sino que dice si funcionó:

tryPut(order) en nuestra cola devuelve false si no hay lugar.

Con él, el sitio puede mostrar «Mucha demanda, vuelve a intentarlo en un minuto» en vez de colgarse. Lo que importa es que la decisión es tuya: el pedido no desaparece en silencio.

Pruébalo

Predice la salida

Una cola con 2 lugares y sin empacadores. ¿Qué imprime?

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);
Respuesta

Pruébalo

Completa

¿Cuántas píldoras pone el sitio?
for (let i = 0; i < ; i++) {
  await queue.put(DONE);
}

Pruébalo

Encuentra el error

Un colega no quiere que el sitio espere y cambia una línea. Ahora el programa nunca se cuelga, pero el informe a veces dice «Empacados: 5 de 8». ¿En qué línea está el error?

Elige la línea con el error.

Míralo

Hilos, Workers y mensajes

Cuando empacar calcula mucho, hacen falta hilos de verdad: Workers. Entre ellos no hay una cola de objetos compartidos. El hilo principal envía cada pedido a un Worker con postMessage, y el Worker lo recibe en 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)); // nunca espera
  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);
    }
  });
}
Salida
Empacados: 8 de 8
Cada pedido una vez: sí
Exprés: 3 pedidos, 90 €
Normales: 5 pedidos, 180 €

Esto también es productor y consumidor, con dos diferencias. Primero, no hay límite: postMessage nunca espera, los mensajes se acumulan en la memoria del Worker y el backpressure tienes que construirlo tú (el Worker responde «listo» y solo entonces recibe el siguiente). Segundo, los pedidos no se toman de una cola compartida; el hilo principal decide qué pedido va a qué Worker (aquí por turnos). El pedido se copia al enviarlo, así que no hay condición de carrera sobre él.

Ejercicio

Dos entradas, tres empacadores

Resuelve la tarea en tu editor.

Los pedidos llegan de dos entradas: el sitio (2001–2006) y la aplicación móvil (3001–3006). Cada entrada es un productor aparte, y hay tres empacadores. La cola tiene 4 lugares. Las píldoras (una por empacador) llegan solo cuando las dos entradas han terminado. Imprime:

Salida
Empacados: 12 de 12
Cada pedido una vez: sí
Total: 325 €

Más tareas

  • Rechazo. Haz la cola de 1 lugar y que el sitio use tryPut. Cuenta los pedidos rechazados e imprime «aceptados + rechazados = 12».
  • Por tipo. Los pedidos exprés deben empacarse primero. ¿Ayuda dos colas? ¿Qué pasa con las píldoras?
  • Una cadena. Añade una segunda cola después de los empacadores: ellos ponen en ella las cajas empacadas, y un mensajero las toma. ¿Cuántas píldoras hacen falta ahora y quién las pone?

Test

Ponte a prueba

1. La cola entre el sitio y los empacadores está llena. ¿Qué hace put?
2. ¿Por qué tiene la cola un límite? Sin él, el sitio nunca espera.
3. Dos sitios ponen pedidos, tres empacadores los toman. ¿Cuántas píldoras hacen falta y cuándo se ponen?
4. ¿Cómo sabes que el programa con la cola es correcto, si el orden de empaque es distinto en cada ejecución?

Creado por Ivo Gergov como parte de la campaña de Evo Tech por una educación accesible en programación ·