Skip to content
wannacode

A pool of threads and tasks

In short: A pool is a few Worker threads in front of a task queue instead of a new Worker per task; in Node.js you write it as a small class.

Problem

A Worker for every image

The shop starts a sale. Sellers upload images of their products, and every image needs a thumbnail for the product list. Making a thumbnail is work that computes: it goes through every pixel (see work that waits and work that computes).

The first version starts a Worker for every uploaded image:

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

With 20 images everything works on your machine: "Done: 20 thumbnails". In the first hour of the sale 10,000 come at once.

  • Every Worker is a whole new V8: its own memory, its own event loop, tens of MB. 10,000 Workers eat the server's memory long before they are done.
  • Even if there is memory, 10,000 Workers on 8 cores only take turns. The processor wastes time switching them, and every image gets slower.

The server does not slow down; it falls over. And a server that is down does not serve the buyers either.

The idea

A few workers and a queue

A thread pool is a few workers, created once, and a queue of tasks in front of them. A task is "shrink this image". A free worker takes the next task from the queue, finishes it and comes back for a new one.

  • The number of workers is constant. 10,000 images do not make 10,000 threads, but 10,000 tasks in the queue. A task is a small object; a thread is expensive.
  • The workers are reused. A thread is not created and destroyed for every image.
  • The queue absorbs the peak. When uploads come faster than the workers can handle, the tasks wait instead of bringing the server down.

This is producer and consumer: the code that accepts uploads puts tasks into the queue, and the workers are the consumers.

You already met the pool in the lesson on threads: runLimited, which runs at most a few promises at once. Here we will see what happens inside, how to order the results, what happens to errors and how many workers you need.

Node runs your JavaScript in one thread with its own event loop. That is enough for work that waits: the network and files wait outside of it. For work that computes, like making thumbnails, you need real threads: Workers from node:worker_threads. Node does not come with a pool of Workers, so here is a small one. Tasks wait in the queue, and a free Worker takes the next one:

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


  // The task goes into the queue; the promise is fulfilled when a Worker finishes it.
  run(task) {
    return new Promise((resolve, reject) => {
      this.#queue.push({ task, resolve, reject });
      this.#next();
    });
  }


  // A free Worker takes the next task from the queue.
  #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()));
  }
}

The Worker is the same file: there isMainThread is false, and it waits for messages and sends back the result or the text of the error:

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

See it

Three workers

Making a thumbnail here is a pause, 2 ms for every KB, and the size comes from the name of the image. That way the output is the same on every run:

// Making a thumbnail: here it is a pause, and the size comes from the name.
function thumbnail(image) {
  if (image.startsWith("broken")) {
    throw new Error("corrupt file");
  }
  const kb = 20 + 3 * image.length;
  // Atomics.wait sleeps synchronously: in a Worker this is allowed.
  Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 2 * kb);
  return kb;
}
Atomics.wait on an empty SharedArrayBuffer sleeps synchronously. In the main thread this is forbidden (it must stay free), but in a Worker it is allowed.

Eight images, a pool of three workers:

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();
Step by step, in words
  1. 8 images wait. The pool is 3 Workers: new WorkerPool(3, import.meta.url).
  2. pool.run puts every image into the queue. The three Workers take the first three.
  3. Worker 3 finishes first: hat is the third image, and its result goes to the third place. The worker does not stop: it takes the next task.
  4. The queue is empty. Worker 1 is done and has nothing to take, so it waits. The others finish their last images.
  5. Promise.all gives the sizes in the order of the images, not in the order they finished. Total 379 KB; pool.close() stops the Workers.
