الـ Worker Threads والـ CPU Profiling: الشغل التقيل بره الـ event loop، والوقت راح فين

// load.js: أربع requests تقيلة مع بعض، و user تاني بيطلب fast كل عشرين ms
const B = "http://localhost:3000";
const t = async (p) => { const s = performance.now(); const r = await fetch(B + p); const body = await r.json(); return { ms: performance.now() - s, body }; };
await t("/fast"); // warm-up
const heavy = Array.from({ length: 4 }, () => t("/primes?n=40000"));
const fast = [];
for (let i = 0; i < 30; i++) { fast.push(t("/fast")); await new Promise((r) => setTimeout(r, 20)); }
const hs = await Promise.all(heavy);
const fs = (await Promise.all(fast)).map((x) => x.ms).sort((a, b) => a - b);
console.log(`primes: ${hs.map((h) => h.body.count).join("، ")} ← الأربعة خلصوا في ${Math.max(...hs.map((h) => h.ms)).toFixed(0)}ms`);
console.log(`fast: p50 ${fs[Math.floor(fs.length / 2)].toFixed(0)}ms | max ${fs.at(-1).toFixed(0)}ms`);

$ node server.js & S=$!; sleep 1; node load.js; kill $S
on 3000
primes: 4203، 4203، 4203، 4203 ← الأربعة خلصوا في 584ms
fast: p50 271ms | max 586ms

// bench.js: بنشغّل الشغل التقيل مرة، علشان نعمله profile
import { countPrimes, countPrimesFast } from "./primes.js";

let t = performance.now();
const a = countPrimes(40000);
console.log(`countPrimes(40000) = ${a} في ${(performance.now() - t).toFixed(0)}ms`);
t = performance.now();
const b = countPrimesFast(40000);
console.log(`countPrimesFast(40000) = ${b} في ${(performance.now() - t).toFixed(1)}ms`);

$ node bench.js
countPrimes(40000) = 4203 في 142ms
countPrimesFast(40000) = 4203 في 2.5ms

// cpu.js: worker pool للشغل التقيل، و قراية CPU profile
import { Worker } from "node:worker_threads";
import { availableParallelism } from "node:os";

export function createPool(file, { size = availableParallelism(), maxQueue = 100, timeoutMs = 5000 } = {}) {
  const idle = [], queue = [], current = new Map(); // worker ← task شغّال عليه
  let closed = false, nextId = 0;

  const spawn = () => {
    const w = new Worker(file);
    w.on("message", ({ id, result, error }) => {
      const t = current.get(w);
      if (!t || t.id !== id) return;
      finish(w, t);
      error ? t.reject(new Error(error)) : t.resolve(result);
      idle.push(w);
      dispatch();
    });
    w.on("error", (e) => replace(w, e)); // crash جُوّه الـ worker
    idle.push(w);
  };
  const finish = (w, t) => { clearTimeout(t.timer); current.delete(w); };
  const replace = (w, err) => {
    const t = current.get(w);
    if (t) { finish(w, t); t.reject(err); }
    const i = idle.indexOf(w);
    if (i >= 0) idle.splice(i, 1);
    w.terminate();
    if (!closed) { spawn(); dispatch(); }
  };
  const dispatch = () => {
    while (idle.length && queue.length) {
      const w = idle.pop(), t = queue.shift();
      t.timer = setTimeout(() => replace(w, new Error("timeout")), timeoutMs);
      current.set(w, t);
      w.postMessage({ id: t.id, data: t.data });
    }
  };
  for (let i = 0; i < size; i++) spawn();

  return {
    run(data) {
      if (closed) return Promise.reject(new Error("pool closed"));
      if (!idle.length && queue.length >= maxQueue) return Promise.reject(new Error("queue full")); // backpressure
      return new Promise((resolve, reject) => {
        queue.push({ id: ++nextId, data, resolve, reject });
        dispatch();
      });
    },
    stats: () => ({ size, busy: current.size, queued: queue.length }),
    async close() {
      closed = true;
      for (const t of queue.splice(0)) t.reject(new Error("pool closed"));
      const workers = [...idle, ...current.keys()];
      for (const [w, t] of current) { finish(w, t); t.reject(new Error("pool closed")); }
      await Promise.all(workers.map((w) => w.terminate()));
      idle.length = 0;
    },
  };
}

