From d3cda0ea9ce7bcf8781e5771663e05d5e7dc672d Mon Sep 17 00:00:00 2001 From: Lann Martin Date: Sun, 23 Aug 2026 14:52:35 -0400 Subject: [PATCH] conformance: stripe the polyengine-deno leg across Deno Web Worker shards (#372) --- conformance/driver-ct/justfile | 9 +- conformance/driver-ct/polyengine/deno.json | 2 +- conformance/driver-ct/polyengine/run.ts | 165 +++++++++++++----- .../driver-ct/polyengine/shard-worker.ts | 127 ++++++++++++++ 4 files changed, 255 insertions(+), 48 deletions(-) create mode 100644 conformance/driver-ct/polyengine/shard-worker.ts diff --git a/conformance/driver-ct/justfile b/conformance/driver-ct/justfile index fbf70fc..25f939b 100644 --- a/conformance/driver-ct/justfile +++ b/conformance/driver-ct/justfile @@ -310,11 +310,10 @@ run-jco: build _jco-prepared # aggregate is what assesses it. run.ts still exits nonzero on any # runner-level error. # -# The suites run as two CONCURRENT processes: each process's case loop -# is single-threaded (ct-runner has no shard option yet — polymorph-components/polyengine#110 -# tracks one; #372 tracks the worker fan-out here), so suite-level -# concurrency is the available parallelism, hiding the signing leg's -# wall time under the shared leg's. +# The suites run as two CONCURRENT processes, and each process's case loop +# is ALSO striped across Deno Web Worker shards via ct-runner's `shard` +# option (run.ts's `--shards`, defaulting to the core count capped at 8), +# so both suite-level and case-level parallelism are in play. run-polyengine: build #!/usr/bin/env bash set -euo pipefail diff --git a/conformance/driver-ct/polyengine/deno.json b/conformance/driver-ct/polyengine/deno.json index 2ad6e0f..6861c0e 100644 --- a/conformance/driver-ct/polyengine/deno.json +++ b/conformance/driver-ct/polyengine/deno.json @@ -11,6 +11,6 @@ }, "minimumDependencyAge": { "age": "P1D", "exclude": ["jsr:@polyengine/*", "jsr:@polymorph/*"] }, "tasks": { - "check": "deno check run.ts browser/worker-entry.ts browser-bundle-entry.ts" + "check": "deno check run.ts shard-worker.ts browser/worker-entry.ts browser-bundle-entry.ts" } } \ No newline at end of file diff --git a/conformance/driver-ct/polyengine/run.ts b/conformance/driver-ct/polyengine/run.ts index d90e876..27c39cd 100644 --- a/conformance/driver-ct/polyengine/run.ts +++ b/conformance/driver-ct/polyengine/run.ts @@ -13,7 +13,7 @@ // `--missing sha1-checked` (shared suite) | SUITES.shared.missing // `--suite conformance-signing-guest-ct` | `--suite signing` // `envelope(target, suite)` | ct-runner's envelope (same identity) -// worker pool, `--jobs` | single-threaded case loop +// worker pool, `--jobs` | Deno Web Worker shard fan-out (`--shards`) // // polyengine is a runtime linker: unlike the jco legs there is no transpile // step, no generated tree, no npm install, and no engine flag — the @@ -21,7 +21,7 @@ // // just conformance-ct::run-polyengine # both legs // … run.ts [--translator ] [--suite shared|signing] [--only SUB] -// [--fresh-cases] [--jspi] +// [--fresh-cases] [--jspi] [--shards N] // // SUITE ARTIFACTS. Both legs run the BARE suites — the same components // the jco leg transpiles (jco/package.json's `transpile` / @@ -42,9 +42,9 @@ // A runner-level problem (translate error, inventory drift, missing // imports) still throws and exits nonzero. // -// CONTAINMENT MODE. This leg runs each suite on ONE component instance -// (ct-runner's `freshCases: false`), not the family's fresh-instance- -// per-case convention. That convention is an artifact of wasmtime +// CONTAINMENT MODE. This leg runs each suite on ONE component instance PER +// SHARD WORKER (ct-runner's `freshCases: false`), not the family's fresh- +// instance-per-case convention. That convention is an artifact of wasmtime // economics — a fresh instance is ~free there (precompiled module, // CoW memory image) — while under a runtime linker each fresh instance // re-copies the suite's ~14 MB of embedded vectors and re-lifts all @@ -68,8 +68,22 @@ // unreliable. Both suites measure zero traps and zero timeouts (every // verdict is provenance `returned`), and any poisoning event mid-run // makes later rows either loudly wrong or correct — never quietly -// green — per the KAT asymmetry above. When debugging any such run, -// `--fresh-cases` restores per-case containment. +// green — per the KAT asymmetry above. With shard workers, a poisoning +// event only affects the rest of that one shard's stripe, not the whole +// run. When debugging any such run, `--fresh-cases` restores per-case +// containment. +// +// SHARDING. Each suite's case loop is striped across `--shards` Deno Web +// Worker shards (default: core count capped at 8): case `i` in census +// order belongs to shard `i % count`, per upstream ct-runner's `shard` +// option (`@polyengine/ct-runner`'s `RunSuiteOptions.shard`). Each shard +// worker runs its own `runSuite` call and emits its own envelope and its +// own `{"segment-end":true}` terminator; the parent keeps exactly one +// envelope (asserting every shard's envelope is byte-identical — this +// also catches a shard reading a different artifact), reorders the rows +// back into census order using the per-row case index, writes the single +// terminator, and sums the per-shard `RunCounts`. `--shards 1` runs the +// whole suite unsharded in a single worker. // // MODULE-IDENTITY CONSTRAINT: this leg loads the embedder from // `deno.json`'s exact-pinned `@polyengine/runtime/embedder` entry. @@ -77,12 +91,8 @@ // any other place this process loads the embedder must resolve the same // version. `just conformance-ct::polyengine-pin-check` gates that. -import { Translator } from "@polyengine/runtime/shim"; -import type { ComponentArtifacts } from "@polyengine/runtime/embedder"; -import { runSuite } from "@polyengine/ct-runner"; -import { wasi } from "@polyengine/wasi"; -import { defaultTranslator } from "@polyengine/translator"; -import { webcryptoImports } from "../../../js/polyengine/src/mod.ts"; +import type { RunCounts } from "@polyengine/ct-runner"; +import type { ShardDone, ShardReply, ShardRequest } from "./shard-worker.ts"; // This file sits at conformance/driver-ct/polyengine/run.ts, so the repo root // is three levels up. @@ -96,7 +106,10 @@ const RESULTS = new URL("conformance/driver-ct/results/", ROOT); * `probe/large-stream` measures 6.1s on a workstation and 72s on the * 2-core Actions runner (run 31394527207) — a hardware ratio, not a * hang. 300s keeps the hang guard while clearing that worst case with - * ~4x margin; the next-slowest case is 10.3s on the same runner. + * ~4x margin; the next-slowest case is 10.3s on the same runner. Shard + * workers contend for cores, so per-case wall time under sharding can + * exceed the single-threaded measurements above; the 300s budget already + * covers that headroom. */ const CASE_TIMEOUT_MS = 300_000; @@ -150,14 +163,19 @@ interface Cli { missing?: string[]; /** Opt back into fresh-instance-per-case (see CONTAINMENT MODE). */ freshCases: boolean; + /** Shard worker count (see SHARDING). */ + shards: number; } +const DEFAULT_SHARDS = Math.min(navigator.hardwareConcurrency ?? 1, 8); + function parseArgs(argv: string[]): Cli { const cli: Cli = { jspi: false, target: "polyengine-deno", suites: [], freshCases: false, + shards: DEFAULT_SHARDS, }; for (let i = 0; i < argv.length; i++) { switch (argv[i]) { @@ -185,6 +203,15 @@ function parseArgs(argv: string[]): Cli { case "--fresh-cases": cli.freshCases = true; break; + case "--shards": { + const raw = argv[++i]; + const n = Number(raw); + if (!Number.isInteger(n) || n < 1) { + throw new Error(`--shards must be an integer >= 1, got '${raw}'`); + } + cli.shards = n; + break; + } default: throw new Error(`unknown argument ${argv[i]}`); } @@ -203,48 +230,102 @@ function parseArgs(argv: string[]): Cli { return cli; } -async function loadArtifacts( - translatorPath: string | undefined, - wasm: URL, -): Promise { - const translator = translatorPath - ? await Translator.create(await Deno.readFile(translatorPath)) - : await defaultTranslator(); - const componentBytes = await Deno.readFile(wasm); - const { plan, adapters } = translator.translate(componentBytes); - return { plan, componentBytes, adapters }; +/** Runs one shard worker to completion and resolves with its reply. */ +function runShard(req: ShardRequest): Promise { + const worker = new Worker(new URL("./shard-worker.ts", import.meta.url), { + type: "module", + }); + return new Promise((resolve, reject) => { + worker.onmessage = (event: MessageEvent) => { + worker.terminate(); + resolve(event.data); + }; + worker.onerror = (event: ErrorEvent) => { + worker.terminate(); + reject(event.error ?? new Error(event.message)); + }; + worker.postMessage(req); + }); +} + +function sumCounts(counts: RunCounts[]): RunCounts { + return counts.reduce( + (acc, c) => ({ + passed: acc.passed + c.passed, + failed: acc.failed + c.failed, + skipped: acc.skipped + c.skipped, + na: acc.na + c.na, + total: acc.total + c.total, + }), + { passed: 0, failed: 0, skipped: 0, na: 0, total: 0 }, + ); } async function runOne(spec: SuiteSpec, cli: Cli): Promise { - const artifacts = await loadArtifacts(cli.translator!, spec.wasm); - // The whole import surface: WASI (no ambient environment — the suites - // read none, exactly as the jco legs' `env: []`) plus every - // `polymorph:webcrypto/*` interface from the host module under test. - // ct-runner adds `polymorph:test/test-context` itself. - const imports = { - ...wasi({ cli: { env: {}, passthrough: false } }), - ...webcryptoImports(), - }; - const out = cli.out ?? spec.out.pathname; - const lines: string[] = []; - const started = performance.now(); - const counts = await runSuite(artifacts, { - imports, + const baseRequest: Omit = { + wasmPath: spec.wasm.pathname, + translatorPath: cli.translator, target: cli.target, suiteName: spec.name, only: cli.only, missing: cli.missing ?? spec.missing, caseTimeoutMs: CASE_TIMEOUT_MS, jspi: cli.jspi, - // One instance per suite run by default — see CONTAINMENT MODE. + // One instance per shard worker by default — see CONTAINMENT MODE. freshCases: cli.freshCases, - emit: (line) => lines.push(line), - }); + }; + + const started = performance.now(); + const replies = await Promise.all( + Array.from({ length: cli.shards }, (_, index) => + runShard( + cli.shards > 1 + ? { ...baseRequest, shard: { index, count: cli.shards } } + : { ...baseRequest }, + )), + ); + + const dones: ShardDone[] = []; + for (const [index, reply] of replies.entries()) { + if (reply.kind === "error") { + throw new Error(`shard ${index} failed: ${reply.error}`); + } + dones.push(reply); + } + + const envelope = dones[0].envelope; + const terminator = dones[0].terminator; + for (const [index, done] of dones.entries()) { + if (done.envelope !== envelope) { + throw new Error( + `shard ${index}'s envelope diverges from shard 0's (artifact mismatch?)`, + ); + } + if (done.terminator !== terminator) { + throw new Error(`shard ${index}'s terminator diverges from shard 0's`); + } + } + + const rows = dones.flatMap((d) => d.rows); + rows.sort((a, b) => a[0] - b[0]); + for (let i = 1; i < rows.length; i++) { + if (rows[i][0] <= rows[i - 1][0]) { + throw new Error( + `merge produced a non-increasing case index at position ${i} ` + + `(${rows[i - 1][0]} then ${rows[i][0]}) — duplicate shard rows`, + ); + } + } + + const lines = [envelope, ...rows.map(([, line]) => line), terminator]; + const out = cli.out ?? spec.out.pathname; await Deno.writeTextFile(out, lines.join("\n") + "\n"); + + const counts = sumCounts(dones.map((d) => d.counts)); console.error( `[${cli.target}/${spec.name}] ${counts.passed} passed | ${counts.failed} failed | ` + `${counts.skipped} skipped | ${counts.na} n/a (${counts.total} total) in ` + - `${((performance.now() - started) / 1000).toFixed(1)}s -> ${out}`, + `${((performance.now() - started) / 1000).toFixed(1)}s (${cli.shards} shards) -> ${out}`, ); } diff --git a/conformance/driver-ct/polyengine/shard-worker.ts b/conformance/driver-ct/polyengine/shard-worker.ts new file mode 100644 index 0000000..95941cc --- /dev/null +++ b/conformance/driver-ct/polyengine/shard-worker.ts @@ -0,0 +1,127 @@ +/// + +// Deno module worker running one shard of a suite: loads the suite +// artifacts, wires the host imports, and runs `runSuite` over its stripe +// (see run.ts's SHARDING section for the parent-side merge contract). + +import { Translator } from "@polyengine/runtime/shim"; +import type { ComponentArtifacts } from "@polyengine/runtime/embedder"; +import { runSuite, type RunCounts } from "@polyengine/ct-runner"; +import { wasi } from "@polyengine/wasi"; +import { defaultTranslator } from "@polyengine/translator"; +import { webcryptoImports } from "../../../js/polyengine/src/mod.ts"; + +/** Posted once by the parent to start this shard's run. */ +export interface ShardRequest { + wasmPath: string; + translatorPath?: string; + target: string; + suiteName: string; + only?: string; + missing: string[]; + caseTimeoutMs: number; + jspi: boolean; + freshCases: boolean; + shard?: { index: number; count: number }; +} + +/** This shard's complete result, ready for the parent to merge. */ +export interface ShardDone { + kind: "done"; + envelope: string; + terminator: string; + rows: [number, string][]; + counts: RunCounts; +} + +export interface ShardError { + kind: "error"; + error: string; +} + +export type ShardReply = ShardDone | ShardError; + +async function loadArtifacts( + translatorPath: string | undefined, + wasmPath: string, +): Promise { + const translator = translatorPath + ? await Translator.create(await Deno.readFile(translatorPath)) + : await defaultTranslator(); + const componentBytes = await Deno.readFile(wasmPath); + const { plan, adapters } = translator.translate(componentBytes); + return { plan, componentBytes, adapters }; +} + +self.onmessage = async (event: MessageEvent) => { + try { + const req = event.data; + const artifacts = await loadArtifacts(req.translatorPath, req.wasmPath); + // The whole import surface: WASI (no ambient environment) plus every + // `polymorph:webcrypto/*` interface from the host module under test. + // ct-runner adds `polymorph:test/test-context` itself. + const imports = { + ...wasi({ cli: { env: {}, passthrough: false } }), + ...webcryptoImports(), + }; + + let envelope: string | undefined; + let terminator: string | undefined; + const rows: [number, string][] = []; + + const counts = await runSuite(artifacts, { + imports, + target: req.target, + suiteName: req.suiteName, + only: req.only, + missing: req.missing, + caseTimeoutMs: req.caseTimeoutMs, + jspi: req.jspi, + freshCases: req.freshCases, + shard: req.shard, + emit: (line, caseIndex) => { + if (caseIndex !== undefined) { + rows.push([caseIndex, line]); + return; + } + if (envelope === undefined) { + envelope = line; + } else if (terminator === undefined) { + terminator = line; + } else { + throw new Error( + `unexpected third no-index emit line (unknown wire shape): ${line}`, + ); + } + }, + }); + + if (envelope === undefined || terminator === undefined) { + throw new Error( + "runSuite completed without emitting both an envelope and a terminator line", + ); + } + + const reply: ShardDone = { kind: "done", envelope, terminator, rows, counts }; + self.postMessage(reply); + } catch (e) { + const reply: ShardError = { + kind: "error", + error: String(e instanceof Error ? e.stack ?? e.message : e), + }; + self.postMessage(reply); + } +}; + +self.onunhandledrejection = (event: PromiseRejectionEvent) => { + event.preventDefault(); + const reply: ShardError = { + kind: "error", + error: String( + event.reason instanceof Error + ? event.reason.stack ?? event.reason.message + : event.reason, + ), + }; + self.postMessage(reply); +};