write finished pages in rank order as they complete and cap worker heap size
Signed-off-by: dawn <90008@klbr.net>
25 unmodified lines26272829303031323334121 unmodified linesExpand all15615715815916016016116216316416516616716817 unmodified lines25 unmodified lines // pages in flight per repo process 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: "2" }, highlighters: { type: "string", default: "0" }, full: { type: "boolean", default: false }, ...BUDGET_OPTIONS }});121 unmodified linesExpand all 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 there execArgv: process.env.SVFE_PAGES_PROFILE ? ["--cpu-prof", "--cpu-prof-dir", process.env.SVFE_PAGES_PROFILE] : [] execArgv: [ "--max-old-space-size=128", "--max-semi-space-size=4", ...(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) => {17 unmodified lines3 unmodified lines4567898910111213141514151617181920232122232425128 unmodified linesExpand all15415515615715815916116016116316216316416516655 unmodified lines22222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126222826326426523123223323423523623723823924024124224324424524624724824925025125225325425525626626726826927027127227327427527627727827928028128225825926026126226326426526626726826927028328428528628728828929029129227227327427527627727827928028129329429529641 unmodified lines33833934034133134234334434534623 unmodified lines37037137237336337437537637737837938037038138238338438510 unmodified lines3 unmodified linesimport { readFileSync } from "node:fs";import { link, mkdir, readdir, rename, rm, symlink, writeFile } from "node:fs/promises";import path from "node:path";import { pathToFileURL } from "node:url";import { parseArgs, promisify } from "node:util";import { gzip as gzipCallback } from "node:zlib";import { parseArgs } from "node:util";import { gzipSync } from "node:zlib";import { rankOf } from "./budget.mjs";import { atomicJson, keyOfUrl, prefixOf, readJson, RENDERER_FORMAT, tokenOf } from "./common.mjs";import { canonicalHtml, linksOf } from "./html.mjs";
const gzip = promisify(gzipCallback);
const { values: opts, positionals } = parseArgs({ allowPositionals: true, options: { server: { type: "string" }, out: { type: "string" }, read: { type: "string" }, jobs: { type: "string", default: "6" }, highlighters: { type: "string", default: "2" } highlighters: { type: "string", default: "0" } }});const hook = JSON.parse(readFileSync(positionals[0], "utf8"));const out = path.resolve(opts.out);128 unmodified linesExpand allconst renderRepo = async (repo, work, budget) => { const prefix = prefixOf(repo); const stateFile = path.join(out, "state", tokenOf(repo.did) + ".json"); const saved = readJson(stateFile); const savedAliases = saved?.aliases ?? []; const prior = work === "all" ? null : saved; const pages = prior ? structuredClone(prior.pages) : {}; const pages = prior ? prior.pages : {}; // routes left out with the reads that failed them, retried when one of those changes const left = prior ? structuredClone(prior.left ?? {}) : {}; const left = prior ? (prior.left ?? {}) : {}; const dirty = new Set(prior ? work : []); const genDir = path.join(storeOf(repo.did), hook.generation); const tmpDir = genDir + ".tmp"; // a run that was superseded or killed leaves its .tmp behind55 unmodified lines room--; return true; }); const results = new Map(); let at = 0; const retire = () => { const ready = []; while (at < wanted.length && results.has(wanted[at])) { const route = wanted[at++]; const known = route in pages; const { deps, gz, links } = results.get(route); results.delete(route); if (known) bytes -= pages[route].bytes; if (!gz || bytes + gz.length > budget.bytes) { if (known) { delete pages[route]; stats.dropped++; } if (gz) overBudget.add(route); else left[route] = deps; ready.push({ route, gz: null }); continue; } bytes += gz.length; delete left[route]; const kept = links.filter((next) => keep(next)); pages[route] = { deps, bytes: gz.length, links: kept }; rendered.add(route); stats.rendered++; ready.push({ route, gz }); for (const next of kept) { if (next.startsWith(`${prefix}/commit/`)) continue; if (!seen.has(next)) { seen.add(next); todo.push(next); } } } return ready; }; await pool(wanted, async (route) => { const page = await renderPage(route); let page = await renderPage(route); stats.render_ms += page.ms; const deps = [...new Set(page.reads.map((r) => r.key))].sort(); if (!accepted(page)) return void results.set(route, { deps }); const gz = await gzip(canonicalHtml(page.html), { level: 9, memLevel: 9 }); // commit pages link their parents and trees at their own commit: only the log // and the front page decide which commits get pages const rawLinks = route.startsWith(`${prefix}/commit/`) ? [] : [...linksOf(page.html, origin, (pathname) => routeOf(prefix, pathname, repo))]; const links = commitSourceSet.has(route) ? rawLinks : rawLinks.filter((next) => !next.startsWith(`${prefix}/commit/`)); results.set(route, { deps, gz, links }); }); const writes = []; for (const route of wanted) { const known = route in pages; const { deps, gz, links } = results.get(route); if (known) bytes -= pages[route].bytes; if (!gz || bytes + gz.length > budget.bytes) { if (known) { delete pages[route]; stats.dropped++; } if (gz) overBudget.add(route); else left[route] = deps; writes.push(route); continue; if (!accepted(page)) { page = null; results.set(route, { deps }); } else { // commit pages link their parents and trees at their own commit: only the log // and the front page decide which commits get pages const rawLinks = route.startsWith(`${prefix}/commit/`) ? [] : [...linksOf(page.html, origin, (pathname) => routeOf(prefix, pathname, repo))]; const links = commitSourceSet.has(route) ? rawLinks : rawLinks.filter((next) => !next.startsWith(`${prefix}/commit/`)); const canonical = canonicalHtml(page.html); page = null; const gz = gzipSync(canonical, { level: 9, memLevel: 9 }); results.set(route, { deps, gz, links }); } bytes += gz.length; delete left[route]; const kept = links.filter((next) => keep(next)); pages[route] = { deps, bytes: gz.length, links: kept }; rendered.add(route); stats.rendered++; writes.push(route); for (const next of kept) { if (next.startsWith(`${prefix}/commit/`)) continue; if (!seen.has(next)) { seen.add(next); todo.push(next); } for (const item of retire()) { const file = fileFor(tmpDir, prefix, item.route); await rm(file, { force: true }); await rm(file + ".gz", { force: true }); if (!item.gz) continue; await mkdir(path.dirname(file), { recursive: true }); await link(empty, file); await writeFile(file + ".gz", item.gz); item.gz = null; } } await pool(writes, async (route) => { const file = fileFor(tmpDir, prefix, route); await rm(file, { force: true }); await rm(file + ".gz", { force: true }); const { gz } = results.get(route); if (!(route in pages)) return; await mkdir(path.dirname(file), { recursive: true }); await link(empty, file); await writeFile(file + ".gz", gz); }); if (commitsRoute in pages || commitsRoute in left || !seen.has(commitsRoute)) { for (const next of currentAllowedCommits()) { if (!seen.has(next)) {41 unmodified lines for (const route of overBudget) over.add(route); stats.over_budget = over.size; checkCurrent();
await publish(repo, saved, genDir, tmpDir); await publish(repo, savedAliases, genDir, tmpDir); await atomicJson(stateFile, { did: repo.did, prefix, aliases: aliasesOf(repo),23 unmodified lines};
// the store dir by rename, then the site link by rename, so nginx sees the old or the new// pages of this repo and never half of eachconst publish = async (repo, saved, genDir, tmpDir) => {const publish = async (repo, savedAliases, genDir, tmpDir) => { await rm(genDir, { recursive: true, force: true }); await rename(tmpDir, genDir); const site = path.join(out, "site", repo.did); await swapLink(genDir, site); const aliases = aliasesOf(repo); for (const alias of aliases) await swapLink(site, path.join(out, "www", alias)); for (const alias of saved?.aliases ?? []) { for (const alias of savedAliases) { if (!aliases.includes(alias)) await rm(path.join(out, "www", alias), { force: true }); }};
10 unmodified lines