reuse worker processes across repos and recycle them at rss or repo limits
Signed-off-by: dawn <90008@klbr.net>
26 unmodified lines2728293031323334353697 unmodified lines13413513613713613813914014114214314414514614714814915015115215315415514114214314415615715815916016116215016316416516615415516716816917017117214 unmodified lines1871881891901781791801911921931941951961971981992002013 unmodified lines2052062072081921931941952092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602611971981992002012022622632642652664 unmodified lines26 unmodified lines jobs: { type: "string", default: "6" }, // repo processes, and the syntax highlighting threads each one runs workers: { type: "string", default: String(Math.max(1, Math.floor(availableParallelism() / 2))) }, highlighters: { type: "string", default: "0" }, "worker-max-repos": { type: "string", default: "12" }, "worker-max-rss-mb": { type: "string", default: "650" }, full: { type: "boolean", default: false }, ...BUDGET_OPTIONS }});97 unmodified lines } // biggest first, so the last worker to finish isn't the one holding ghostty plans.sort((a, b) => b.weight - a.weight);
const workers = Math.max(1, Math.min(Number(opts.workers), plans.length)); const workers = Math.min(Number(opts.workers), plans.length); const queue = [...plans]; const familyOf = (name) => name.toLowerCase().replace(/(-fork|\d+)$/, ""); const activeFamilies = new Set(); const takePlan = (lastFamily) => { if (queue.length === 0) return undefined; if (lastFamily) { const sameIdx = queue.findIndex((p) => familyOf(p.repo.name) === lastFamily); if (sameIdx !== -1) return queue.splice(sameIdx, 1)[0]; } const distinctIdx = queue.findIndex((p) => !activeFamilies.has(familyOf(p.repo.name))); const plan = queue.splice(distinctIdx !== -1 ? distinctIdx : 0, 1)[0]; activeFamilies.add(familyOf(plan.repo.name)); return plan; }; const totals = { rendered: 0, kept: 0, dropped: 0, over_budget: 0, pages: 0, bytes: 0, reads: 0 }; const failures = []; await Promise.all( Array.from({ length: workers }, async () => { for (let plan; (plan = queue.shift()) !== undefined; ) { const t = performance.now(); const result = await runWorker(children, plan, { hookPath, out, budget }).catch((error) => ({ error: String(error) })); Array.from({ length: workers }, () => runWorkerLoop(children, queue, takePlan, activeFamilies, familyOf, { hookPath, out, budget }, async (plan, result, wallMs) => { if (result.error) { failures.push(plan.repo.did); status.repos[plan.repo.did] = { error: result.error }; } else { nextRepos[plan.repo.did] = { generation: hook.generation, prefix: prefixOf(plan.repo), pages: result.pages }; status.repos[plan.repo.did] = { ...result, wall_ms: Math.round(performance.now() - t) }; status.repos[plan.repo.did] = { ...result, wall_ms: wallMs }; for (const key of Object.keys(totals)) totals[key] += result[key] ?? 0; } await writeStatus(); } }) }) ) );
if (failures.length === 0) { await atomicJson(path.join(stateDir, "knot.json"), {14 unmodified lines await writeStatus(summary); process.stdout.write(JSON.stringify({ ...summary, repos: status.repos }) + "\n", () => process.exit(failures.length === 0 ? 0 : 1));}
function runWorker(children, plan, { hookPath, out, budget }) { return new Promise((resolve, reject) => { const args = [hookPath, "--out", out, "--server", opts.server, "--jobs", opts.jobs, "--highlighters", opts.highlighters];async function runWorkerLoop(children, queue, takePlan, activeFamilies, familyOf, { hookPath, out, budget }, onRepoDone) { const maxRepos = Math.max(1, Number(opts["worker-max-repos"]) || 12); const maxRssMb = Math.max(128, Number(opts["worker-max-rss-mb"]) || 650); let lastFamily = null; while (queue.length > 0) { const hl = Number(opts.highlighters) > 0 && queue.every((p) => p.weight <= 12) ? "1" : opts.highlighters; const args = [hookPath, "--out", out, "--server", opts.server, "--jobs", opts.jobs, "--highlighters", hl]; if (opts.read) args.push("--read", opts.read); const child = fork(new URL("./worker.mjs", import.meta.url).pathname, args, { stdio: ["ignore", "inherit", "inherit", "ipc"], // SVFE_PAGES_PROFILE=<dir> writes a cpu profile per worker there3 unmodified lines ...(process.env.SVFE_PAGES_PROFILE ? ["--cpu-prof", "--cpu-prof-dir", process.env.SVFE_PAGES_PROFILE] : []) ] }); children.add(child); let result = null; child.on("message", (message) => { if (message.ready) child.send({ repo: plan.repo, work: plan.work, budget }); else result = message; await new Promise((resolve) => { let current = null; let tStart = 0; let served = 0; const next = () => { const plan = takePlan(lastFamily); if (!plan) { current = null; if (child.connected) child.disconnect(); return; } current = plan; lastFamily = familyOf(plan.repo.name); tStart = performance.now(); child.send({ repo: plan.repo, work: plan.work, budget }); }; child.on("message", async (message) => { if (message.ready) { next(); return; } const finished = current; const wallMs = Math.round(performance.now() - tStart); current = null; served++; const fam = familyOf(finished.repo.name); const hasSameFamilyNext = queue.some((p) => familyOf(p.repo.name) === fam); if (!hasSameFamilyNext) activeFamilies.delete(fam); const rssMb = message.rss_mb ?? 0; const { rss_mb: _rss, ...cleanMessage } = message; await onRepoDone(finished, cleanMessage, wallMs); const recycle = message.error || served >= maxRepos || rssMb >= (hasSameFamilyNext ? maxRssMb * 1.25 : maxRssMb); if (recycle) { if (child.connected) child.disconnect(); } else { next(); } }); child.on("exit", async (code, signal) => { children.delete(child); const failed = current ?? (served === 0 ? takePlan(lastFamily) : null); if (failed) { const wallMs = tStart ? Math.round(performance.now() - tStart) : 0; current = null; activeFamilies.delete(familyOf(failed.repo.name)); await onRepoDone(failed, { error: `worker for ${failed.repo.did} exited ${signal ?? code}` }, wallMs); } resolve(); }); }); child.on("exit", (code, signal) => { children.delete(child); if (result) resolve(result); else reject(new Error(`worker for ${plan.repo.did} exited ${signal ?? code}`)); }); }); }}
async function removeRepo(out, did, prior) { await rm(path.join(out, "site", did), { force: true });4 unmodified lines187 unmodified linesExpand all188189190191192193194195196197198199200201202251 unmodified linesExpand all45445545645745145845946046146246346446546646746846947023 unmodified lines494495496497483498499500501502503504505187 unmodified linesExpand all return seen;};
const renderRepo = async (repo, work, budget) => { reads.count = 0; reads.ms = 0; sizes.clear(); files.clear(); readCache.clear(); readCacheChars = 0;
const prefix = prefixOf(repo); const stateFile = path.join(out, "state", tokenOf(repo.did) + ".json"); const saved = readJson(stateFile); const savedAliases = saved?.aliases ?? [];251 unmodified linesExpand all // the state names the new generation now, so the old one can go for (const entry of await readdir(storeOf(repo.did))) { if (entry !== hook.generation) await rm(path.join(storeOf(repo.did), entry), { recursive: true, force: true }); } return { pages: Object.keys(pages).length, bytes, reads: reads.count, read_ms: Math.round(reads.ms), ...stats, render_ms: Math.round(stats.render_ms) }; return { pages: Object.keys(pages).length, bytes, reads: reads.count, read_ms: Math.round(reads.ms), ...stats, render_ms: Math.round(stats.render_ms), rss_mb: Math.round(process.memoryUsage().rss / 1_048_576) };};
// urls svfe serves the repo under: the handle form it links to, and the owner did form,// which hydrates and then redirects to the handle like the SPA does23 unmodified lines
process.on("message", async ({ repo, work, budget }) => { try { const result = await renderRepo(repo, work, budget); process.send(result, () => process.exit(0)); process.send(result); } catch (error) { const message = error instanceof Superseded ? "superseded" : String(error?.stack ?? error); process.send({ error: message }, () => process.exit(error instanceof Superseded ? 3 : 1)); }});process.on("disconnect", () => process.exit(0));process.send({ ready: true });