الـ Worker Threads: شغل CPU تقيل من غير ما توقف الـ server

01 / ملخص الدرس

إزاي حساب CPU تقيل بيوقف الـ event loop (بنقيسه بنبضات setInterval)، ونقله لـ Worker بـ workerData و postMessage. الـ errors والـ exit code، و pool بعدد ثابت من الـ workers وطابور شغل، و availableParallelism. وإمتى الـ workers مش هي الحل (الـ I/O).

ضمن مسارباك إند Node.js للإنتاج — من تحت الصفر19 / 122

02 / شرح الدرس مكتوبًا

نص شرح المعلم والكود الذي كتبه ونتيجة تشغيله، بنفس ترتيب الدرس.

أهلًا بيك. في درس الـ event loop شفنا إن أي حساب sync تقيل بيوقف كل حاجة. النهارده هنقيس ده بنبضات قلب: setInterval بيزوّد عداد كل خمسين millisecond، ولو الـ loop شغال، العداد بيزيد. و fib dot js فيه function بتحسب Fibonacci بالطريقة البطيئة، علشان نعمل load حقيقي على الـ CPU. نحسب fib تمانية وتلاتين على الـ main thread. نشغّل.

main.js

import { fib } from "./fib.js";

let beats = 0;
const heart = setInterval(() => beats++, 50);

const t0 = Date.now();
const result = fib(38);
const ms = Date.now() - t0;

setTimeout(() => {
  clearInterval(heart);
  console.log(`fib(38) = ${result}، خد حوالي ${Math.round(ms / 100) * 100}ms`);
  console.log(`نبضات القلب أثناء الحساب: ${beats}`);
}, 0);

المخرجات

$ node main.js
fib(38) = 39088169، خد حوالي 500ms
نبضات القلب أثناء الحساب: 1

الحساب خد مئات الـ milliseconds، والمفروض يبقى فيه نبضة كل خمسين، بس طلعت واحدة بس! ودي كمان جت بعد ما الحساب خلص. يعني طول الوقت ده القلب كان واقف. لو ده server، كل الـ requests، حتى أبسط health check، كانت هتستنى. الحل: Worker Threads: thread تاني بالـ JavaScript engine بتاعه، و event loop بتاعه، والـ main thread بيكلمه برسايل. worker dot js: بياخد n من workerData، ويحسب، ويبعت النتيجة بـ postMessage. وفي main dot js: runFib بتعمل Worker جديد، وبتلفه في Promise: message يعني نجح، و error يعني رمى، و exit بـ code غير zero يعني مات. ونفس النبضات. نشغّل.

main.js

import { Worker } from "node:worker_threads";

function runFib(n) {
  return new Promise((resolve, reject) => {
    const w = new Worker(new URL("./worker.js", import.meta.url), { workerData: { n } });
    w.once("message", resolve);
    w.once("error", reject);
    w.once("exit", (code) => code !== 0 && reject(new Error(`worker exit ${code}`)));
  });
}

let beats = 0;
const heart = setInterval(() => beats++, 50);

const t0 = Date.now();
const result = await runFib(38);
clearInterval(heart);
console.log(`fib(38) = ${result}، خد حوالي ${Math.round((Date.now() - t0) / 100) * 100}ms`);
console.log(`نبضات القلب أثناء الحساب: ${beats > 3 ? "القلب فضل شغال" : beats}`);

المخرجات

$ node main.js
fib(38) = 39088169، خد حوالي 1000ms
نبضات القلب أثناء الحساب: القلب فضل شغال

نفس النتيجة، والقلب فضل شغال طول الحساب. الـ main thread كان فاضي يرد على أي حد، والـ worker هو اللي شايل الشغل التقيل. بس فيه تمن: عمل worker جديد بياخد وقت و memory، علشان بيشغّل engine كامل. فلو عندك مهام كتير، ما تعملش worker لكل مهمة. الحل: pool، يعني عدد تابت من الـ workers عايشين، وطابور مهام. Pool بيعمل workers بعدد size. run بتحط المهمة في الطابور. و next: لو فيه worker فاضي ومهمة مستنية، يديهاله، ولما يخلص يرجع للـ idle ويشوف اللي بعده. وهنا الـ worker مختلف: بيسمع على message ويرد، ويفضل عايش. أربع مهام، و pool فيه اتنين بس. نشغّل.

main.js

import { Worker } from "node:worker_threads";
import os from "node:os";

