Ir al contenido
wannacode

Grupo de hilos y tareas

En resumen: Un grupo son unos pocos hilos Worker delante de una cola de tareas en lugar de un Worker nuevo por tarea; en Node.js lo escribes como una clase pequeña.

Problema

Un Worker por cada imagen

La tienda empieza unas rebajas. Los vendedores suben imágenes de sus productos, y cada imagen necesita una miniatura para la lista de productos. Hacer una miniatura es trabajo que calcula: recorre cada píxel (mira trabajo que espera y trabajo que calcula).

La primera versión inicia un Worker por cada imagen subida:

const workers = [];
for (const image of uploads) {
  const worker = new Worker(new URL(import.meta.url), { workerData: image });
  workers.push(worker);
}

Con 20 imágenes todo funciona en tu computadora: «Listas: 20 miniaturas». En la primera hora de las rebajas llegan 10 000 a la vez.

  • Cada Worker es un V8 completamente nuevo: memoria propia, bucle de eventos propio, decenas de MB. 10 000 Workers se comen la memoria del servidor mucho antes de terminar.
  • Aunque haya memoria, 10 000 Workers en 8 núcleos solo se turnan. El procesador pierde tiempo cambiando de uno a otro, y cada imagen se vuelve más lenta.

El servidor no se vuelve más lento; se cae. Y un servidor caído tampoco atiende a los compradores.

La idea

Unos pocos trabajadores y una cola

Un grupo de hilos son unos pocos trabajadores, creados una vez, y delante de ellos una cola de tareas. Una tarea es «reduce esta imagen». Un trabajador libre toma la siguiente tarea de la cola, la termina y vuelve por una nueva.

  • El número de trabajadores es constante. 10 000 imágenes no hacen 10 000 hilos, sino 10 000 tareas en la cola. Una tarea es un objeto pequeño; un hilo es caro.
  • Los trabajadores se reutilizan. No se crea ni se destruye un hilo por cada imagen.
  • La cola absorbe el pico. Cuando las subidas llegan más rápido de lo que los trabajadores pueden atender, las tareas esperan en lugar de tumbar el servidor.

Esto es productor y consumidor: el código que acepta las subidas pone tareas en la cola, y los trabajadores son los consumidores.

Ya conociste el grupo en la lección sobre hilos: runLimited, que ejecuta como máximo unas pocas promesas a la vez. Aquí veremos qué pasa por dentro, cómo ordenar los resultados, qué pasa con los errores y cuántos trabajadores necesitas.

Node ejecuta tu JavaScript en un hilo con su propio bucle de eventos. Eso basta para el trabajo que espera: la red y los archivos esperan fuera de él. Para el trabajo que calcula, como hacer miniaturas, necesitas hilos de verdad: los Worker de node:worker_threads. Node no trae un grupo de Workers, así que aquí va uno pequeño. Las tareas esperan en la cola, y un Worker libre toma la siguiente:

class WorkerPool {
  #workers = [];
  #idle = [];
  #queue = [];


  constructor(size, file) {
    for (let i = 0; i < size; i++) {
      const worker = new Worker(new URL(file));
      this.#workers.push(worker);
      this.#idle.push(worker);
    }
  }


  // La tarea entra en la cola; la promesa se cumple cuando un Worker la termina.
  run(task) {
    return new Promise((resolve, reject) => {
      this.#queue.push({ task, resolve, reject });
      this.#next();
    });
  }


  // Un Worker libre toma la siguiente tarea de la cola.
  #next() {
    if (this.#idle.length === 0 || this.#queue.length === 0) return;
    const worker = this.#idle.pop();
    const { task, resolve, reject } = this.#queue.shift();
    worker.once("message", ({ value, error }) => {
      this.#idle.push(worker);
      if (error === undefined) resolve(value);
      else reject(new Error(error));
      this.#next();
    });
    worker.postMessage(task);
  }


