Zum Inhalt
wannacode

Thread-Pool und Aufgaben

Kurz gesagt: Ein Pool sind ein paar Worker-Threads vor einer Aufgabenwarteschlange statt eines neuen Workers pro Aufgabe; in Node.js schreibst du ihn als kleine Klasse.

Problem

Ein Worker für jedes Bild

Der Shop startet einen Sale. Verkäufer laden Bilder ihrer Produkte hoch, und jedes Bild braucht ein Vorschaubild für die Produktliste. Ein Vorschaubild zu erzeugen ist Arbeit, die rechnet: Sie geht durch jeden Pixel (siehe Arbeit, die wartet, und Arbeit, die rechnet).

Die erste Version startet für jedes hochgeladene Bild einen Worker:

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

Mit 20 Bildern funktioniert alles auf deinem Rechner: „Fertig: 20 Vorschaubilder“. In der ersten Stunde des Sales kommen 10.000 auf einmal.

  • Jeder Worker ist ein ganz neues V8: eigener Speicher, eigener Event Loop, Dutzende MB. 10.000 Worker fressen den Speicher des Servers, lange bevor sie fertig sind.
  • Selbst wenn Speicher da ist, wechseln sich 10.000 Worker auf 8 Kernen nur ab. Der Prozessor verschwendet Zeit mit dem Umschalten, und jedes Bild wird langsamer.

Der Server wird nicht langsamer, er fällt um. Und ein Server, der down ist, bedient auch die Kunden nicht.

Die Idee

Ein paar Worker und eine Warteschlange

Ein Thread-Pool sind ein paar Worker, einmal erzeugt, und davor eine Warteschlange von Aufgaben. Eine Aufgabe ist „verkleinere dieses Bild“. Ein freier Worker nimmt die nächste Aufgabe aus der Warteschlange, erledigt sie und kommt für eine neue zurück.

  • Die Zahl der Worker ist konstant. 10.000 Bilder ergeben nicht 10.000 Threads, sondern 10.000 Aufgaben in der Warteschlange. Eine Aufgabe ist ein kleines Objekt; ein Thread ist teuer.
  • Die Worker werden wiederverwendet. Für jedes Bild wird kein Thread erzeugt und zerstört.
  • Die Warteschlange fängt die Spitze ab. Kommen Uploads schneller, als die Worker sie bearbeiten, warten die Aufgaben, statt den Server zu Fall zu bringen.

Das ist Producer und Consumer: Der Code, der Uploads annimmt, legt Aufgaben in die Warteschlange, und die Worker sind die Consumer.

Den Pool hast du schon in der Lektion über Threads kennengelernt: runLimited, das höchstens ein paar Promises gleichzeitig ausführt. Hier sehen wir, was darin passiert, wie man die Ergebnisse ordnet, was mit Fehlern geschieht und wie viele Worker du brauchst.

Node führt dein JavaScript in einem Thread mit eigenem Event Loop aus. Das reicht für Arbeit, die wartet: Netzwerk und Dateien warten außerhalb davon. Für Arbeit, die rechnet, wie Vorschaubilder, brauchst du echte Threads: Worker aus node:worker_threads. Node bringt keinen Pool von Workern mit, hier ist also ein kleiner. Die Aufgaben warten in der Warteschlange, und ein freier Worker nimmt die nächste:

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


  // Die Aufgabe kommt in die Warteschlange; das Promise wird erfüllt, wenn ein Worker sie beendet.
  run(task) {
    return new Promise((resolve, reject) => {
      this.#queue.push({ task, resolve, reject });
      this.#next();
    });
  }


  // Ein freier Worker nimmt die nächste Aufgabe aus der Warteschlange.
  #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()));
  }
}

Der Worker ist dieselbe Datei: Dort ist isMainThread false, und er wartet auf Nachrichten und schickt das Ergebnis oder den Text des Fehlers zurück:

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

Sieh es dir an

Drei Worker

Ein Vorschaubild zu erzeugen ist hier eine Pause, 2 ms für jedes KB, und die Größe kommt aus dem Namen des Bildes. So ist die Ausgabe bei jedem Lauf gleich:

// Vorschaubild erzeugen: Hier ist es eine Pause, und die Größe kommt aus dem Namen.
function thumbnail(image) {
  if (image.startsWith("defekt")) {
    throw new Error("defekte Datei");
  }
  const kb = 20 + 3 * image.length;
  // Atomics.wait schläft synchron: In einem Worker ist das erlaubt.
  Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 2 * kb);
  return kb;
}
Atomics.wait auf einem leeren SharedArrayBuffer schläft synchron. Im Hauptthread ist das verboten (er muss frei bleiben), aber in einem Worker ist es erlaubt.

Acht Bilder, ein Pool aus drei Workern:

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(`Gesamt: ${total} KB`);
await pool.close();
Schritt für Schritt, in Worten
  1. 8 Bilder warten. Der Pool besteht aus 3 Workern: new WorkerPool(3, import.meta.url).
  2. pool.run legt jedes Bild in die Warteschlange. Die drei Worker nehmen die ersten drei.
  3. Worker 3 wird als Erstes fertig: Hut ist das dritte Bild, und sein Ergebnis kommt an dritter Stelle. Der Worker hält nicht an: Er nimmt die nächste Aufgabe.
  4. Die Warteschlange ist leer. Worker 1 ist fertig und hat nichts mehr zu nehmen, also wartet er. Die anderen beenden ihre letzten Bilder.
  5. Promise.all liefert die Größen in der Reihenfolge der Bilder, nicht in der Reihenfolge des Fertigwerdens. Gesamt 379 KB; pool.close() stoppt die Worker.