class Pool {
  #idle = [];
  #queue = [];
  constructor(file, size) {
    for (let i = 0; i < size; i++) this.#idle.push(new Worker(file));
  }
  run(data) {
    return new Promise((resolve, reject) => {
      this.#queue.push({ data, resolve, reject });
      this.#next();
    });
  }
  #next() {
    if (!this.#idle.length || !this.#queue.length) return;
    const w = this.#idle.pop();
    const job = this.#queue.shift();
    w.once("message", (r) => {
      this.#idle.push(w);
      job.resolve(r);
      this.#next();
    });
    w.postMessage(job.data);
  }
  close() {
    for (const w of this.#idle) w.terminate();
  }
}

const pool = new Pool(new URL("./pool-worker.js", import.meta.url), 2);
const t0 = Date.now();
const results = await Promise.all([34, 34, 34, 34].map((n) => pool.run(n)));
console.log(results, `${Math.round((Date.now() - t0) / 100) * 100}ms`);
pool.close();
console.log("cores:", os.availableParallelism() > 1 ? "أكتر من واحد" : 1);

المخرجات

$ node main.js
[ 5702887, 5702887, 5702887, 5702887 ] 300ms
cores: أكتر من واحد

الأربع نتايج، والوقت. اتنين اشتغلوا مع بعض، ولما خلصوا الاتنين التانيين أخدوا مكانهم. وعدد الـ workers المناسب غالبًا عدد الـ cores، و os dot availableParallelism بيقولك العدد. في الإنتاج فيه مكتبات جاهزة زي piscina بتعمل نفس الفكرة مع حاجات زيادة.

إمتى تستخدم الـ workers؟ للـ CPU: ضغط صور، hashing كتير، parsing ملفات ضخمة، تقارير. مش للـ I O: الـ database والشبكة والملفات Node بيعملهم async أصلًا من غير ما يوقف، والـ worker مش هيسرّعهم. وفيه بديل تاني: إنك تشغّل الشغل التقيل في process منفصل خالص، أو في queue زي BullMQ، وده هنعمله في قسم Redis. التحدي: worker ليه مهلة. runWithTimeout بتحسب في worker. ولو اتأخر عن المهلة، توقفه بجد بـ terminate، وترجّع timeout. وتشغّل حالتين مع بعض: واحدة سريعة، وواحدة مستحيل تلحق. والقلب لازم يفضل شغال. الخطوات في الكومنت.

تمرين

runWithTimeout(n, ms): Worker بـ workerData، setTimeout يعمل terminate ويرجّع "timeout"، و message يلغي التايمر. شغّل (25, 1000) و (45, 300) مع بعض.

الحل. Worker جديد، و timer: لو خلص، terminate و resolve بـ timeout. message و error بيلغوا الـ timer الأول. والاتنين مع بعض بـ Promise dot all، وفي الآخر clearInterval. نشغّل.

main.js

import { Worker } from "node:worker_threads";

let beats = 0;
const heart = setInterval(() => beats++, 50);

function runWithTimeout(n, ms) {
  return new Promise((resolve, reject) => {
    const w = new Worker(new URL("./worker.js", import.meta.url), { workerData: { n } });
    const timer = setTimeout(() => {
      w.terminate();
      resolve("timeout");
    }, ms);
    w.once("message", (r) => {
      clearTimeout(timer);
      resolve(r);
    });
    w.once("error", (err) => {
      clearTimeout(timer);
      reject(err);
    });
  });
}

const results = await Promise.all([runWithTimeout(25, 1000), runWithTimeout(45, 300)]);
clearInterval(heart);
console.log(results);
console.log(beats > 3 ? "القلب فضل شغال" : `beats: ${beats}`);

المخرجات

$ node main.js
[ 75025, 'timeout' ]
القلب فضل شغال

خمسة وسبعين ألف وخمسة وعشرين، و timeout، والقلب فضل شغال. والـ worker اللي اتأخر اتقفل فعلًا، وإلا كان البرنامج هيفضل مفتوح لحد ما يخلص. الدرس الجاي: الـ errors على مستوى الـ process: uncaughtException و unhandledRejection، وإمتى البرنامج لازم يموت ويقوم من جديد.

worker.js

import { parentPort, workerData } from "node:worker_threads";
import { fib } from "./fib.js";

parentPort.postMessage(fib(workerData.n));

pool-worker.js

import { parentPort } from "node:worker_threads";
import { fib } from "./fib.js";

// worker بيفضل عايش ويستقبل شغل كتير
parentPort.on("message", (n) => parentPort.postMessage(fib(n)));

03 / دروس المسار (122)