// CPU profile (من node --cpu-prof): مين أكل الوقت؟
// self = الوقت جُوّه الـ function نفسها. total = هي واللي نادتهم (عرض الـ bar في الـ flame graph)
export function topFunctions(profile, { n = 5, by = "self" } = {}) {
  const byId = new Map(profile.nodes.map((nd) => [nd.id, nd]));
  const parent = new Map();
  for (const nd of profile.nodes) for (const c of nd.children ?? []) parent.set(c, nd.id);
  const key = (nd) => { const f = nd.callFrame; return `${f.functionName || "(anonymous)"} ${(f.url || "").split("/").pop()}:${f.lineNumber + 1}`; };
  const rows = new Map();
  const row = (k) => rows.get(k) ?? rows.set(k, { name: k, selfMs: 0, totalMs: 0 }).get(k);
  let all = 0;
  profile.samples.forEach((id, i) => {
    const ms = (profile.timeDeltas[i] ?? 0) / 1000; // microseconds ← ms
    all += ms;
 
…

$ rm -rf prof; node --cpu-prof --cpu-prof-dir=prof bench.js && node top.js
countPrimes(40000) = 4203 في 144ms
countPrimesFast(40000) = 4203 في 2.5ms

أعلى خمسة بالـ self:
   84.1%  self  143ms  total  143ms  countPrimes primes.js:7
    6.4%  self   11ms  total   11ms  (program) :0
    2.6%  self    4ms  total    4ms  (idle) :0
    1.9%  self    3ms  total    5ms  compileForInternalLoader realm:383
    0.7%  self    1ms  total    1ms  wrappedFn errors:537

أعلى خمسة بالـ total:
  100.0%  self    0ms  total  170ms  (root) :0
   87.2%  self    0ms  total  149ms  (anonymous) bench.js:1
   87.2%  self    0ms  total  149ms  evaluate :0
   87.2%  self    0ms  total  149ms  run module_job:419
   84.1%  self  143ms  total  143ms  countPrimes primes.js:7

// server.js: الشغل التقيل في worker pool، والـ event loop فاضي للباقيين
import express from "express";
import { availableParallelism } from "node:os";
import { createPool } from "./cpu.js";

const pool = createPool(new URL("./primes-worker.js", import.meta.url), { size: availableParallelism(), maxQueue: 8, timeoutMs: 2000 });
const app = express();
app.get("/fast", (req, res) => res.json({ ok: true }));
app.get("/stats", (req, res) => res.json(pool.stats()));
app.get("/primes", async (req, res) => {
  const n = Math.min(Number(req.query.n ?? 40000), 100000);
  const data = req.query.op ? { op: req.query.op } : { n }; // op للمعمل بس: throw و forever
  try {
    res.json({ n, count: await pool.run(data) });
  } catch (e) {
    if (e.message === "queue full") return res.set("Retry-After", "1").status(503).json({ error: e.message });
    if (e.message === "timeout") return res.status(504).json({ error: e.message });
    res.status(500).json({ error: e.message });
  }
});
const server = app.listen(3000, () => console.log("on 3000"));
process.on("SIGTERM", () => server.close(() => pool.close()));

// primes.js: شغل CPU حقيقي، من غير أي I/O
export function isPrime(x) {
  if (x < 2) return false;
  for (let d = 2; d < x; d++) if (x % d === 0) return false; // بيجرّب كل الأرقام لحد x
  return true;
}
export function countPrimes(n) {
  let c = 0;
  for (let i = 0; i < n; i++) if (isPrime(i)) c++;
  return c;
}
// نفس النتيجة، بس بيجرّب لحد الجذر التربيعي بس
export function isPrimeFast(x) {
  if (x < 2) return false;
  for (let d = 2; d * d <= x; d++) if (x % d === 0) return false;
  return true;
}
export function countPrimesFast(n) {
  let c = 0;
  for (let i = 0; i < n; i++) if (isPrimeFast(i)) c++;
  return c;
}

// primes-worker.js: بيشتغل في thread لوحده. بياخد { id, data } ويرجّع { id, result } أو { id, error }
import { parentPort } from "node:worker_threads";
import { countPrimes } from "./primes.js";

parentPort.on("message", ({ id, data }) => {
  try {
    if (data.op === "throw") throw new Error("bad input");
    if (data.op === "forever") for (;;); // للمعمل: task ما بتخلصش
    parentPort.postMessage({ id, result: countPrimes(data.n) });
  } catch (e) {
    parentPort.postMessage({ id, error: e.message });
  }
});

// top.js: بيقرا آخر profile ويطبع أكتر functions أكلت وقت
import { readdirSync, readFileSync } from "node:fs";
import { topFunctions } from "./cpu.js";

const f = readdirSync("prof").filter((x) => x.endsWith(".cpuprofile")).sort().pop();
const p = JSON.parse(readFileSync(`prof/${f}`, "utf8"));
for (const by of ["self", "total"]) {
  console.log(`\nأعلى خمسة بالـ ${by}:`);
  for (const r of topFunctions(p, { n: 5, by })) console.log(`  ${r.percent.toFixed(1).padStart(5)}%  self ${r.selfMs.toFixed(0).padStart(4)}ms  total ${r.totalMs.toFixed(0).padStart(4)}ms  ${r.name}`);
}

import { spawnSync } from "node:child_process";
import { mkdtempSync, readdirSync, readFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { createPool, topFunctions } from "./cpu.js";

let ok = 0, all = 0;
const check = (name, cond, extra = "") => { all++; if (cond) ok++; console.log(cond ? "✅" : "❌", name, extra); };
const near = (a, b) => Math.abs(a - b) < 1e-6;
const err = (p) => p.then(() => "no error", (e) => e.message);
const B = "http://localhost:3000";

const fr = (id, name, line, children) => ({ id, callFrame: { functionName: name, url: "file:///app/x.js", lineNumber: line - 1 }, children });
const prof = {
  nodes: [fr(1, "(root)", 0, [2, 6]), fr(2, "main", 1, [3, 4]), fr(3, "a", 5), fr(4, "b", 9, [5]), fr(5, "b", 9), fr(6, "", 20)],
  samples: [3, 3, 4, 5, 2, 6], timeDeltas: [1000, 1000, 2000, 3000, 500, 500],
};
const self = topFunctions(prof);
check("self: b أول واحد بخمسة ms", self[0].name === "b x.js:9" && near(self[0].selfMs, 5));
check("self: a اتنين، و main نص ms", near(self.find((r) => r.name === "a x.js:5").selfMs, 2) && near(self.find((r) => r.name === "main x.js:1").selfMs, 0.5));
check("percent من الوقت الكلي", near(self[0].percent, (5 / 8) * 100));
check("function من غير اسم ← (anonymous)", self.some((r) => r.name === "(anonymous) x.js:20"));
const tot = topFunctions(prof, { by: "total", n: 2 });
check("total: root فوق بتمانية، وبعده main بسبعة ونص", tot.length === 2 && tot[0].name.startsWith("(root)") && near(tot[0].totalMs, 8) && near(tot[1].totalMs, 7.5));
check("recursion: b الـ total بتاعه خمسة مش تمانية", near(topFunctions(prof, { by: "total", n: 9 }).find((r) => r.name === "b x.js:9").totalMs, 5));

const dir = mkdtempSync(`${tmpdir()}/prof-`);
spawnSync(process.execPath, ["--cpu-prof", `--cpu-prof-dir=${dir}`, "bench.js"]);
const real = topFunctions(JSON.parse(readFileSync(`${dir}/${readdirSync(dir)[0]}`, "utf8")), { n: 1 });
check("profile حقيقي: أكتر function أكلت وقت countPrimes", real[0]?.name.startsWith("countPrimes primes.js"), real[0]?.name);

const W = new URL("./primes-worker.js", import.meta.url);
const pool = createPool(W, { size: 2, maxQueue: 4, timeoutMs: 1000 });
check("pool: countPrimes لمية ← خمسة وعشرين", (await pool.run({ n: 100 })) === 25);
let s = performance.now();
await pool.run({ n: 40000 });
const one = performance.now() - s;
s = performance.now();
const two = pool.run({ n: 40000 }), three = pool.run({ n: 40000 });
await new Promise((r) => setTimeout(r, 10));
check("stats: الاتنين busy", pool.stats().busy === 2 && pool.stats().size === 2);
await Promise.all([two, three]);
const par = performance.now() - s;
check("اتنين مع بعض على اتنين workers أسرع من ورا بعض", par < one * 1.6, `واحد ${one.toFixed(0)}ms، اتنين مع بعض ${par.toFixed(0)}ms`);
check("worker بيرمي error ← reject بنفس الرسالة", (await err(pool.run({ op: "throw" }))) === "bad input");
check("وبعدها الـ pool لسه شغّال", (await pool.run({ n: 10 })) === 4);
await pool.close();
check("بعد close ← pool closed", (await e
…