const images = ["Kleid.jpg", "Mantel.jpg", "Hut.jpg", "Jacke.jpg", "Rucksack.jpg", "Gürtel.jpg", "Mütze.jpg", "Uhr.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(`Gesamt: ${total} KB`);
await pool.close();
Ausgabe
Kleid.jpg: 47 KB
Mantel.jpg: 50 KB
Hut.jpg: 41 KB
Jacke.jpg: 47 KB
Rucksack.jpg: 56 KB
Gürtel.jpg: 50 KB
Mütze.jpg: 47 KB
Uhr.jpg: 41 KB
Gesamt: 379 KB
  • new WorkerPool(3, import.meta.url) startet drei Worker mit dieser Datei.
  • pool.run(image) legt das Bild in die Warteschlange und gibt sofort ein Promise zurück. map wartet auf nichts.
  • Promise.all wartet auf alle und liefert ein Array in der Reihenfolge der Promises, egal, was zuerst fertig wurde.
  • pool.close() stoppt die Worker. Ohne das endet das Programm nicht: Lebende Worker halten den Prozess am Laufen.
Genauer, für Neugierige

Node hat durchaus einen kleinen eigenen Pool: libuv, die Bibliothek unter Node, hält 4 Threads für Arbeit, die nicht ohne Thread warten kann (Dateien, crypto.scrypt, zlib). Dein JavaScript läuft nicht darin.

In einem Projekt schreibst du keinen Pool von Workern selbst: Das Paket Piscina macht dasselbe, mit begrenzter Warteschlange, Timeouts und dem Abbrechen von Aufgaben.

Wie es in den anderen Sprachen ist: Java, C# und Python bringen einen Thread-Pool mit (ExecutorService, ThreadPool, ThreadPoolExecutor).

Probier es aus

Finde den Fehler

Ein Kollege hat die Vorschaubilder mit einem Pool umgeschrieben, und der Sale wurde nicht schneller. Sechs Bilder brauchen so lange wie ohne 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(`Dauerte über 0,5 s: ${Date.now() - start > 500}`);
await pool.close();
Ausgabe
Dauerte über 0,5 s: true

Wähle die Zeile mit dem Fehler.

Die Idee

Ergebnisse in Eingabereihenfolge

Aufgaben werden in einer Reihenfolge fertig, die du nicht kennst: Sie hängt von den Bildern ab, von der Last, davon, welcher Worker frei war. Gib also nie „in der Reihenfolge des Fertigwerdens“ aus, wenn du jedes Mal dieselbe Ausgabe willst.

Zwei Wege liefern bei jedem Lauf dasselbe Ergebnis:

  • In Eingabereihenfolge. Du hältst die Promises in einer Liste in der Reihenfolge der Aufgaben und wartest in dieser Reihenfolge darauf. So funktioniert der Code oben.
  • Eine Summe oder eine Anzahl. Brauchst du nur das Ergebnis insgesamt, ist die Reihenfolge egal: 47 + 50 ist dasselbe wie 50 + 47.

Manchmal willst du die Ergebnisse in der Reihenfolge des Fertigwerdens: für einen Balken „3 von 8 fertig“ oder um die ersten fertigen Bilder zu zeigen. eine Fortsetzung (await) für jede Aufgabe einzeln liefert sie so. Dann sammelst du nur Dinge, die nicht von der Reihenfolge abhängen:

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(`Fertig: ${count} von ${images.length}, gesamt: ${total} KB`);
await pool.close();
Ausgabe
Fertig: 8 von 8, gesamt: 379 KB
Jedes Bild hat seine eigene async-Funktion, die auf ihr Ergebnis wartet und es addiert. Hier ist total += kb sicher: JavaScript im Hauptthread unterbricht eine Zeile nicht in der Mitte. Aber total += await pool.run(image) wäre falsch: Es liest total vor dem await und schreibt danach. Das ist genau ein Wettlauf zwischen awaits.

Die Idee

Ein Fehler in einer Aufgabe

Eines der Bilder ist defekt, und thumbnail wirft eine Exception. Sie kommt nicht in dem Thread heraus, der die Aufgabe gestartet hat: Der ist längst weitergezogen. Der Pool fängt sie ab und bewahrt sie im Promise des Ergebnisses auf. Sie kommt heraus, wenn du auf dieses Ergebnis wartest:

const images = ["Kleid.jpg", "Mantel.jpg", "defekt.jpg", "Hut.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}: Fehler: ${result.reason.message}`);
  }
});
await pool.close();
Ausgabe
Kleid.jpg: 47 KB
Mantel.jpg: 50 KB
defekt.jpg: Fehler: defekte Datei
Hut.jpg: 41 KB
Promise.allSettled wartet auf alle und gibt jedem Promise einen status: "fulfilled" mit einem value oder "rejected" mit einem reason. Der Worker wirft nicht über die Thread-Grenze: Er schickt den Text des Fehlers, und WorkerPool macht daraus einen Error.

Probier es aus

Wohin ist der Fehler gegangen

Das zweite Bild ist defekt. Was gibt das Programm aus?

const images = ["Kleid.jpg", "defekt.jpg", "Hut.jpg"];
const pool = new WorkerPool(3, import.meta.url);
try {
  const sizes = await Promise.all(images.map((image) => pool.run(image)));
  console.log(`Größen: ${sizes.join(", ")}`);
} catch (error) {
  console.log(`Abgefangen: ${error.message}`);
}
await pool.close();
Antwort

Die Idee

Wie viele Worker

Das hängt davon ab, was sie tun:

  • Arbeit, die rechnet (Vorschaubilder, PDFs, Komprimierung): so viele wie Kerne. Ein neunter Worker auf 8 Kernen hat keinen Platz zum Laufen; er wechselt sich nur mit den anderen ab.
  • Arbeit, die wartet (E-Mail, Zahlung, Datenbank): mehr. Während ein Worker auf das Netzwerk wartet, ist der Kern für einen anderen frei. Wartet eine Aufgabe W ms und rechnet C ms, passen etwa 1 + W / C Worker auf einen Kern.
function poolSize(cores, waitMs, computeMs) {
  return cores * (1 + waitMs / computeMs);
}


console.log(`Vorschaubilder (nur Rechnen): ${poolSize(8, 0, 50)}`);
console.log(`E-Mails (90 ms Warten, 10 ms Rechnen): ${poolSize(8, 90, 10)}`);
Ausgabe
Vorschaubilder (nur Rechnen): 8
E-Mails (90 ms Warten, 10 ms Rechnen): 80

