Към съдържанието
wannacode

Пул от нишки и задачи

Накратко: Пулът е няколко Worker нишки пред опашка от задачи, вместо нов Worker за всяка задача; в Node.js се пише като малък клас.

Проблем

Worker за всяка снимка

Магазинът пуска разпродажба. Продавачите качват снимки на продуктите, а за всяка трябва умалено копие за списъка с продукти. Смаляването е работа, която смята: минава през всеки пиксел (виж работа, която чака, и работа, която смята).

Първият вариант пуска Worker за всяка качена снимка:

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

С 20 снимки при теб всичко минава: „Готови: 20 умалени снимки“. В първия час на разпродажбата идват 10 000 наведнъж.

  • Всеки Worker е цял нов V8: своя памет, свой event loop, десетки MB. 10 000 Worker-а изяждат паметта на сървъра много преди да са свършили.
  • Дори да има памет, 10 000 Worker-а на 8 ядра само се редуват. Процесорът губи време да ги превключва, а всяка снимка става по-бавна.

Сървърът не забавя, а пада. И падналият сървър не обслужва и купувачите.

Идеята

Няколко работника и опашка

Пул от нишки (thread pool) е няколко работника, създадени веднъж, и опашка от задачи пред тях. Задачата е „смали тази снимка“. Свободният работник взима следващата задача от опашката, свършва я и се връща за нова.

  • Броят работници е постоянен. 10 000 снимки не правят 10 000 нишки, а 10 000 задачи в опашката. Задачата е малък обект, нишката е скъпа.
  • Работниците се ползват отново. Нишката не се създава и не умира за всяка снимка.
  • Опашката поема пика. Когато качванията са повече, отколкото работниците смогват, задачите чакат, вместо да сриват сървъра.

Това е производител и потребител: кодът, който приема качванията, слага задачи в опашката, а работниците са потребителите.

С пула вече се срещна в урока за нишките: runLimited, който пуска най-много няколко обещания наведнъж. Тук ще видим какво става вътре, как да подредиш резултатите, какво става с грешките и колко работника трябват.

Node изпълнява твоя JavaScript в една нишка със свой event loop. Това стига за работа, която чака: мрежата и файловете чакат извън нея. За работа, която смята, като смаляването, трябват истински нишки: Worker-и от node:worker_threads. Пул от Worker-и Node не дава готов, затова ето един малък. Задачите чакат в опашката, свободният Worker взима следващата:

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


  // Задачата влиза в опашката; обещанието се изпълнява, когато Worker я свърши.
  run(task) {
    return new Promise((resolve, reject) => {
      this.#queue.push({ task, resolve, reject });
      this.#next();
    });
  }


  // Свободен Worker взима следващата задача от опашката.
  #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()));
  }
}

Worker-ът е същият файл: в него isMainThread е false, а той чака съобщения и връща резултата или текста на грешката:

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

Виж го

Трима работника

Смаляването тук е пауза, 2 ms за всеки KB, а размерът идва от името на снимката. Така изходът е един и същ при всяко пускане:

// Смаляване на снимка: тук е пауза, а размерът идва от името.
function thumbnail(image) {
  if (image.startsWith("повредена")) {
    throw new Error("повреден файл");
  }
  const kb = 20 + 3 * image.length;
  // Atomics.wait спи синхронно: в Worker това е позволено.
  Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 2 * kb);
  return kb;
}
Atomics.wait с празен SharedArrayBuffer спи синхронно. В главната нишка това е забранено (тя трябва да остане свободна), а в Worker е позволено.