  close() {
    return Promise.all(this.#workers.map((worker) => worker.terminate()));
  }
}

El Worker es el mismo archivo: allí isMainThread es false, y espera mensajes y devuelve el resultado o el texto del error:

parentPort.on("message", (image) => {
  try {
    parentPort.postMessage({ value: thumbnail(image) });
  } catch (error) {
    parentPort.postMessage({ error: error.message });
  }
});

Míralo

Tres trabajadores

Hacer una miniatura aquí es una pausa, 2 ms por cada KB, y el tamaño sale del nombre de la imagen. Así la salida es la misma en cada ejecución:

// Crear una miniatura: aquí es una pausa, y el tamaño sale del nombre.
function thumbnail(image) {
  if (image.startsWith("dañada")) {
    throw new Error("archivo dañado");
  }
  const kb = 20 + 3 * image.length;
  // Atomics.wait duerme de forma síncrona: en un Worker está permitido.
  Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 2 * kb);
  return kb;
}
Atomics.wait sobre un SharedArrayBuffer vacío duerme de forma síncrona. En el hilo principal está prohibido (debe quedar libre), pero en un Worker está permitido.

Ocho imágenes, un grupo de tres trabajadores:

const pool = new WorkerPool(3, import.meta.url);
const sizes = await Promise.all(
  images.map((image) => pool.run(image)),
);
let total = 0;
images.forEach((image, i) => {
  console.log(`${image}: ${sizes[i]} KB`);
  total += sizes[i];
});
console.log(`Total: ${total} KB`);
await pool.close();
Paso a paso, en palabras
  1. 8 imágenes esperan. El grupo son 3 Workers: new WorkerPool(3, import.meta.url).
  2. pool.run pone cada imagen en la cola. Los tres Workers toman las tres primeras.
  3. Worker 3 termina primero: Top es la tercera imagen, y su resultado va al tercer lugar. El trabajador no se detiene: toma la siguiente tarea.
  4. La cola está vacía. Worker 1 terminó y no tiene nada que tomar, así que espera. Los demás terminan sus últimas imágenes.
  5. Promise.all da los tamaños en el orden de las imágenes, no en el orden en que terminaron. Total 379 KB; pool.close() detiene los Workers.
const images = ["Bolso.jpg", "Zapato.jpg", "Top.jpg", "Falda.jpg", "Pantalón.jpg", "Camisa.jpg", "Botín.jpg", "Boa.jpg"];
const pool = new WorkerPool(3, import.meta.url);
const sizes = await Promise.all(
  images.map((image) => pool.run(image)),
);
let total = 0;
images.forEach((image, i) => {
  console.log(`${image}: ${sizes[i]} KB`);
  total += sizes[i];
});
console.log(`Total: ${total} KB`);
await pool.close();
Salida
Bolso.jpg: 47 KB
Zapato.jpg: 50 KB
Top.jpg: 41 KB
Falda.jpg: 47 KB
Pantalón.jpg: 56 KB
Camisa.jpg: 50 KB
Botín.jpg: 47 KB
Boa.jpg: 41 KB
Total: 379 KB
  • new WorkerPool(3, import.meta.url) inicia tres Workers con este archivo.
  • pool.run(image) pone la imagen en la cola y devuelve una promesa de inmediato. map no espera nada.
  • Promise.all espera a todas y da un arreglo en el orden de las promesas, sea cual sea la que terminó primero.
  • pool.close() detiene los Workers. Sin eso el programa no termina: los Workers vivos mantienen el proceso en marcha.
Más precisamente, para curiosos

Node sí tiene un pequeño grupo propio: libuv, la biblioteca bajo Node, mantiene 4 hilos para el trabajo que no puede esperar sin un hilo (archivos, crypto.scrypt, zlib). Tu JavaScript no se ejecuta en ellos.

En un proyecto no escribes tú un grupo de Workers: el paquete Piscina hace lo mismo, con cola limitada, timeouts y cancelación de tareas.

Cómo es en los otros lenguajes: Java, C# y Python traen un grupo de hilos (ExecutorService, ThreadPool, ThreadPoolExecutor).