Die Zahl der Kerne kommt aus os.availableParallelism(). Die Formel ist ein Anfang, kein Gesetz: Miss nach. Gehen die E-Mails über einen Server mit einem Limit von 20 Verbindungen, warten 80 Worker nur in einer Schlange davor.

In JavaScript braucht Arbeit, die wartet, keinen Pool: Der Event Loop wartet in einem Thread auf Tausende Anfragen (die Lektion async ist kein Thread). Ein Pool von Workern ist nur für Arbeit, die rechnet, und dann so groß wie die Zahl der Kerne.

Probier es aus

Ergänze

Wie viele Worker für eine Aufgabe, die waitMs wartet und computeMs rechnet?

function poolSize(cores, waitMs, computeMs) {
  return cores * (1 + );
}
Genauer, für Neugierige

Die Formel gilt auch für Worker: Arbeit, die in einem Worker wartet, lässt den Kern ebenfalls frei. Aber in JavaScript steckt Arbeit, die wartet, selten in einem Worker: Sie ist im Hauptthread mit await.

Übung

Die Rechnungen des Sales

Löse die Aufgabe in deinem Editor.

Nach dem Sale braucht jede Bestellung eine Rechnung. invoice(amount) erzeugt das PDF (hier eine Pause von 50 ms) und gibt den Betrag mit 20 % MwSt. zurück. Eine Bestellung mit einem Betrag von 0 oder weniger ist ungültig, und die Funktion wirft eine Exception mit dem Text „ungültiger Betrag“.

Die Bestellungen sind 1001 bis 1008 mit den Beträgen 50 €, 125 €, 80 €, 0 €, 240 €, 35 €, 90 € und 15 €. Erzeuge die Rechnungen in einem Pool aus 4 Workern. Gib sie in der Reihenfolge der Bestellungen aus, die ungültige mit dem Text des Fehlers. Am Ende: die Summe der guten Rechnungen und wie viele gescheitert sind.

Ausgabe
Bestellung 1001: 60 €
Bestellung 1002: 150 €
Bestellung 1003: 96 €
Bestellung 1004: Fehler: ungültiger Betrag
Bestellung 1005: 288 €
Bestellung 1006: 42 €
Bestellung 1007: 108 €
Bestellung 1008: 18 €
Gesamt: 762 €, mit Fehler: 1
Nimm WorkerPool aus der Lektion. Der Worker ruft invoice statt thumbnail auf.

Weitere Aufgaben

  • Eine begrenzte Warteschlange. Lass den Pool höchstens 10 Aufgaben in der Warteschlange warten. Was passiert mit der elften? (run wartet, wenn die Warteschlange voll ist)
  • Fortschritt. Gib nach jeder fertigen Rechnung „3 von 8 fertig“ aus, und am Ende noch einmal die geordnete Liste.
  • Ein Timeout. Eine Rechnung, die über 1 Sekunde dauert, gilt als gescheitert, ohne die anderen zu stoppen.

Test

Teste dich

1. Warum ein Pool aus ein paar Threads und nicht ein Thread für jedes hochgeladene Bild?
2. Ein Pool aus 3 Workern verkleinert 8 Bilder. Das dritte wird als Erstes fertig. Wo steht seine Größe, wenn du sie in Listenreihenfolge mit get, result, Promise.all oder sizes[i] liest?
3. Eine Aufgabe im Pool wirft eine Exception. Wann siehst du sie?
4. Der Server hat 8 Kerne. Jede E-Mail wartet 90 ms auf das Netzwerk und rechnet 10 ms. Etwa wie viele Worker halten die Kerne beschäftigt?
5. Wie viele Worker für Vorschaubilder, wenn die Arbeit nur rechnet und der Server 8 Kerne hat?

Erstellt von Ivo Gergov im Rahmen der Kampagne von Evo Tech für zugängliche Programmierausbildung ·