Осем снимки, пул от трима работника:

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} KB`);
await pool.close();
Стъпка по стъпка, с думи
  1. 8 снимки чакат. Пулът е 3 Worker-а: new WorkerPool(3, import.meta.url).
  2. pool.run слага всяка снимка в опашката. Трите Worker-а взимат първите три.
  3. Worker 3 свършва пръв: яке е третата снимка и резултатът отива на третото място. Работникът не спира: взима следващата задача.
  4. Опашката е празна. Worker 1 свърши и няма какво да вземе, затова чака. Другите довършват последните снимки.
  5. Promise.all дава размерите по реда на снимките, не по реда на свършване. Общо 379 KB; pool.close() спира Worker-ите.
const images = ["чанта.jpg", "обувки.jpg", "яке.jpg", "шапка.jpg", "часовник.jpg", "раница.jpg", "колан.jpg", "шал.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} KB`);
await pool.close();
Изход
чанта.jpg: 47 KB
обувки.jpg: 50 KB
яке.jpg: 41 KB
шапка.jpg: 47 KB
часовник.jpg: 56 KB
раница.jpg: 50 KB
колан.jpg: 47 KB
шал.jpg: 41 KB
Общо: 379 KB
  • new WorkerPool(3, import.meta.url) пуска три Worker-а с този файл.
  • pool.run(image) слага снимката в опашката и веднага дава обещание. map не чака нищо.
  • Promise.all чака всички и дава масив по реда на обещанията, каквото и да е свършило първо.
  • pool.close() спира Worker-ите. Без него програмата не свършва: живите Worker-и държат процеса.
По-точно, за любопитните

Node все пак има свой малък пул: libuv, библиотеката под Node, пази 4 нишки за работа, която не може да чака без нишка (файлове, crypto.scrypt, zlib). Твоят JavaScript не върви в тях.

В проект пул от Worker-и не се пише сам: пакетът Piscina прави същото, с ограничена опашка, таймаути и отказ на задачи.

Как е в другите езици: Java, C# и Python имат готов пул от нишки (ExecutorService, ThreadPool, ThreadPoolExecutor).

Пробвай

Намери грешката

Колега е пренаписал смаляването с пул, а разпродажбата не стана по-бърза. Шест снимки отнемат колкото без пул:

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(`Отне над 0,5 s: ${Date.now() - start > 500}`);
await pool.close();
Изход
Отне над 0,5 s: true

Избери реда с грешката.

Идеята

Резултатите по реда на входа

Задачите свършват в ред, който не знаеш: зависи от снимките, от натоварването, от това кой работник е бил свободен. Затова никога не печатай „по реда на свършване“, ако искаш един и същ изход.

Два начина дават еднакъв резултат при всяко пускане:

  • По реда на входа. Пазиш обещанията в списък по реда на задачите и ги чакаш в този ред. Така работи кодът отгоре.
  • Сбор или брой. Ако ти трябва само общото, редът няма значение: 47 + 50 е колкото 50 + 47.

Понякога искаш резултатите по реда на свършване: за лента „готови 3 от 8“, или за да покажеш първите готови снимки. продължение (await) за всяка задача поотделно ги дава така. Тогава събираш само неща, които не зависят от реда:

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(`Готови: ${count} от ${images.length}, общо: ${total} KB`);
await pool.close();
Изход
Готови: 8 от 8, общо: 379 KB
Всяка снимка има своя async функция, която чака своя резултат и го добавя. Тук total += kb е безопасно: JavaScript в главната нишка не прекъсва ред по средата. Но total += await pool.run(image) би било грешно: то прочита total преди await и записва след него. Точно това е race между await-ове.

Идеята

Грешка в задачата

Една от снимките е повредена и thumbnail хвърля изключение. То не излиза в нишката, която е пуснала задачата: тя отдавна е продължила. Пулът го хваща и го пази в обещанието за резултата. Излиза, когато чакаш този резултат:

const images = ["чанта.jpg", "обувки.jpg", "повредена.jpg", "яке.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}: грешка: ${result.reason.message}`);
  }
});
await pool.close();
Изход
чанта.jpg: 47 KB
обувки.jpg: 50 KB
повредена.jpg: грешка: повреден файл
яке.jpg: 41 KB
Promise.allSettled чака всички и дава за всяко обещание status: "fulfilled" с value или "rejected" с reason. Worker-ът не хвърля през границата на нишката: той праща текста на грешката, а WorkerPool прави от него Error.

Пробвай

Къде отиде грешката

Втората снимка е повредена. Какво ще отпечата програмата?