Pruébalo

Encuentra el error

Un compañero reescribió las miniaturas con un grupo, y las rebajas no fueron más rápidas. Seis imágenes tardan lo mismo que sin grupo:

const start = Date.now();
const pool = new WorkerPool(3, import.meta.url);
const sizes = [];
for (const image of images) {
  const size = pool.run(image);
  sizes.push(await size);
}
console.log(`Tardó más de 0,5 s: ${Date.now() - start > 500}`);
await pool.close();
Salida
Tardó más de 0,5 s: true

Elige la línea con el error.

La idea

Resultados en el orden de entrada

Las tareas terminan en un orden que no conoces: depende de las imágenes, de la carga, de qué trabajador estaba libre. Así que nunca imprimas «en el orden en que terminaron» si quieres la misma salida cada vez.

Dos formas dan el mismo resultado en cada ejecución:

  • En el orden de entrada. Guardas las promesas en una lista en el orden de las tareas y las esperas en ese orden. Así funciona el código de arriba.
  • Una suma o una cuenta. Si solo necesitas el total, el orden no importa: 47 + 50 es lo mismo que 50 + 47.

A veces quieres los resultados en el orden en que terminaron: para una barra de «3 de 8 listas» o para mostrar las primeras imágenes listas. una continuación (await) para cada tarea por separado los da así. Entonces recoges solo cosas que no dependen del orden:

const pool = new WorkerPool(3, import.meta.url);
let count = 0;
let total = 0;
await Promise.all(
  images.map(async (image) => {
    const kb = await pool.run(image);
    total += kb;
    count++;
  }),
);
console.log(`Listas: ${count} de ${images.length}, en total: ${total} KB`);
await pool.close();
Salida
Listas: 8 de 8, en total: 379 KB
Cada imagen tiene su propia función async, que espera su resultado y lo suma. Aquí total += kb es seguro: JavaScript en el hilo principal no interrumpe una línea a la mitad. Pero total += await pool.run(image) sería incorrecto: lee total antes del await y escribe después. Eso es justo una carrera entre awaits.

La idea

Un error dentro de una tarea

Una de las imágenes está dañada, y thumbnail lanza una excepción. No sale en el hilo que inició la tarea: ese hilo siguió su camino hace rato. El grupo la captura y la guarda en la promesa del resultado. Sale cuando esperas ese resultado:

const images = ["Bolso.jpg", "Zapato.jpg", "dañada.jpg", "Top.jpg"];
const pool = new WorkerPool(3, import.meta.url);
const results = await Promise.allSettled(
  images.map((image) => pool.run(image)),
);
images.forEach((image, i) => {
  const result = results[i];
  if (result.status === "fulfilled") {
    console.log(`${image}: ${result.value} KB`);
  } else {
    console.log(`${image}: error: ${result.reason.message}`);
  }
});
await pool.close();
Salida
Bolso.jpg: 47 KB
Zapato.jpg: 50 KB
dañada.jpg: error: archivo dañado
Top.jpg: 41 KB
Promise.allSettled espera a todas y da a cada promesa un status: "fulfilled" con un value o "rejected" con un reason. El Worker no lanza a través del límite del hilo: envía el texto del error, y WorkerPool lo convierte en un Error.

Pruébalo

Adónde fue el error

La segunda imagen está dañada. ¿Qué imprime el programa?

const images = ["Bolso.jpg", "dañada.jpg", "Top.jpg"];
const pool = new WorkerPool(3, import.meta.url);
try {
  const sizes = await Promise.all(images.map((image) => pool.run(image)));
  console.log(`Tamaños: ${sizes.join(", ")}`);
} catch (error) {
  console.log(`Capturado: ${error.message}`);
}
await pool.close();
Respuesta

La idea

Cuántos trabajadores