const images = ["boots.jpg", "jacket.jpg", "hat.jpg", "scarf.jpg", "backpack.jpg", "gloves.jpg", "watch.jpg", "cap.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();
Output
boots.jpg: 47 KB
jacket.jpg: 50 KB
hat.jpg: 41 KB
scarf.jpg: 47 KB
backpack.jpg: 56 KB
gloves.jpg: 50 KB
watch.jpg: 47 KB
cap.jpg: 41 KB
Total: 379 KB
  • new WorkerPool(3, import.meta.url) starts three Workers with this file.
  • pool.run(image) puts the image into the queue and gives back a promise right away. map does not wait for anything.
  • Promise.all waits for all of them and gives an array in the order of the promises, whatever finished first.
  • pool.close() stops the Workers. Without it the program does not end: live Workers keep the process running.
More precisely, if you are curious

Node does have a small pool of its own: libuv, the library under Node, keeps 4 threads for work that cannot wait without a thread (files, crypto.scrypt, zlib). Your JavaScript does not run in them.

In a project you do not write a pool of Workers yourself: the Piscina package does the same, with a bounded queue, timeouts and cancelling tasks.

How it is in the other languages: Java, C# and Python come with a thread pool (ExecutorService, ThreadPool, ThreadPoolExecutor).

Try it

Find the bug

A colleague rewrote the thumbnails with a pool, and the sale did not get faster. Six images take as long as without a pool:

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(`Took over 0.5 s: ${Date.now() - start > 500}`);
await pool.close();
Output
Took over 0.5 s: true

Pick the line with the bug.

The idea

Results in input order

Tasks finish in an order you do not know: it depends on the images, on the load, on which worker was free. So never print "in the order they finished" if you want the same output every time.

Two ways give the same result on every run:

  • In input order. You keep the promises in a list in the order of the tasks and wait for them in that order. That is how the code above works.
  • A sum or a count. If you only need the total, the order does not matter: 47 + 50 is the same as 50 + 47.

Sometimes you want the results in the order they finished: for a "3 of 8 done" bar, or to show the first ready images. a continuation (await) for every task separately gives them that way. Then you only collect things that do not depend on the order:

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(`Done: ${count} of ${images.length}, total: ${total} KB`);
await pool.close();
Output
Done: 8 of 8, total: 379 KB
Every image has its own async function, which waits for its result and adds it. Here total += kb is safe: JavaScript in the main thread does not interrupt a line halfway. But total += await pool.run(image) would be wrong: it reads total before the await and writes after it. That is exactly a race between awaits.

The idea

An error inside a task

One of the images is corrupt, and thumbnail throws an exception. It does not come out in the thread that started the task: that thread has long moved on. The pool catches it and keeps it in the promise of the result. It comes out when you wait for that result:

const images = ["boots.jpg", "jacket.jpg", "broken.jpg", "hat.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();
Output
boots.jpg: 47 KB
jacket.jpg: 50 KB
broken.jpg: error: corrupt file
hat.jpg: 41 KB
Promise.allSettled waits for all of them and gives every promise a status: "fulfilled" with a value or "rejected" with a reason. The Worker does not throw across the thread boundary: it sends the text of the error, and WorkerPool turns it into an Error.

Try it

Where did the error go

The second image is corrupt. What does the program print?

const images = ["boots.jpg", "broken.jpg", "hat.jpg"];
const pool = new WorkerPool(3, import.meta.url);
try {
  const sizes = await Promise.all(images.map((image) => pool.run(image)));
  console.log(`Sizes: ${sizes.join(", ")}`);
} catch (error) {
  console.log(`Caught: ${error.message}`);
}
await pool.close();
Answer

The idea

How many workers

It depends on what they do:

  • Work that computes (thumbnails, PDFs, compression): as many as the cores. A ninth worker on 8 cores has nowhere to run; it only takes turns with the others.
  • Work that waits (e-mail, payment, database): more. While one worker waits for the network, the core is free for another. If a task waits W ms and computes C ms, about 1 + W / C workers fit on a core.
function poolSize(cores, waitMs, computeMs) {
  return cores * (1 + waitMs / computeMs);
}


console.log(`Thumbnails (computing only): ${poolSize(8, 0, 50)}`);
console.log(`E-mails (90 ms waiting, 10 ms computing): ${poolSize(8, 90, 10)}`);
Output
Thumbnails (computing only): 8
E-mails (90 ms waiting, 10 ms computing): 80

The number of cores comes from os.availableParallelism(). The formula is a start, not a law: measure. If the e-mails go through one server with a limit of 20 connections, 80 workers will only wait in a queue in front of it.

In JavaScript work that waits needs no pool: the event loop waits for thousands of requests in one thread (the lesson async). A pool of Workers is only for work that computes, and then it is as big as the number of cores.

Try it

Fill in

How many workers for a task that waits waitMs and computes computeMs?

function poolSize(cores, waitMs, computeMs) {
  return cores * (1 + );
}
More precisely, if you are curious

The formula holds for Workers too: work that waits inside a Worker also leaves the core free. But in JavaScript work that waits is rarely in a Worker: it is in the main thread with await.

Exercise

The invoices of the sale

Solve the task in your editor.

After the sale every order needs an invoice. invoice(amount) makes the PDF (here a pause of 50 ms) and returns the amount with 20% VAT. An order with an amount of 0 or less is invalid, and the function throws an exception with the text "invalid amount".

The orders are 1001 to 1008 with amounts €50, €125, €80, €0, €240, €35, €90 and €15. Make the invoices in a pool of 4 workers. Print them in the order of the orders, and the invalid one with the text of the error. At the end: the sum of the good invoices and how many failed.

Output
Order 1001: €60
Order 1002: €150
Order 1003: €96
Order 1004: error: invalid amount
Order 1005: €288
Order 1006: €42
Order 1007: €108
Order 1008: €18
Total: €762, failed: 1
Use WorkerPool from the lesson. The Worker calls invoice instead of thumbnail.

More tasks

  • A bounded queue. Make the pool keep at most 10 tasks waiting in the queue. What happens to the eleventh? (run waits if the queue is full)
  • Progress. Print "3 of 8 done" after every finished invoice, and at the end the ordered list again.
  • A timeout. An invoice that takes over 1 second counts as failed, without stopping the others.

Test

Check yourself

1. Why a pool of a few threads, and not a thread for every uploaded image?
2. A pool of 3 workers shrinks 8 images. The third one finishes first. Where is its size when you read them in list order with get, result, Promise.all or sizes[i]?
3. A task in the pool throws an exception. When do you see it?
4. The server has 8 cores. Every e-mail waits 90 ms for the network and computes for 10 ms. About how many workers keep the cores busy?
5. How many workers for making thumbnails, if it only computes and the server has 8 cores?

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