const images = ["чанта.jpg", "повредена.jpg", "яке.jpg"];
const pool = new WorkerPool(3, import.meta.url);
try {
  const sizes = await Promise.all(images.map((image) => pool.run(image)));
  console.log(`Размери: ${sizes.join(", ")}`);
} catch (error) {
  console.log(`Хванато: ${error.message}`);
}
await pool.close();
Отговор

Идеята

Колко работника

Зависи от това какво правят:

  • Работа, която смята (смаляване, PDF, компресия): колкото са ядрата. Девети работник на 8 ядра няма къде да върви, само се редува с другите.
  • Работа, която чака (имейл, плащане, база): повече. Докато един работник чака мрежата, ядрото е свободно за друг. Ако задачата чака W ms и смята C ms, на ядро се събират около 1 + W / C работника.
function poolSize(cores, waitMs, computeMs) {
  return cores * (1 + waitMs / computeMs);
}


console.log(`Снимки (само смятане): ${poolSize(8, 0, 50)}`);
console.log(`Имейли (90 ms чакане, 10 ms смятане): ${poolSize(8, 90, 10)}`);
Изход
Снимки (само смятане): 8
Имейли (90 ms чакане, 10 ms смятане): 80

Броят ядра дава os.availableParallelism(). Формулата е начало, не закон: измери. Ако имейлите ползват един сървър с ограничение от 20 връзки, 80 работника само ще чакат на опашка пред него.

В JavaScript работата, която чака, не иска пул: event loop-ът чака хиляди заявки в една нишка (урокът async). Пул от Worker-и трябва само за работа, която смята, и тогава е колкото ядрата.

Пробвай

Попълни

Колко работника за задача, която чака waitMs и смята computeMs?

function poolSize(cores, waitMs, computeMs) {
  return cores * (1 + );
}
По-точно, за любопитните

Формулата важи и за Worker-ите: работа, която чака вътре в Worker, също оставя ядрото свободно. Но в JavaScript работата, която чака, рядко е в Worker: тя е в главната нишка с await.

Упражнение

Фактурите на разпродажбата

Реши задачата в редактора си.

След разпродажбата всяка поръчка иска фактура. invoice(amount) прави PDF-а (тук е пауза от 50 ms) и връща сумата с 20% ДДС. Поръчка със сума 0 или по-малко е грешна и функцията хвърля изключение с текст „невалидна сума“.

Поръчките са 1001 до 1008 със суми 50, 125, 80, 0, 240, 35, 90 и 15 €. Направи фактурите в пул от 4 работника. Отпечатай ги по реда на поръчките, а грешната с текста на грешката. Накрая: сборът на добрите фактури и колко са с грешка.

Изход
Поръчка 1001: 60 €
Поръчка 1002: 150 €
Поръчка 1003: 96 €
Поръчка 1004: грешка: невалидна сума
Поръчка 1005: 288 €
Поръчка 1006: 42 €
Поръчка 1007: 108 €
Поръчка 1008: 18 €
Общо: 762 €, с грешка: 1
Ползвай WorkerPool от урока. Worker-ът вика invoice вместо thumbnail.

Още задачи

  • Ограничена опашка. Направи пула така, че в опашката да чакат най-много 10 задачи. Какво става с единадесетата? (run чака, ако опашката е пълна)
  • Напредък. Отпечатвай „готови 3 от 8“ след всяка свършена фактура, а накрая пак подредения списък.
  • Таймаут. Фактура, която се прави над 1 секунда, да се брои за грешна, без да спира останалите.

Тест

Провери се

1. Защо пул от няколко нишки, а не нишка за всяка качена снимка?
2. Пул от 3 работника смалява 8 снимки. Третата свършва първа. Къде е размерът ѝ, когато ги взимаш по реда на списъка с get, result, Promise.all или sizes[i]?
3. Задача в пула хвърля изключение. Кога го виждаш?
4. Сървърът има 8 ядра. Всеки имейл чака мрежата 90 ms и смята 10 ms. Около колко работника трябват, за да са заети ядрата?
5. Колко работника за смаляване на снимки, ако то само смята, а сървърът има 8 ядра?

Създаден от Иво Гергов като част от кампанията на Evo Tech за достъпно обучение по програмиране ·