Depende de lo que hagan:

  • Trabajo que calcula (miniaturas, PDF, compresión): tantos como núcleos. Un noveno trabajador en 8 núcleos no tiene dónde ejecutarse; solo se turna con los demás.
  • Trabajo que espera (correo, pago, base de datos): más. Mientras un trabajador espera a la red, el núcleo queda libre para otro. Si una tarea espera W ms y calcula C ms, caben unos 1 + W / C trabajadores en un núcleo.
function poolSize(cores, waitMs, computeMs) {
  return cores * (1 + waitMs / computeMs);
}


console.log(`Miniaturas (solo cálculo): ${poolSize(8, 0, 50)}`);
console.log(`Correos (90 ms de espera, 10 ms de cálculo): ${poolSize(8, 90, 10)}`);
Salida
Miniaturas (solo cálculo): 8
Correos (90 ms de espera, 10 ms de cálculo): 80

El número de núcleos viene de os.availableParallelism(). La fórmula es un punto de partida, no una ley: mide. Si los correos pasan por un servidor con un límite de 20 conexiones, 80 trabajadores solo esperarán en una cola delante de él.

En JavaScript el trabajo que espera no necesita un grupo: el bucle de eventos espera miles de peticiones en un hilo (la lección async no es un hilo). Un grupo de Workers es solo para el trabajo que calcula, y entonces es tan grande como el número de núcleos.

Pruébalo

Completa

¿Cuántos trabajadores para una tarea que espera waitMs y calcula computeMs?

function poolSize(cores, waitMs, computeMs) {
  return cores * (1 + );
}
Más precisamente, para curiosos

La fórmula vale también para los Workers: el trabajo que espera dentro de un Worker también deja libre el núcleo. Pero en JavaScript el trabajo que espera rara vez está en un Worker: está en el hilo principal con await.

Ejercicio

Las facturas de las rebajas

Resuelve la tarea en tu editor.

Después de las rebajas cada pedido necesita una factura. invoice(amount) crea el PDF (aquí una pausa de 50 ms) y devuelve el importe con 20 % de IVA. Un pedido con un importe de 0 o menos no es válido, y la función lanza una excepción con el texto «importe no válido».

Los pedidos son del 1001 al 1008 con los importes 50 €, 125 €, 80 €, 0 €, 240 €, 35 €, 90 € y 15 €. Haz las facturas en un grupo de 4 trabajadores. Imprímelas en el orden de los pedidos, y la no válida con el texto del error. Al final: la suma de las facturas buenas y cuántas fallaron.

Salida
Pedido 1001: 60 €
Pedido 1002: 150 €
Pedido 1003: 96 €
Pedido 1004: error: importe no válido
Pedido 1005: 288 €
Pedido 1006: 42 €
Pedido 1007: 108 €
Pedido 1008: 18 €
Total: 762 €, con error: 1
Usa WorkerPool de la lección. El Worker llama a invoice en lugar de thumbnail.

Más tareas

  • Una cola limitada. Haz que el grupo mantenga como máximo 10 tareas esperando en la cola. ¿Qué pasa con la undécima? (run espera si la cola está llena)
  • Progreso. Imprime «3 de 8 listas» después de cada factura terminada, y al final otra vez la lista ordenada.
  • Un timeout. Una factura que tarda más de 1 segundo cuenta como fallida, sin detener a las demás.

Test

Ponte a prueba

1. ¿Por qué un grupo de pocos hilos y no un hilo por cada imagen subida?
2. Un grupo de 3 trabajadores reduce 8 imágenes. La tercera termina primera. ¿Dónde está su tamaño cuando las lees en el orden de la lista con get, result, Promise.all o sizes[i]?
3. Una tarea del grupo lanza una excepción. ¿Cuándo la ves?
4. El servidor tiene 8 núcleos. Cada correo espera 90 ms a la red y calcula 10 ms. ¿Cuántos trabajadores mantienen ocupados los núcleos, más o menos?
5. ¿Cuántos trabajadores para hacer miniaturas, si solo calcula y el servidor tiene 8 núcleos?

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