From b8c395ab2b0894131ef84089272884889bae0536 Mon Sep 17 00:00:00 2001 From: Brandon Payton Date: Sat, 11 Jul 2026 07:40:00 -0400 Subject: [PATCH 1/3] host: preserve fork-child pthread launches --- .../test/fork-child-thread.spec.ts | 67 ++++ docs/architecture.md | 13 +- host/src/browser-kernel-worker-entry.ts | 67 +++- host/src/kernel-worker.ts | 40 +- host/src/node-kernel-worker-entry.ts | 67 +++- host/src/worker-adapter-browser.ts | 7 +- host/src/worker-main.ts | 375 +++++++++--------- host/src/worker-protocol.ts | 14 + host/test/browser-worker-adapter.test.ts | 12 + host/test/fork-instrument-coverage.test.ts | 10 + host/test/multi-worker.test.ts | 114 ++++++ host/test/thread-wasm-patch.test.ts | 126 ++++++ programs/p_10_fork_child_creates_thread.c | 71 ++++ 13 files changed, 778 insertions(+), 205 deletions(-) create mode 100644 apps/browser-demos/test/fork-child-thread.spec.ts create mode 100644 host/test/thread-wasm-patch.test.ts create mode 100644 programs/p_10_fork_child_creates_thread.c diff --git a/apps/browser-demos/test/fork-child-thread.spec.ts b/apps/browser-demos/test/fork-child-thread.spec.ts new file mode 100644 index 0000000000..41fc3c904b --- /dev/null +++ b/apps/browser-demos/test/fork-child-thread.spec.ts @@ -0,0 +1,67 @@ +import { expect, test } from "@playwright/test"; +import { dirname, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; +import { tryResolveBinary } from "../../../host/src/binary-resolver"; + +const __dirname = dirname(fileURLToPath(import.meta.url)); +const browserKernelModulePath = resolve( + __dirname, + "../../../host/src/browser-kernel-host.ts", +); +const programPath = tryResolveBinary("programs/p_10_fork_child_creates_thread.wasm"); + +test("fork child creates and joins a pthread in the browser host", async ({ page, baseURL }) => { + test.setTimeout(120_000); + test.skip(!programPath, "P-10 fixture is not built"); + expect(baseURL).toBeTruthy(); + + await page.goto(new URL("/trap-signal-test.html", baseURL).href); + expect(await page.evaluate(() => crossOriginIsolated)).toBe(true); + + const result = await page.evaluate( + async ({ moduleUrl, programUrl }) => { + const { BrowserKernel } = await import(/* @vite-ignore */ moduleUrl); + const programResponse = await fetch(programUrl); + if (!programResponse.ok) { + throw new Error(`failed to fetch P-10 fixture: ${programResponse.status}`); + } + + let stdout = ""; + let stderr = ""; + const decoder = new TextDecoder(); + const kernel = new BrowserKernel({ + onStdout: (data: Uint8Array) => { stdout += decoder.decode(data); }, + onStderr: (data: Uint8Array) => { stderr += decoder.decode(data); }, + }); + + try { + await kernel.initFromImage({ vfsImage: "default" }); + const programBytes = await programResponse.arrayBuffer(); + let timeoutId: ReturnType | undefined; + const timeout = new Promise((_, reject) => { + timeoutId = setTimeout( + () => reject(new Error("P-10 browser run timed out")), + 30_000, + ); + }); + const exitCode = await Promise.race([ + kernel.spawn(programBytes, ["p_10_fork_child_creates_thread"]), + timeout, + ]).finally(() => clearTimeout(timeoutId)); + return { exitCode, stdout, stderr }; + } finally { + await kernel.destroy(); + } + }, + { + moduleUrl: new URL(`/@fs/${browserKernelModulePath}`, baseURL).href, + programUrl: new URL(`/@fs/${programPath}`, baseURL).href, + }, + ); + + expect(result.exitCode, `stderr=${result.stderr}\nstdout=${result.stdout}`).toBe(0); + expect(result.stdout).toContain("CHILD_THREAD: ok"); + expect(result.stdout).toContain("CHILD: joined"); + expect(result.stdout).toContain("PASS: P-10"); + expect(result.stderr).toBe(""); +}); diff --git a/docs/architecture.md b/docs/architecture.md index 5599e6b487..1ea5a2dc6c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -300,12 +300,13 @@ fell back to fork. ### clone() (threads) -1. User calls `clone(CLONE_VM | CLONE_THREAD, ...)` → kernel returns clone request -2. Host asks the kernel to reserve one dynamic pthread control slot in the same process address space -3. Host grows the process `WebAssembly.Memory` only far enough to cover that slot -4. Host spawns a new worker that shares the parent's `WebAssembly.Memory` -5. Thread worker runs `centralizedThreadWorkerMain`, calls `__wasm_thread_init` to set up TLS -6. Thread starts executing the given function pointer with the given argument +1. User calls `clone(CLONE_VM | CLONE_THREAD, ...)`; the kernel provisionally allocates a TID. +2. Host asks the kernel to reserve one dynamic pthread control slot in the same process address space. +3. Host grows the process `WebAssembly.Memory` only far enough to cover that slot. +4. Host builds the thread module: it removes `__wasm_init_memory`'s start section and only neutralizes `__wasm_call_ctors` when an exact export, name-section entry, or the preserved `__abi_version` linker wrapper identifies that function. A C module with no constructors is not otherwise rewritten. +5. Host spawns a new worker that shares the parent's `WebAssembly.Memory`. +6. Thread worker runs `centralizedThreadWorkerMain`, initializes TLS, stack, and channel state, resolves the requested table entry, and reports `thread_ready` without executing it yet. +7. The kernel worker publishes clone's TID and `CLONE_PARENT_SETTID` result, then releases the worker to execute the function pointer. A setup failure before `thread_ready` rolls back the provisional kernel thread, channel, control slot, and Worker before clone returns `EAGAIN`. Threads share memory with the parent (CLONE_VM) but have their own channel, fork-save scratch page, and TLS/control page. diff --git a/host/src/browser-kernel-worker-entry.ts b/host/src/browser-kernel-worker-entry.ts index 16e4143aba..2cf2d42459 100644 --- a/host/src/browser-kernel-worker-entry.ts +++ b/host/src/browser-kernel-worker-entry.ts @@ -62,6 +62,7 @@ if (typeof globalThis.setImmediate === "undefined") { import { CAPTURED_STDIO, CentralizedKernelWorker, TERMINAL_STDIO } from "./kernel-worker"; import type { + CloneLaunchResult, ForkFromThreadContext, ResolvedSpawnProgram, SpawnProgramResolution, @@ -1253,7 +1254,7 @@ async function handleClone( tlsPtr: number, ctidPtr: number, memory: WebAssembly.Memory, -): Promise { +): Promise { const processInfo = processes.get(pid); if (!processInfo) throw new Error(`Unknown pid ${pid} for clone`); threadedProcessPids.add(pid); @@ -1307,7 +1308,14 @@ async function handleClone( kernelAbiVersion: kernelWorker.getKernelAbiVersion(), }; - const threadWorker = workerAdapter.createWorker(threadInitData); + let threadWorker: ReturnType; + try { + threadWorker = workerAdapter.createWorker(threadInitData); + } catch (error) { + kernelWorker.removeChannel(pid, alloc.channelOffset); + processInfo.threadAllocator.free(alloc.basePage); + throw error; + } if (!threadWorkers.has(pid)) threadWorkers.set(pid, []); const threadEntry: ThreadWorkerInfo = { worker: threadWorker, @@ -1327,6 +1335,7 @@ async function handleClone( if (threads) { const idx = threads.indexOf(threadEntry); if (idx >= 0) threads.splice(idx, 1); + if (threads.length === 0) threadWorkers.delete(pid); } }; const terminateThreadEntry = (): Promise => { @@ -1340,9 +1349,37 @@ async function handleClone( }; threadExits.register(pid, alloc.channelOffset, terminateThreadEntry); - const failThread = (reason: string) => { + let launchState: "pending" | "ready" | "failed" = "pending"; + let finished = false; + let resolveLaunch!: (result: CloneLaunchResult) => void; + let rejectLaunch!: (error: Error) => void; + const launchPromise = new Promise((resolve, reject) => { + resolveLaunch = resolve; + rejectLaunch = reject; + }); + + const reportThreadFailure = (reason: string) => { const text = `[kernel-worker] pid=${pid} tid=${tid}: ${reason}\n`; post({ type: "stderr", pid, data: new TextEncoder().encode(text) }); + }; + + const failLaunch = (reason: string): boolean => { + if (launchState !== "pending") return false; + launchState = "failed"; + finished = true; + reportThreadFailure(reason); + kernelWorker.removeChannel(pid, alloc.channelOffset); + void terminateThreadEntry().then( + () => rejectLaunch(new Error(reason)), + (error) => rejectLaunch(error instanceof Error ? error : new Error(String(error))), + ); + return true; + }; + + const failThread = (reason: string) => { + if (failLaunch(reason) || finished) return; + finished = true; + reportThreadFailure(reason); const disposition = threadWorkerFailureDisposition(reason); kernelWorker.finalizeThreadExit(pid, tid, alloc.channelOffset); void terminateThreadEntry(); @@ -1353,7 +1390,23 @@ async function handleClone( threadWorker.on("message", (msg: unknown) => { const m = msg as WorkerToHostMessage; - if (m.type === "thread_exit") { + if (m.type === "thread_ready" && m.pid === pid && m.tid === tid) { + if (launchState !== "pending") return; + launchState = "ready"; + resolveLaunch({ + tid, + start: () => { + if (finished) return; + try { + threadWorker.postMessage({ type: "thread_start", pid, tid }); + } catch (error) { + failThread(`unable to start initialized worker: ${error}`); + } + }, + }); + } else if (m.type === "thread_exit") { + if (failLaunch("worker exited before reporting thread readiness")) return; + finished = true; void terminateThreadEntry(); } else if ((m as { type?: string }).type === "error") { // worker-main posted {type:"error"} — instantiation failure, top-level @@ -1365,8 +1418,12 @@ async function handleClone( console.error(`[kernel-worker] thread worker error pid=${pid} tid=${tid}:`, err.message); failThread(`worker error: ${err.message ?? err}`); }); + threadWorker.on("exit", (code: number) => { + if (finished || intentionallyTerminated.has(threadWorker as object)) return; + failThread(`worker exited before thread completion (code=${code})`); + }); - return tid; + return launchPromise; } function handleThreadExit(pid: number, channelOffset: number): boolean { diff --git a/host/src/kernel-worker.ts b/host/src/kernel-worker.ts index 9a182a4971..4caf5f0321 100644 --- a/host/src/kernel-worker.ts +++ b/host/src/kernel-worker.ts @@ -548,6 +548,12 @@ export interface SpawnResolveError { export type SpawnProgramResolution = ArrayBuffer | ResolvedSpawnProgram | SpawnResolveError; +export interface CloneLaunchResult { + tid: number; + /** Release the initialized Worker only after clone's result is visible. */ + start: () => void; +} + function isSpawnResolveError( resolution: SpawnProgramResolution, ): resolution is SpawnResolveError { @@ -629,9 +635,11 @@ export interface CentralizedKernelCallbacks { /** * Called when a process calls clone (thread creation). The callback should - * spawn a thread Worker sharing the parent's Memory. Returns the TID. + * initialize a thread Worker sharing the parent's Memory. A launch result's + * start callback is invoked only after clone success is visible to the guest. + * Numeric returns remain accepted for hosts without the two-phase handshake. */ - onClone?: (pid: number, tid: number, fnPtr: number, argPtr: number, stackPtr: number, tlsPtr: number, ctidPtr: number, memory: WebAssembly.Memory) => Promise; + onClone?: (pid: number, tid: number, fnPtr: number, argPtr: number, stackPtr: number, tlsPtr: number, ctidPtr: number, memory: WebAssembly.Memory) => Promise; /** * Called after a pthread channel reaches SYS_EXIT and the kernel worker has @@ -6320,16 +6328,12 @@ export class CentralizedKernelWorker { const tid = retVal; - // CLONE_PARENT_SETTID: write TID to ptid_ptr in process memory. - // The host writes this because ptid_ptr is in process memory, not kernel - // memory. + // The host writes CLONE_PARENT_SETTID because ptid_ptr is in process + // memory, not kernel memory. Delay it until the backing Worker is ready: + // Linux only publishes the child TID when clone succeeds. const CLONE_PARENT_SETTID = 0x00100000; const flags = origArgs[0]; const ptidPtr = origArgs[2]; - if (flags & CLONE_PARENT_SETTID && ptidPtr !== 0) { - const procView = new DataView(channel.memory.buffer); - procView.setInt32(ptidPtr, tid, true); - } // Read fnPtr and argPtr from the channel's CH_DATA area (written by kernel_clone stub) // These are always written as u32 by the glue (even on wasm64, table indices are i32) @@ -6348,7 +6352,10 @@ export class CentralizedKernelWorker { this.callbacks.onClone( channel.pid, tid, fnPtr, argPtr, stackPtr, tlsPtr, ctidPtr, channel.memory, - ).then((assignedTid) => { + ).then((launchResult) => { + const assignedTid = typeof launchResult === "number" + ? launchResult + : launchResult.tid; if (!this.processes.has(channel.pid)) { if (ctidPtr !== 0) { this.threadCtidPtrs.delete(`${channel.pid}:${tid}`); @@ -6359,13 +6366,24 @@ export class CentralizedKernelWorker { this.threadCtidPtrs.delete(`${channel.pid}:${tid}`); this.threadCtidPtrs.set(`${channel.pid}:${assignedTid}`, ctidPtr); } + if (flags & CLONE_PARENT_SETTID && ptidPtr !== 0) { + const procView = new DataView(channel.memory.buffer); + procView.setInt32(ptidPtr, assignedTid, true); + } this.completeChannel(channel, SYS_CLONE, origArgs, undefined, assignedTid, 0); + if (typeof launchResult !== "number") launchResult.start(); }).catch((err) => { if (ctidPtr !== 0) { this.threadCtidPtrs.delete(`${channel.pid}:${tid}`); } console.error(`[kernel-worker] onClone failed: ${err}`); - this.completeChannel(channel, SYS_CLONE, origArgs, undefined, -1, 12); // ENOMEM + // kernel_clone provisionally allocated ThreadInfo before the host could + // instantiate the backing Worker. Roll it back on launch failure so + // signals, /proc state, and future TID allocation reflect reality. + this.notifyThreadExit(channel.pid, tid); + if (this.processes.has(channel.pid)) { + this.completeChannel(channel, SYS_CLONE, origArgs, undefined, -1, 11); // EAGAIN + } }); } diff --git a/host/src/node-kernel-worker-entry.ts b/host/src/node-kernel-worker-entry.ts index 7775ddfa79..9920b8a35a 100644 --- a/host/src/node-kernel-worker-entry.ts +++ b/host/src/node-kernel-worker-entry.ts @@ -20,6 +20,7 @@ import { join } from "node:path"; import { fileURLToPath } from "node:url"; import { CAPTURED_STDIO, CentralizedKernelWorker, TERMINAL_STDIO } from "./kernel-worker"; import type { + CloneLaunchResult, ForkFromThreadContext, ResolvedSpawnProgram, SpawnProgramResolution, @@ -1061,7 +1062,7 @@ async function handleClone( tlsPtr: number, ctidPtr: number, memory: WebAssembly.Memory, -): Promise { +): Promise { const processInfo = processes.get(pid); if (!processInfo) throw new Error(`Unknown pid ${pid} for clone`); @@ -1109,7 +1110,14 @@ async function handleClone( kernelAbiVersion: kernelWorker.getKernelAbiVersion(), }; - const threadWorker = workerAdapter.createWorker(threadInitData); + let threadWorker: ReturnType; + try { + threadWorker = workerAdapter.createWorker(threadInitData); + } catch (error) { + kernelWorker.removeChannel(pid, alloc.channelOffset); + processInfo.threadAllocator.free(alloc.basePage); + throw error; + } if (!threadWorkers.has(pid)) threadWorkers.set(pid, []); const threadEntry: ThreadWorkerInfo = { worker: threadWorker, @@ -1129,6 +1137,7 @@ async function handleClone( if (threads) { const idx = threads.indexOf(threadEntry); if (idx >= 0) threads.splice(idx, 1); + if (threads.length === 0) threadWorkers.delete(pid); } }; const terminateThreadEntry = (): Promise => { @@ -1139,9 +1148,37 @@ async function handleClone( }; threadExits.register(pid, alloc.channelOffset, terminateThreadEntry); - const failThread = (reason: string) => { + let launchState: "pending" | "ready" | "failed" = "pending"; + let finished = false; + let resolveLaunch!: (result: CloneLaunchResult) => void; + let rejectLaunch!: (error: Error) => void; + const launchPromise = new Promise((resolve, reject) => { + resolveLaunch = resolve; + rejectLaunch = reject; + }); + + const reportThreadFailure = (reason: string) => { const text = `[kernel-worker] pid=${pid} tid=${tid}: ${reason}\n`; post({ type: "stderr", pid, data: new TextEncoder().encode(text) }); + }; + + const failLaunch = (reason: string): boolean => { + if (launchState !== "pending") return false; + launchState = "failed"; + finished = true; + reportThreadFailure(reason); + kernelWorker.removeChannel(pid, alloc.channelOffset); + void terminateThreadEntry().then( + () => rejectLaunch(new Error(reason)), + (error) => rejectLaunch(error instanceof Error ? error : new Error(String(error))), + ); + return true; + }; + + const failThread = (reason: string) => { + if (failLaunch(reason) || finished) return; + finished = true; + reportThreadFailure(reason); const disposition = threadWorkerFailureDisposition(reason); kernelWorker.finalizeThreadExit(pid, tid, alloc.channelOffset); void terminateThreadEntry(); @@ -1152,15 +1189,35 @@ async function handleClone( }; threadWorker.on("message", (msg: unknown) => { const m = msg as WorkerToHostMessage; - if (m.type === "thread_exit") { + if (m.type === "thread_ready" && m.pid === pid && m.tid === tid) { + if (launchState !== "pending") return; + launchState = "ready"; + resolveLaunch({ + tid, + start: () => { + if (finished) return; + try { + threadWorker.postMessage({ type: "thread_start", pid, tid }); + } catch (error) { + failThread(`unable to start initialized worker: ${error}`); + } + }, + }); + } else if (m.type === "thread_exit") { + if (failLaunch("worker exited before reporting thread readiness")) return; + finished = true; void terminateThreadEntry(); } else if (m.type === "error") { failThread(m.message); } }); threadWorker.on("error", (err: Error) => failThread(`worker error: ${err.message ?? err}`)); + threadWorker.on("exit", (code: number) => { + if (finished || intentionallyTerminated.has(threadWorker as object)) return; + failThread(`worker exited before thread completion (code=${code})`); + }); - return tid; + return launchPromise; } function handleThreadExit(pid: number, channelOffset: number): boolean { diff --git a/host/src/worker-adapter-browser.ts b/host/src/worker-adapter-browser.ts index c4c2917b9e..8690f35833 100644 --- a/host/src/worker-adapter-browser.ts +++ b/host/src/worker-adapter-browser.ts @@ -11,7 +11,12 @@ export class BrowserWorkerAdapter implements WorkerAdapter { const worker = new Worker(this.entryUrl, { type: "module" }); // Web Workers don't have workerData — send init data via postMessage const handle = new BrowserWorkerHandle(worker); - worker.postMessage(workerData); + try { + worker.postMessage(workerData); + } catch (error) { + worker.terminate(); + throw error; + } return handle; } } diff --git a/host/src/worker-main.ts b/host/src/worker-main.ts index 9c72ac3c3d..98c9f5c25e 100644 --- a/host/src/worker-main.ts +++ b/host/src/worker-main.ts @@ -1359,8 +1359,15 @@ function sendForkSyscall(memory: WebAssembly.Memory, channelOffset: number): num * * This function: * 1. Removes the Start section so `__wasm_init_memory` doesn't auto-run. - * 2. Finds the constructor function by scanning the known LLVM helper exports - * for their common call target and replaces that function body with a no-op. + * 2. Identifies `__wasm_call_ctors` only from authoritative linker evidence + * and replaces that function body with a no-op. + * + * `wasm-fork-instrument` deliberately preserves `__abi_version` in its raw + * linker-wrapper form. When constructors exist, wasm-ld prefixes that wrapper + * with `call $__wasm_call_ctors`; when they do not, the marker starts with its + * constant return. This distinction is load-bearing. A large C binary can + * have hundreds of unrelated exported call targets and no constructors at + * all, so selecting a merely shared or first call target corrupts the module. */ export function patchWasmForThread(bytes: ArrayBuffer): ArrayBuffer { const src = new Uint8Array(bytes); @@ -1390,11 +1397,36 @@ export function patchWasmForThread(bytes: ArrayBuffer): ArrayBuffer { return result; } - // Parse all sections + function readName(pos: number): [string, number] { + const [length, lengthBytes] = readLEB128(src, pos); + const start = pos + lengthBytes; + return [new TextDecoder().decode(src.subarray(start, start + length)), start + length]; + } + + function skipLimits(pos: number): number { + const [flags, flagsBytes] = readLEB128(src, pos); + pos += flagsBytes; + const [, minBytes] = readLEB128(src, pos); + pos += minBytes; + if (flags & 1) { + const [, maxBytes] = readLEB128(src, pos); + pos += maxBytes; + } + return pos; + } + + function skipValueType(pos: number): number { + const type = src[pos++]; + // Typed references encode a heap type after ref.null/ref. + if (type === 0x63 || type === 0x64) { + const [, heapTypeBytes] = readLEB128(src, pos); + pos += heapTypeBytes; + } + return pos; + } + interface Section { id: number; offset: number; totalSize: number; contentOffset: number; contentSize: number; } const sections: Section[] = []; - let numFuncImports = 0; - let hasStartSection = false; let offset = 8; while (offset < src.length) { @@ -1403,92 +1435,72 @@ export function patchWasmForThread(bytes: ArrayBuffer): ArrayBuffer { const contentOffset = offset + 1 + sizeBytes; const totalSize = 1 + sizeBytes + sectionSize; sections.push({ id: sectionId, offset, totalSize, contentOffset, contentSize: sectionSize }); - if (sectionId === 8) hasStartSection = true; offset += totalSize; } - if (!hasStartSection) return bytes; + const functionTypeIndices: number[] = []; + let numFuncImports = 0; + const exportFuncIndicesByName = new Map(); + let typeSection: Section | null = null; + let codeSection: Section | null = null; - // Count function imports from Import section (id=2) for (const sec of sections) { - if (sec.id === 2) { + if (sec.id === 1) { + typeSection = sec; + } else if (sec.id === 2) { let pos = sec.contentOffset; const [importCount, countBytes] = readLEB128(src, pos); pos += countBytes; for (let i = 0; i < importCount; i++) { - const [modLen, modLenBytes] = readLEB128(src, pos); - pos += modLenBytes + modLen; - const [fieldLen, fieldLenBytes] = readLEB128(src, pos); - pos += fieldLenBytes + fieldLen; + [, pos] = readName(pos); + [, pos] = readName(pos); const kind = src[pos++]; - if (kind === 0) { // function import + if (kind === 0) { + const [typeIndex, typeIndexBytes] = readLEB128(src, pos); + pos += typeIndexBytes; + functionTypeIndices.push(typeIndex); numFuncImports++; - const [, typeIdxBytes] = readLEB128(src, pos); - pos += typeIdxBytes; - } else if (kind === 1) { // table - pos++; // reftype - const flags = src[pos++]; - const [, minBytes] = readLEB128(src, pos); pos += minBytes; - if (flags & 1) { const [, maxBytes] = readLEB128(src, pos); pos += maxBytes; } - } else if (kind === 2) { // memory - const flags = src[pos++]; - const [, minBytes] = readLEB128(src, pos); pos += minBytes; - if (flags & 1) { const [, maxBytes] = readLEB128(src, pos); pos += maxBytes; } - } else if (kind === 3) { // global - pos++; // valtype - pos++; // mutability + } else if (kind === 1) { + pos = skipValueType(pos); + pos = skipLimits(pos); + } else if (kind === 2) { + pos = skipLimits(pos); + } else if (kind === 3) { + pos = skipValueType(pos) + 1; + } else if (kind === 4) { + pos++; // tag attribute + const [, typeIndexBytes] = readLEB128(src, pos); + pos += typeIndexBytes; } } - break; - } - } - - // Find the constructor function by looking at the exported helper wrappers. - // Plain lld output puts `call $__wasm_call_ctors` first. After - // wasm-fork-instrument, wrappers have a rewind prolog before the original - // body, so scan instructions and choose the call target shared by the known - // helper exports instead of assuming opcode 0 is the constructor call. - let ctorFuncIndex = -1; - let exportedFuncIndices: number[] = []; - const exportFuncIndicesByName = new Map(); - - // Collect exported function indices from Export section (id=7) - for (const sec of sections) { - if (sec.id === 7) { + } else if (sec.id === 3) { + let pos = sec.contentOffset; + const [functionCount, countBytes] = readLEB128(src, pos); + pos += countBytes; + for (let i = 0; i < functionCount; i++) { + const [typeIndex, typeIndexBytes] = readLEB128(src, pos); + pos += typeIndexBytes; + functionTypeIndices.push(typeIndex); + } + } else if (sec.id === 7) { let pos = sec.contentOffset; const [exportCount, countBytes] = readLEB128(src, pos); pos += countBytes; for (let i = 0; i < exportCount; i++) { - const [nameLen, nameLenBytes] = readLEB128(src, pos); - pos += nameLenBytes; - const name = new TextDecoder().decode(src.subarray(pos, pos + nameLen)); - pos += nameLen; + let name: string; + [name, pos] = readName(pos); const kind = src[pos++]; const [idx, idxBytes] = readLEB128(src, pos); pos += idxBytes; - if (kind === 0) { // function export - exportedFuncIndices.push(idx); - exportFuncIndicesByName.set(name, idx); - } + if (kind === 0) exportFuncIndicesByName.set(name, idx); } - break; + } else if (sec.id === 10) { + codeSection = sec; } } - function skipLEB(pos: number): number { - const [, n] = readLEB128(src, pos); - return pos + n; - } - - function skipMemArg(pos: number): number { - pos = skipLEB(pos); // alignment - return skipLEB(pos); // offset - } - - function getInstructionStartAndEnd( - codeSection: Section, - funcIndex: number, - ): { start: number; end: number } | null { + function getInstructionBounds(funcIndex: number): { start: number; end: number } | null { + if (!codeSection) return null; const codeEntry = funcIndex - numFuncImports; if (codeEntry < 0) return null; @@ -1509,130 +1521,112 @@ export function patchWasmForThread(bytes: ArrayBuffer): ArrayBuffer { const [localCount, localCountBytes] = readLEB128(src, pos); pos += localCountBytes; for (let i = 0; i < localCount; i++) { - pos = skipLEB(pos); // count - pos++; // valtype + const [, localRunLengthBytes] = readLEB128(src, pos); + pos += localRunLengthBytes; + pos = skipValueType(pos); } return { start: pos, end: bodyEnd }; } - function scanCallTargets(codeSection: Section, funcIndex: number): number[] { - const bounds = getInstructionStartAndEnd(codeSection, funcIndex); - if (!bounds) return []; - - const calls: number[] = []; - let pos = bounds.start; - while (pos < bounds.end) { - const op = src[pos++]; - if (op === 0x10) { // call - const [target, n] = readLEB128(src, pos); - pos += n; - calls.push(target); - } else if (op === 0x11 || op === 0x13) { // call_indirect / return_call_indirect - pos = skipLEB(pos); - pos = skipLEB(pos); - } else if (op === 0x12 || op === 0x14 || op === 0x15) { - pos = skipLEB(pos); - } else if (op === 0x02 || op === 0x03 || op === 0x04) { - // blocktype: empty marker, valtype, or signed type index. - pos = src[pos] === 0x40 || src[pos] >= 0x70 ? pos + 1 : skipLEB(pos); - } else if (op === 0x0c || op === 0x0d || (op >= 0x20 && op <= 0x26) || op === 0xd0 || op === 0xd2) { - pos = skipLEB(pos); - } else if (op === 0x0e) { // br_table - const [count, n] = readLEB128(src, pos); - pos += n; - for (let i = 0; i <= count; i++) pos = skipLEB(pos); - } else if (op >= 0x28 && op <= 0x3e) { - pos = skipMemArg(pos); - } else if (op === 0x3f || op === 0x40) { - pos++; - } else if (op === 0x41 || op === 0x42) { - pos = skipLEB(pos); - } else if (op === 0x43) { - pos += 4; - } else if (op === 0x44) { - pos += 8; - } else if (op === 0xfc) { - const [subop, n] = readLEB128(src, pos); - pos += n; - if (subop === 8 || subop === 10 || subop === 12 || subop === 14) { - pos = skipLEB(skipLEB(pos)); - } else if (subop >= 9 && subop <= 17) { - pos = skipLEB(pos); - } - } else if (op === 0xfe) { - pos = skipLEB(pos); - pos = skipMemArg(pos); - } else if (op === 0xfd) { - // SIMD is not expected in the helper wrappers. Stop before treating - // SIMD immediates as opcodes and collecting false call targets. - break; - } else { - // Most numeric, parametric, and control opcodes have no immediates. - } - } - return calls; - } + const ctorCandidates = new Map(); + const addCtorCandidate = (index: number | undefined, source: string) => { + if (index === undefined) return; + const sources = ctorCandidates.get(index) ?? []; + sources.push(source); + ctorCandidates.set(index, sources); + }; - // Find the Code section and identify a call target shared by LLVM helper exports. + addCtorCandidate(exportFuncIndicesByName.get("__wasm_call_ctors"), "function export"); + + // Unstripped modules can retain the exact synthetic function name even when + // it is not exported. Fork instrumentation appends to this same name map. for (const sec of sections) { - if (sec.id === 10 && exportedFuncIndices.length > 0) { - const helperNames = [ - "__wasm_init_tls", - "__abi_version", - "__get_channel_base_addr", - "_start", - "__wasm_thread_init", - ]; - const counts = new Map(); - let order = 0; - for (const name of helperNames) { - const funcIndex = exportFuncIndicesByName.get(name); - if (funcIndex === undefined) continue; - const perFunction = new Set(scanCallTargets(sec, funcIndex).filter(target => target >= numFuncImports)); - for (const target of perFunction) { - const entry = counts.get(target); - if (entry) { - entry.count++; - } else { - counts.set(target, { count: 1, firstOrder: order++ }); + if (sec.id !== 0) continue; + let pos = sec.contentOffset; + let customName: string; + [customName, pos] = readName(pos); + if (customName !== "name") continue; + const sectionEnd = sec.contentOffset + sec.contentSize; + while (pos < sectionEnd) { + const subsectionId = src[pos++]; + const [subsectionSize, sizeBytes] = readLEB128(src, pos); + pos += sizeBytes; + const subsectionEnd = pos + subsectionSize; + if (subsectionId === 1) { + const [nameCount, countBytes] = readLEB128(src, pos); + pos += countBytes; + for (let i = 0; i < nameCount; i++) { + const [funcIndex, indexBytes] = readLEB128(src, pos); + pos += indexBytes; + let functionName: string; + [functionName, pos] = readName(pos); + if (functionName === "__wasm_call_ctors") { + addCtorCandidate(funcIndex, "name section"); } } } + pos = subsectionEnd; + } + } - let best: { target: number; count: number; firstOrder: number } | null = null; - for (const [target, value] of counts) { - if ( - value.count >= 2 && - (!best || value.count > best.count || - (value.count === best.count && value.firstOrder < best.firstOrder)) - ) { - best = { target, count: value.count, firstOrder: value.firstOrder }; - } - } - - if (best) { - ctorFuncIndex = best.target; - } else { - // Fallback for very small legacy binaries: use the first call in an - // exported function whose body starts with that call. - for (const funcIndex of exportedFuncIndices) { - const bounds = getInstructionStartAndEnd(sec, funcIndex); - if (!bounds || src[bounds.start] !== 0x10) continue; - const [target] = readLEB128(src, bounds.start + 1); - if (target >= numFuncImports) { - ctorFuncIndex = target; - break; - } - } - } - break; + const abiMarkerIndex = exportFuncIndicesByName.get("__abi_version"); + if (abiMarkerIndex !== undefined && extractAbiVersion(bytes) !== null) { + const bounds = getInstructionBounds(abiMarkerIndex); + if (bounds && src[bounds.start] === 0x10) { + const [target] = readLEB128(src, bounds.start + 1); + addCtorCandidate(target, "__abi_version linker wrapper"); } } - const ctorCodeEntry = ctorFuncIndex >= 0 ? ctorFuncIndex - numFuncImports : -1; - if (ctorFuncIndex < 0) { - // No ctor found — still strip start section but can't neuter the ctor body + if (ctorCandidates.size > 1) { + const evidence = [...ctorCandidates] + .map(([index, sources]) => `${index} (${sources.join(", ")})`) + .join("; "); + throw new Error(`Conflicting __wasm_call_ctors evidence: ${evidence}`); + } + + const ctorFuncIndex = ctorCandidates.keys().next().value as number | undefined; + const hasStartSection = sections.some((sec) => sec.id === 8); + if (ctorFuncIndex === undefined && !hasStartSection) return bytes; + + let ctorCodeEntry = -1; + if (ctorFuncIndex !== undefined) { + ctorCodeEntry = ctorFuncIndex - numFuncImports; + const typeIndex = functionTypeIndices[ctorFuncIndex]; + if (ctorCodeEntry < 0 || !codeSection || typeIndex === undefined || !typeSection) { + throw new Error(`__wasm_call_ctors function ${ctorFuncIndex} has no defined function body`); + } + + let pos = typeSection.contentOffset; + const [typeCount, countBytes] = readLEB128(src, pos); + pos += countBytes; + let signature: { params: number; results: number } | null = null; + for (let i = 0; i < typeCount; i++) { + const form = src[pos++]; + if (form !== 0x60) break; + const [paramCount, paramCountBytes] = readLEB128(src, pos); + pos += paramCountBytes; + for (let param = 0; param < paramCount; param++) pos = skipValueType(pos); + const [resultCount, resultCountBytes] = readLEB128(src, pos); + pos += resultCountBytes; + for (let result = 0; result < resultCount; result++) pos = skipValueType(pos); + if (i === typeIndex) signature = { params: paramCount, results: resultCount }; + } + + const evidence = ctorCandidates.get(ctorFuncIndex) ?? []; + const markerProvesVoidSignature = evidence.includes("__abi_version linker wrapper"); + if (!signature && !markerProvesVoidSignature) { + throw new Error( + `Cannot inspect __wasm_call_ctors function ${ctorFuncIndex} signature from this type section`, + ); + } + if (signature && (signature.params !== 0 || signature.results !== 0)) { + throw new Error( + `__wasm_call_ctors function ${ctorFuncIndex} must have type () -> (), ` + + `found ${signature.params} parameter(s) and ${signature.results} result(s)`, + ); + } } // Build output: always skip Start section; optionally neuter constructor function @@ -1808,6 +1802,33 @@ export async function centralizedThreadWorkerMain( throw new Error(`Thread function at table index ${fnPtr} is null`); } + const startPromise = new Promise((resolve) => { + let started = false; + port.on("message", (raw: unknown) => { + const message = raw as { type?: string; pid?: number; tid?: number }; + if ( + !started && + message.type === "thread_start" && + message.pid === pid && + message.tid === tid + ) { + started = true; + resolve(); + } + }); + }); + + // clone() must not report success until the Worker has instantiated the + // patched module, initialized TLS/stack/channel state, and resolved the + // requested table entry. It waits here until the kernel has published the + // successful clone result to the caller's channel. + port.postMessage({ + type: "thread_ready", + pid, + tid, + } satisfies WorkerToHostMessage); + await startPromise; + const threadArg = ptrWidth === 8 ? BigInt(argPtr) : argPtr; let result = 0; if (hasForkInstrumentation) { diff --git a/host/src/worker-protocol.ts b/host/src/worker-protocol.ts index b7c4b127cd..b7663fdc6e 100644 --- a/host/src/worker-protocol.ts +++ b/host/src/worker-protocol.ts @@ -3,6 +3,7 @@ export type HostToWorkerMessage = | CentralizedWorkerInitMessage | CentralizedThreadInitMessage + | ThreadStartMessage | WorkerTerminateMessage | DeliverSignalMessage | ExecReplyMessage; @@ -86,6 +87,12 @@ export interface CentralizedThreadInitMessage { kernelAbiVersion?: number; } +export interface ThreadStartMessage { + type: "thread_start"; + pid: number; + tid: number; +} + export interface WorkerTerminateMessage { type: "terminate"; } @@ -94,6 +101,7 @@ export interface WorkerTerminateMessage { export type WorkerToHostMessage = | WorkerReadyMessage + | ThreadReadyMessage | WorkerExitMessage | ThreadExitMessage | WorkerErrorMessage @@ -106,6 +114,12 @@ export interface WorkerReadyMessage { pid: number; } +export interface ThreadReadyMessage { + type: "thread_ready"; + pid: number; + tid: number; +} + export interface WorkerExitMessage { type: "exit"; pid: number; diff --git a/host/test/browser-worker-adapter.test.ts b/host/test/browser-worker-adapter.test.ts index 0d2af487f4..829b30c418 100644 --- a/host/test/browser-worker-adapter.test.ts +++ b/host/test/browser-worker-adapter.test.ts @@ -95,6 +95,18 @@ describe("BrowserWorkerAdapter", () => { expect(lastMockWorker!.sentMessages[0]).toEqual(initData); }); + it("terminates a Worker whose init message cannot be cloned", () => { + const postMessage = vi.spyOn(MockBrowserWorker.prototype, "postMessage") + .mockImplementationOnce(() => { + throw new DOMException("could not clone", "DataCloneError"); + }); + const adapter = new BrowserWorkerAdapter("worker.js"); + + expect(() => adapter.createWorker({ pid: 42 })).toThrow(/could not clone/); + expect(lastMockWorker!.terminated).toBe(true); + postMessage.mockRestore(); + }); + it("should return a WorkerHandle", () => { const adapter = new BrowserWorkerAdapter("worker.js"); const handle = adapter.createWorker({}); diff --git a/host/test/fork-instrument-coverage.test.ts b/host/test/fork-instrument-coverage.test.ts index 2fe823daaa..8383dc39f5 100644 --- a/host/test/fork-instrument-coverage.test.ts +++ b/host/test/fork-instrument-coverage.test.ts @@ -470,6 +470,16 @@ describe("fork_instrument_coverage / P-* process & threading", () => { execPrograms: echoExecMap, }); }); + + // P-10: a fork child creates its first pthread. This is the Tcl notifier + // restart shape: patch the child program into a thread module, start the + // thread worker, and reclaim it before the child exits. + it("P-10 fork child creates and joins a pthread", async () => { + await runFixture("programs/p_10_fork_child_creates_thread.wasm", { + contains: ["PRE_FORK", "CHILD_THREAD: ok", "CHILD: joined", "PASS: P-10"], + timeout: 10_000, + }); + }); }); // --------------------------------------------------------------------------- diff --git a/host/test/multi-worker.test.ts b/host/test/multi-worker.test.ts index c5bdd1d6b4..be99be99b3 100644 --- a/host/test/multi-worker.test.ts +++ b/host/test/multi-worker.test.ts @@ -301,6 +301,120 @@ describe("CentralizedKernelWorker Process Management", () => { expect((kw as any).completeChannel).toHaveBeenCalled(); }); + it("publishes clone success before releasing the initialized thread worker", async () => { + const pid = 127; + const mainChannelOffset = WASM_PAGE_SIZE; + const tid = 80; + const ptidPtr = 0x00030000; + const memory = new WebAssembly.Memory({ initial: 16, maximum: 16, shared: true }); + const processView = new DataView(memory.buffer, mainChannelOffset); + processView.setUint32(CH_DATA, 11, true); + processView.setUint32(CH_DATA + 4, 22, true); + new DataView(memory.buffer).setInt32(ptidPtr, -1, true); + + const kernelMemory = new WebAssembly.Memory({ initial: 1, maximum: 1 }); + const kernelView = new DataView(kernelMemory.buffer); + const completeChannel = vi.fn(); + const start = vi.fn(() => { + expect(completeChannel).toHaveBeenCalledTimes(1); + expect(new DataView(memory.buffer).getInt32(ptidPtr, true)).toBe(tid); + }); + const onClone = vi.fn(async () => ({ tid, start })); + + const kw = Object.assign(Object.create(CentralizedKernelWorker.prototype), { + callbacks: { onClone }, + kernel: { toKernelPtr: (value: number | bigint) => Number(value) }, + kernelMemory, + scratchOffset: 0, + currentHandlePid: 0, + processes: new Map([[pid, { channels: [{ channelOffset: mainChannelOffset }] }]]), + threadCtidPtrs: new Map(), + completeChannel, + bindKernelTidForChannel: vi.fn(), + kernelInstance: { + exports: { + kernel_handle_channel: vi.fn(() => { + kernelView.setBigInt64(CH_RETURN, BigInt(tid), true); + kernelView.setUint32(CH_ERRNO, 0, true); + return 0; + }), + }, + }, + }); + + const CLONE_PARENT_SETTID = 0x00100000; + (kw as any).handleClone( + { pid, channelOffset: mainChannelOffset, memory }, + [CLONE_PARENT_SETTID, 0x00800000, ptidPtr, 0x00900000, 0, 0], + ); + + await Promise.resolve(); + await Promise.resolve(); + expect(start).toHaveBeenCalledTimes(1); + }); + + it("rolls back the provisional kernel thread when its Worker cannot launch", async () => { + const pid = 128; + const mainChannelOffset = WASM_PAGE_SIZE; + const tid = 81; + const ptidPtr = 0x00030000; + const ctidPtr = 0x00040000; + const memory = new WebAssembly.Memory({ initial: 16, maximum: 16, shared: true }); + const processView = new DataView(memory.buffer, mainChannelOffset); + processView.setUint32(CH_DATA, 11, true); + processView.setUint32(CH_DATA + 4, 22, true); + new DataView(memory.buffer).setInt32(ptidPtr, -1, true); + + const kernelMemory = new WebAssembly.Memory({ initial: 1, maximum: 1 }); + const kernelView = new DataView(kernelMemory.buffer); + const completeChannel = vi.fn(); + const notifyThreadExit = vi.fn(); + const onClone = vi.fn(async () => { + throw new WebAssembly.CompileError("invalid thread module"); + }); + const threadCtidPtrs = new Map(); + + const kw = Object.assign(Object.create(CentralizedKernelWorker.prototype), { + callbacks: { onClone }, + kernel: { toKernelPtr: (value: number | bigint) => Number(value) }, + kernelMemory, + scratchOffset: 0, + currentHandlePid: 0, + processes: new Map([[pid, { channels: [{ channelOffset: mainChannelOffset }] }]]), + threadCtidPtrs, + completeChannel, + notifyThreadExit, + bindKernelTidForChannel: vi.fn(), + kernelInstance: { + exports: { + kernel_handle_channel: vi.fn(() => { + kernelView.setBigInt64(CH_RETURN, BigInt(tid), true); + kernelView.setUint32(CH_ERRNO, 0, true); + return 0; + }), + }, + }, + }); + + const CLONE_PARENT_SETTID = 0x00100000; + const args = [CLONE_PARENT_SETTID, 0x00800000, ptidPtr, 0x00900000, ctidPtr, 0]; + (kw as any).handleClone({ pid, channelOffset: mainChannelOffset, memory }, args); + + await Promise.resolve(); + await Promise.resolve(); + expect(notifyThreadExit).toHaveBeenCalledWith(pid, tid); + expect(threadCtidPtrs.has(`${pid}:${tid}`)).toBe(false); + expect(new DataView(memory.buffer).getInt32(ptidPtr, true)).toBe(-1); + expect(completeChannel).toHaveBeenCalledWith( + expect.anything(), + ABI_SYSCALLS.Clone, + args, + undefined, + -1, + 11, + ); + }); + it("does not lower compact process max_addr when adding dynamic pthread channels", () => { const setMaxAddr = vi.fn(() => 0); const kw = Object.assign(Object.create(CentralizedKernelWorker.prototype), { diff --git a/host/test/thread-wasm-patch.test.ts b/host/test/thread-wasm-patch.test.ts new file mode 100644 index 0000000000..ed419360f1 --- /dev/null +++ b/host/test/thread-wasm-patch.test.ts @@ -0,0 +1,126 @@ +import { describe, expect, it } from "vitest"; +import { patchWasmForThread } from "../src/worker-main"; + +function uleb(value: number): number[] { + const encoded: number[] = []; + do { + let byte = value & 0x7f; + value >>>= 7; + if (value !== 0) byte |= 0x80; + encoded.push(byte); + } while (value !== 0); + return encoded; +} + +function section(id: number, content: number[]): number[] { + return [id, ...uleb(content.length), ...content]; +} + +function name(value: string): number[] { + const bytes = new TextEncoder().encode(value); + return [...uleb(bytes.length), ...bytes]; +} + +function moduleBytes(options: { + types: Array<{ params: number[]; results: number[] }>; + functionTypes: number[]; + bodies: number[][]; + exports: Array<{ name: string; index: number }>; + start?: number; +}): ArrayBuffer { + const bytes = [0x00, 0x61, 0x73, 0x6d, 0x01, 0x00, 0x00, 0x00]; + const typeContent = [...uleb(options.types.length)]; + for (const type of options.types) { + typeContent.push( + 0x60, + ...uleb(type.params.length), + ...type.params, + ...uleb(type.results.length), + ...type.results, + ); + } + bytes.push(...section(1, typeContent)); + bytes.push(...section(3, [ + ...uleb(options.functionTypes.length), + ...options.functionTypes.flatMap(uleb), + ])); + bytes.push(...section(7, [ + ...uleb(options.exports.length), + ...options.exports.flatMap((entry) => [ + ...name(entry.name), + 0x00, + ...uleb(entry.index), + ]), + ])); + if (options.start !== undefined) { + bytes.push(...section(8, uleb(options.start))); + } + const codeContent = [...uleb(options.bodies.length)]; + for (const instructions of options.bodies) { + const body = [0x00, ...instructions, 0x0b]; + codeContent.push(...uleb(body.length), ...body); + } + bytes.push(...section(10, codeContent)); + return new Uint8Array(bytes).buffer; +} + +const VOID = { params: [], results: [] }; +const I32_RESULT = { params: [], results: [0x7f] }; + +describe("patchWasmForThread", () => { + it("does not rewrite an unrelated exported call target when no constructors exist", async () => { + const original = moduleBytes({ + types: [VOID, I32_RESULT], + functionTypes: [0, 1, 1, 1], + bodies: [ + [], + [0x41, 0x07], + [0x10, 0x01], + [0x41, 0x12], + ], + exports: [ + { name: "p10_errno_address", index: 2 }, + { name: "__abi_version", index: 3 }, + ], + start: 0, + }); + + const patched = patchWasmForThread(original); + expect(WebAssembly.validate(patched)).toBe(true); + const { instance } = await WebAssembly.instantiate(patched); + expect((instance.exports.p10_errno_address as () => number)()).toBe(7); + }); + + it("neuters the constructor identified by the ABI linker wrapper", async () => { + const original = moduleBytes({ + types: [VOID, I32_RESULT], + functionTypes: [0, 0, 1], + bodies: [ + [], + [0x00], // unreachable if the constructor is not neutralized + [0x10, 0x01, 0x41, 0x12], + ], + exports: [{ name: "__abi_version", index: 2 }], + start: 0, + }); + + const patched = patchWasmForThread(original); + expect(WebAssembly.validate(patched)).toBe(true); + const { instance } = await WebAssembly.instantiate(patched); + expect((instance.exports.__abi_version as () => number)()).toBe(18); + }); + + it("rejects constructor evidence that does not point to a () -> () function", () => { + const original = moduleBytes({ + types: [VOID, I32_RESULT], + functionTypes: [0, 1], + bodies: [[], [0x41, 0x01]], + exports: [{ name: "__wasm_call_ctors", index: 1 }], + start: 0, + }); + + expect(() => patchWasmForThread(original)).toThrow( + /must have type \(\) -> \(\).*1 result/, + ); + }); +}); diff --git a/programs/p_10_fork_child_creates_thread.c b/programs/p_10_fork_child_creates_thread.c new file mode 100644 index 0000000000..f12a847031 --- /dev/null +++ b/programs/p_10_fork_child_creates_thread.c @@ -0,0 +1,71 @@ +// P-10 - fork child creates its first pthread. +// +// This covers the process-worker -> fork-child-worker -> pthread-worker path. +// The exported errno helper intentionally begins with a call returning an i32 +// pointer. It mirrors large C runtimes such as Tcl, where treating the first +// exported call target as __wasm_call_ctors corrupts the thread module. + +#include +#include +#include +#include +#include +#include + +__attribute__((export_name("p10_errno_address"), noinline)) +int *p10_errno_address(void) { + return &errno; +} + +static void *child_thread(void *arg) { + (void)arg; + printf("CHILD_THREAD: ok\n"); + fflush(stdout); + return (void *)(uintptr_t)42; +} + +int main(void) { + printf("PRE_FORK\n"); + fflush(stdout); + + pid_t pid = fork(); + if (pid < 0) { + printf("FAIL: fork errno=%d\n", errno); + return 1; + } + + if (pid == 0) { + pthread_t thread; + int rc = pthread_create(&thread, NULL, child_thread, NULL); + if (rc != 0) { + printf("FAIL: child pthread_create rc=%d\n", rc); + fflush(stdout); + _exit(2); + } + + void *result = NULL; + rc = pthread_join(thread, &result); + if (rc != 0 || (uintptr_t)result != 42) { + printf("FAIL: child pthread_join rc=%d result=%lu\n", + rc, (unsigned long)(uintptr_t)result); + fflush(stdout); + _exit(3); + } + printf("CHILD: joined\n"); + fflush(stdout); + _exit(0); + } + + int status = 0; + if (waitpid(pid, &status, 0) != pid) { + printf("FAIL: waitpid errno=%d\n", errno); + return 1; + } + if (!WIFEXITED(status) || WEXITSTATUS(status) != 0) { + printf("FAIL: child status=%d\n", status); + return 1; + } + + printf("PASS: P-10\n"); + return 0; +} From 61535ab20ca36062fa1e248dfead944b079f49d3 Mon Sep 17 00:00:00 2001 From: Brandon Payton Date: Sat, 11 Jul 2026 20:02:57 -0400 Subject: [PATCH 2/3] host: make pthread launch publication atomic Require every clone host callback to return an abortable ready-state handle. Validate parent-TID writes before provisional allocation, tie cleanup to process generations and memory identity, and keep all rollback before CH_COMPLETE. Retire pending launches during exec/exit on both Node and browser hosts, preserve custom PlatformIO and MariaDB runners, and cover invalid pointers, stale generations, publication failure, reused coordinators, and the P-10 custom-host path. --- docs/architecture.md | 4 +- docs/porting-guide.md | 7 +- host/src/browser-kernel-worker-entry.ts | 112 ++++--- host/src/kernel-worker.ts | 264 ++++++++++++----- host/src/node-kernel-worker-entry.ts | 115 +++++--- host/src/thread-exit-coordinator.ts | 12 +- host/test/centralized-test-helper.ts | 147 ++++++++- host/test/fork-instrument-coverage.test.ts | 15 + host/test/multi-worker.test.ts | 311 ++++++++++++++++++-- host/test/thread-exit-coordinator.test.ts | 14 + packages/registry/mariadb/test/run-tests.ts | 128 ++++++-- 11 files changed, 933 insertions(+), 196 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index 1ea5a2dc6c..30708c7027 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -303,10 +303,10 @@ fell back to fork. 1. User calls `clone(CLONE_VM | CLONE_THREAD, ...)`; the kernel provisionally allocates a TID. 2. Host asks the kernel to reserve one dynamic pthread control slot in the same process address space. 3. Host grows the process `WebAssembly.Memory` only far enough to cover that slot. -4. Host builds the thread module: it removes `__wasm_init_memory`'s start section and only neutralizes `__wasm_call_ctors` when an exact export, name-section entry, or the preserved `__abi_version` linker wrapper identifies that function. A C module with no constructors is not otherwise rewritten. +4. Host builds the thread module: it removes `__wasm_init_memory`'s start section and only neutralizes `__wasm_call_ctors` when an exact export or the preserved `__abi_version` linker wrapper identifies that function. Non-semantic name-section metadata is never sufficient. A C module with no constructors is not otherwise rewritten. 5. Host spawns a new worker that shares the parent's `WebAssembly.Memory`. 6. Thread worker runs `centralizedThreadWorkerMain`, initializes TLS, stack, and channel state, resolves the requested table entry, and reports `thread_ready` without executing it yet. -7. The kernel worker publishes clone's TID and `CLONE_PARENT_SETTID` result, then releases the worker to execute the function pointer. A setup failure before `thread_ready` rolls back the provisional kernel thread, channel, control slot, and Worker before clone returns `EAGAIN`. +7. The kernel worker revalidates the process generation, commits clone's TID and `CLONE_PARENT_SETTID` result, then releases the worker to execute the function pointer. Any failure before publication aborts the initialized Worker and reclaims the provisional kernel thread, channel, control slot, and host registry entry. After publication, failure is ordinary thread exit and cannot turn into a second clone completion. Threads share memory with the parent (CLONE_VM) but have their own channel, fork-save scratch page, and TLS/control page. diff --git a/docs/porting-guide.md b/docs/porting-guide.md index 2f081d93ab..b2044116d4 100644 --- a/docs/porting-guide.md +++ b/docs/porting-guide.md @@ -165,8 +165,11 @@ const kernelWorker = new CentralizedKernelWorker( // Return 0 on success, -2 (ENOENT) if not found }, onClone: async (pid, tid, fnPtr, argPtr, stackPtr, tlsPtr, ctidPtr, memory) => { - // Allocate thread channel, spawn thread worker - // Return tid on success + // Allocate the thread channel and initialize a Worker. Resolve only + // after it reports thread_ready, then return { tid, start, abort }: + // start posts thread_start after clone success is published; abort + // terminates the unpublished Worker and reclaims its channel and slot. + // See host/src/node-kernel-worker-entry.ts for the full lifecycle. }, onExit: (pid, status) => { // Handle process exit diff --git a/host/src/browser-kernel-worker-entry.ts b/host/src/browser-kernel-worker-entry.ts index 2cf2d42459..291649a0d7 100644 --- a/host/src/browser-kernel-worker-entry.ts +++ b/host/src/browser-kernel-worker-entry.ts @@ -138,6 +138,7 @@ interface ForkReplayContext { } interface ProcessInfo { + active: boolean; memory: WebAssembly.Memory; programBytes: ArrayBuffer; programModule?: WebAssembly.Module; @@ -222,6 +223,7 @@ interface ThreadWorkerInfo { tid: number; basePage: number; termination?: Promise; + cancel?: (reason: string) => Promise; } const threadWorkers = new Map(); const threadExits = new ThreadExitCoordinator(); @@ -303,10 +305,10 @@ async function terminateThreadWorkers(pid: number): Promise { threadWorkers.delete(pid); for (const t of threads) { await ( + t.cancel?.("process generation retired") ?? t.termination ?? terminateTrackedWorker(t.worker, THREADED_WORKER_TERMINATION_SETTLE_MS) ); - threadExits.release(pid, t.channelOffset); } } const ptyByPid = new Map(); @@ -859,6 +861,7 @@ async function handleSpawn(msg: Extract) const worker = workerAdapter.createWorker(initData); processes.set(pid, { + active: true, memory, programBytes, worker, @@ -1029,6 +1032,7 @@ async function handleFork( const childWorker = workerAdapter.createWorker(childInitData); processes.set(childPid, { + active: true, memory: childMemory, programBytes: parentInfo.programBytes, programModule: parentInfo.programModule, @@ -1058,17 +1062,23 @@ async function handleExec( const { programBytes: bytes, argv: launchArgv } = resolved; // Program found — run kernel exec setup + const oldInfo = processes.get(pid); + if (!oldInfo?.active) return -3; // ESRCH + oldInfo.active = false; const setupResult = kernelWorker.kernelExecSetup(pid); - if (setupResult < 0) return setupResult; + if (setupResult < 0) { + if (processes.get(pid) === oldInfo) oldInfo.active = true; + return setupResult; + } kernelWorker.prepareProcessForExec(pid); + await terminateThreadWorkers(pid); // Terminate old worker. Mark it as intentionally terminated *before* // calling terminate(): the synthesized "exit" event from // BrowserWorkerHandle would otherwise fire installProcessWorkerListeners' // crash detector and tear down the kernel's view of the still-alive // (post-exec) process. - const oldInfo = processes.get(pid); if (oldInfo?.worker) { intentionallyTerminated.add(oldInfo.worker as object); await oldInfo.worker.terminate().catch(() => {}); @@ -1126,6 +1136,7 @@ async function handleExec( threadModuleCache.delete(pid); processes.set(pid, { + active: true, memory: newMemory, programBytes: bytes, worker: newWorker, @@ -1230,6 +1241,7 @@ async function handlePosixSpawn( const newWorker = workerAdapter.createWorker(initData); processes.set(childPid, { + active: true, memory: newMemory, programBytes, worker: newWorker, @@ -1256,7 +1268,10 @@ async function handleClone( memory: WebAssembly.Memory, ): Promise { const processInfo = processes.get(pid); - if (!processInfo) throw new Error(`Unknown pid ${pid} for clone`); + if (!processInfo?.active || processInfo.memory !== memory) { + throw new Error(`Unknown or retired pid ${pid} for clone`); + } + const threadAllocator = processInfo.threadAllocator; threadedProcessPids.add(pid); // Auto-compile thread module if not already cached. @@ -1268,12 +1283,18 @@ async function handleClone( if (!threadModule) { const patched = patchWasmForThread(processInfo.programBytes); threadModule = await WebAssembly.compile(patched); + if (!processInfo.active || processes.get(pid) !== processInfo || processInfo.memory !== memory) { + throw new Error(`Process ${pid} changed while compiling its thread module`); + } threadModuleCache.set(pid, threadModule); } + if (!processInfo.active || processes.get(pid) !== processInfo || processInfo.memory !== memory) { + throw new Error(`Process ${pid} changed before thread allocation`); + } let alloc: ReturnType; try { - alloc = processInfo.threadAllocator.allocate(memory); + alloc = threadAllocator.allocate(memory); } catch (e) { const message = e instanceof Error ? e.message : String(e); post({ @@ -1312,8 +1333,8 @@ async function handleClone( try { threadWorker = workerAdapter.createWorker(threadInitData); } catch (error) { - kernelWorker.removeChannel(pid, alloc.channelOffset); - processInfo.threadAllocator.free(alloc.basePage); + kernelWorker.removeChannel(pid, alloc.channelOffset, memory); + threadAllocator.free(alloc.basePage); throw error; } if (!threadWorkers.has(pid)) threadWorkers.set(pid, []); @@ -1326,19 +1347,19 @@ async function handleClone( threadWorkers.get(pid)!.push(threadEntry); let reclaimed = false; - const reclaimThread = () => { + function reclaimThread(): void { if (reclaimed) return; reclaimed = true; - processInfo.threadAllocator.free(alloc.basePage); - threadExits.release(pid, alloc.channelOffset); + threadAllocator.free(alloc.basePage); + threadExits.release(pid, alloc.channelOffset, terminateThreadEntry); const threads = threadWorkers.get(pid); if (threads) { const idx = threads.indexOf(threadEntry); if (idx >= 0) threads.splice(idx, 1); if (threads.length === 0) threadWorkers.delete(pid); } - }; - const terminateThreadEntry = (): Promise => { + } + function terminateThreadEntry(): Promise { if (!threadEntry.termination) { threadEntry.termination = terminateTrackedWorker( threadWorker, @@ -1346,10 +1367,10 @@ async function handleClone( ).finally(reclaimThread); } return threadEntry.termination; - }; + } threadExits.register(pid, alloc.channelOffset, terminateThreadEntry); - let launchState: "pending" | "ready" | "failed" = "pending"; + let launchState: "pending" | "ready" | "started" | "failed" = "pending"; let finished = false; let resolveLaunch!: (result: CloneLaunchResult) => void; let rejectLaunch!: (error: Error) => void; @@ -1368,7 +1389,7 @@ async function handleClone( launchState = "failed"; finished = true; reportThreadFailure(reason); - kernelWorker.removeChannel(pid, alloc.channelOffset); + kernelWorker.removeChannel(pid, alloc.channelOffset, memory); void terminateThreadEntry().then( () => rejectLaunch(new Error(reason)), (error) => rejectLaunch(error instanceof Error ? error : new Error(String(error))), @@ -1376,12 +1397,30 @@ async function handleClone( return true; }; + const abortLaunch = async (): Promise => { + if (launchState === "failed" || finished) { + if (threadEntry.termination) await threadEntry.termination; + return; + } + launchState = "failed"; + finished = true; + kernelWorker.removeChannel(pid, alloc.channelOffset, memory); + await terminateThreadEntry(); + }; + threadEntry.cancel = async (reason: string): Promise => { + if (failLaunch(reason)) { + if (threadEntry.termination) await threadEntry.termination; + return; + } + await abortLaunch(); + }; + const failThread = (reason: string) => { if (failLaunch(reason) || finished) return; finished = true; reportThreadFailure(reason); const disposition = threadWorkerFailureDisposition(reason); - kernelWorker.finalizeThreadExit(pid, tid, alloc.channelOffset); + kernelWorker.finalizeThreadExit(pid, tid, alloc.channelOffset, memory); void terminateThreadEntry(); if (disposition.kind === "guest-fatal-trap") { handleExit(pid, disposition.exitStatus, disposition.signum); @@ -1390,22 +1429,38 @@ async function handleClone( threadWorker.on("message", (msg: unknown) => { const m = msg as WorkerToHostMessage; - if (m.type === "thread_ready" && m.pid === pid && m.tid === tid) { + if (m.type === "thread_ready") { + if (m.pid !== pid || m.tid !== tid) { + failLaunch( + `worker reported readiness for pid=${m.pid} tid=${m.tid}; expected pid=${pid} tid=${tid}`, + ); + return; + } if (launchState !== "pending") return; + if (!processInfo.active || processes.get(pid) !== processInfo || processInfo.memory !== memory) { + failLaunch(`process ${pid} changed before thread readiness`); + return; + } launchState = "ready"; resolveLaunch({ tid, start: () => { - if (finished) return; + if (finished || launchState !== "ready") return; + launchState = "started"; try { threadWorker.postMessage({ type: "thread_start", pid, tid }); } catch (error) { failThread(`unable to start initialized worker: ${error}`); } }, + abort: abortLaunch, }); } else if (m.type === "thread_exit") { if (failLaunch("worker exited before reporting thread readiness")) return; + if (launchState === "ready") { + failThread("worker exited before thread start"); + return; + } finished = true; void terminateThreadEntry(); } else if ((m as { type?: string }).type === "error") { @@ -1446,6 +1501,8 @@ async function finishProcessExit( if (processTeardowns.has(pid)) return; const info = processes.get(pid); + if (info) info.active = false; + kernelWorker.retireProcessGeneration(pid, info?.memory); reportNonzeroProcessExitDiagnostic(pid, exitStatus, "kernel process exit"); const threadedSettleMs = threadedProcessPids.has(pid) ? THREADED_WORKER_TERMINATION_SETTLE_MS @@ -1532,22 +1589,11 @@ function handleReadVfsFile(msg: Extract) { const pid = msg.pid; + const retiringInfo = processes.get(pid); + if (retiringInfo) retiringInfo.active = false; + kernelWorker.retireProcessGeneration(pid, retiringInfo?.memory); - // Terminate thread workers - const threads = threadWorkers.get(pid); - if (threads) { - for (const t of threads) { - await ( - t.termination ?? - terminateTrackedWorker(t.worker, THREADED_WORKER_TERMINATION_SETTLE_MS) - ); - try { - kernelWorker.notifyThreadExit(pid, t.tid); - kernelWorker.removeChannel(pid, t.channelOffset); - } catch {} - } - threadWorkers.delete(pid); - } + await terminateThreadWorkers(pid); // Terminate main process worker const info = processes.get(pid); diff --git a/host/src/kernel-worker.ts b/host/src/kernel-worker.ts index 4caf5f0321..300f5417bf 100644 --- a/host/src/kernel-worker.ts +++ b/host/src/kernel-worker.ts @@ -454,6 +454,8 @@ interface ProcessRegistration { pid: number; memory: WebAssembly.Memory; channels: ChannelInfo[]; + /** False as soon as exit or exec begins for this process generation. */ + active: boolean; /** Pointer width: 4 for wasm32, 8 for wasm64. */ ptrWidth: 4 | 8; /** @@ -550,8 +552,10 @@ export type SpawnProgramResolution = ArrayBuffer | ResolvedSpawnProgram | SpawnR export interface CloneLaunchResult { tid: number; - /** Release the initialized Worker only after clone's result is visible. */ + /** Release the initialized Worker only after clone's result is visible. Must not throw. */ start: () => void; + /** Tear down an initialized but unpublished Worker and all of its host state. */ + abort: () => Promise; } function isSpawnResolveError( @@ -635,11 +639,11 @@ export interface CentralizedKernelCallbacks { /** * Called when a process calls clone (thread creation). The callback should - * initialize a thread Worker sharing the parent's Memory. A launch result's - * start callback is invoked only after clone success is visible to the guest. - * Numeric returns remain accepted for hosts without the two-phase handshake. + * initialize a thread Worker sharing the parent's Memory. The start callback + * is invoked only after clone success is visible to the guest; abort is + * invoked if publication cannot complete. */ - onClone?: (pid: number, tid: number, fnPtr: number, argPtr: number, stackPtr: number, tlsPtr: number, ctidPtr: number, memory: WebAssembly.Memory) => Promise; + onClone?: (pid: number, tid: number, fnPtr: number, argPtr: number, stackPtr: number, tlsPtr: number, ctidPtr: number, memory: WebAssembly.Memory) => Promise; /** * Called after a pthread channel reaches SYS_EXIT and the kernel worker has @@ -743,8 +747,11 @@ export class CentralizedKernelWorker { retVal: number; errVal: number; }>(); - /** Maps "pid:tid" to ctidPtr for CLONE_CHILD_CLEARTID on thread exit */ - private threadCtidPtrs = new Map(); + /** Generation-tagged CLONE_CHILD_CLEARTID state for each live pthread. */ + private threadCtidPtrs = new Map(); /** TCP listeners: "pid:fd" → { server, pid, port, connections } */ private tcpListeners = new Map ch.pid !== pid); @@ -1640,6 +1663,8 @@ export class CentralizedKernelWorker { } deactivateProcess(pid: number): void { + const registration = this.processes.get(pid); + if (registration) this.retireProcessGeneration(pid, registration.memory); this.activeChannels = this.activeChannels.filter((ch) => ch.pid !== pid); this.processes.delete(pid); this.stdinFinite.delete(pid); @@ -1694,6 +1719,8 @@ export class CentralizedKernelWorker { * Does NOT cancel timers (POSIX: timers are preserved across exec). */ prepareProcessForExec(pid: number): void { + const registration = this.processes.get(pid); + if (registration) this.retireProcessGeneration(pid, registration.memory); // Remove channels from active list (stops listening on old memory) this.activeChannels = this.activeChannels.filter((ch) => ch.pid !== pid); @@ -1781,9 +1808,17 @@ export class CentralizedKernelWorker { /** * Remove a channel from a process registration (e.g. when a thread exits). */ - removeChannel(pid: number, channelOffset: number): void { + removeChannel( + pid: number, + channelOffset: number, + expectedMemory?: WebAssembly.Memory, + ): void { const registration = this.processes.get(pid); - if (!registration) return; + if ( + !registration || + !registration.active || + (expectedMemory && registration.memory !== expectedMemory) + ) return; registration.channels = registration.channels.filter( (ch) => ch.channelOffset !== channelOffset, @@ -2656,9 +2691,8 @@ export class CentralizedKernelWorker { argDescs: SyscallArgDesc[] | undefined, retVal: number, errVal: number, + beforePublish?: () => void, ): void { - const processView = new DataView(channel.memory.buffer, channel.channelOffset); - // Copy output data from kernel scratch back to process memory if (argDescs) { const processMem = new Uint8Array(channel.memory.buffer); @@ -2732,13 +2766,6 @@ export class CentralizedKernelWorker { } } - // Clear handling flag (channel is done — poller can pick it up for next syscall) - channel.handling = false; - - // Write result to process channel - processView.setBigInt64(CH_RETURN, BigInt(retVal), true); - processView.setUint32(CH_ERRNO, errVal, true); - // Cancel any pending socket timeout timer for this channel this.clearSocketTimeout(channel); @@ -2750,17 +2777,35 @@ export class CentralizedKernelWorker { // response data to the browser without waiting for the next pump cycle this.flushTcpSendPipes(channel.pid); + // Everything above may fail while the syscall is still unpublished. The + // optional hook lets clone commit CLONE_PARENT_SETTID in the same final, + // non-throwing region as its return value and CH_COMPLETE transition. + const processView = new DataView(channel.memory.buffer, channel.channelOffset); + beforePublish?.(); + // Clear only after every fallible pre-publication step. Otherwise a + // polling host can re-enter the still-PENDING syscall while rollback runs. + channel.handling = false; + processView.setBigInt64(CH_RETURN, BigInt(retVal), true); + processView.setUint32(CH_ERRNO, errVal, true); + // Set status to COMPLETE and notify process const i32View = new Int32Array(channel.memory.buffer, channel.channelOffset); Atomics.store(i32View, CH_STATUS / 4, CH_COMPLETE); Atomics.notify(i32View, CH_STATUS / 4, 1); - - // Drain kernel wakeup events and process targeted wakeups. - this.drainAndProcessWakeupEvents(); - - // Re-listen for next syscall - this.relistenChannel(channel); + // CH_COMPLETE is the syscall's publication boundary. Cleanup after this + // point must not throw back into callers that could otherwise attempt a + // second completion or roll back state the guest has already observed. + try { + this.drainAndProcessWakeupEvents(); + } catch (error) { + console.error("[kernel-worker] post-completion wakeup drain failed:", error); + } + try { + this.relistenChannel(channel); + } catch (error) { + console.error("[kernel-worker] post-completion relisten failed:", error); + } } /** @@ -6299,6 +6344,29 @@ export class CentralizedKernelWorker { this.completeChannel(channel, SYS_CLONE, origArgs, undefined, -1, 38); return; } + const processRegistration = this.processes.get(channel.pid); + if (!processRegistration || !processRegistration.active) { + this.completeChannel(channel, SYS_CLONE, origArgs, undefined, -1, 3); // ESRCH + return; + } + const CLONE_PARENT_SETTID = 0x00100000; + const flags = origArgs[0]; + const ptidPtr = origArgs[2]; + const parentTidPointerIsValid = () => + (flags & CLONE_PARENT_SETTID) === 0 || ( + Number.isSafeInteger(ptidPtr) && + ptidPtr > 0 && + ptidPtr <= channel.memory.buffer.byteLength - 4 + ); + if (!parentTidPointerIsValid()) { + this.completeChannel(channel, SYS_CLONE, origArgs, undefined, -1, 14); // EFAULT + return; + } + const isCurrentGeneration = () => + processRegistration.active && + this.processes.get(channel.pid) === processRegistration && + processRegistration.memory === channel.memory && + processRegistration.channels.includes(channel); // Route through kernel_handle_channel — the kernel allocates a TID and // stores ThreadInfo. The dispatch table remaps args correctly. @@ -6328,13 +6396,6 @@ export class CentralizedKernelWorker { const tid = retVal; - // The host writes CLONE_PARENT_SETTID because ptid_ptr is in process - // memory, not kernel memory. Delay it until the backing Worker is ready: - // Linux only publishes the child TID when clone succeeds. - const CLONE_PARENT_SETTID = 0x00100000; - const flags = origArgs[0]; - const ptidPtr = origArgs[2]; - // Read fnPtr and argPtr from the channel's CH_DATA area (written by kernel_clone stub) // These are always written as u32 by the glue (even on wasm64, table indices are i32) const processView = new DataView(channel.memory.buffer, channel.channelOffset); @@ -6347,44 +6408,96 @@ export class CentralizedKernelWorker { // Register the clear-TID pointer before starting the host Worker. A very // short-lived pthread can reach SYS_EXIT before onClone resolves. if (ctidPtr !== 0) { - this.threadCtidPtrs.set(`${channel.pid}:${tid}`, ctidPtr); - } - - this.callbacks.onClone( - channel.pid, tid, fnPtr, argPtr, stackPtr, tlsPtr, ctidPtr, channel.memory, - ).then((launchResult) => { - const assignedTid = typeof launchResult === "number" - ? launchResult - : launchResult.tid; - if (!this.processes.has(channel.pid)) { - if (ctidPtr !== 0) { - this.threadCtidPtrs.delete(`${channel.pid}:${tid}`); + this.threadCtidPtrs.set(`${channel.pid}:${tid}`, { + ptr: ctidPtr, + registration: processRegistration, + }); + } + + const failCloneAttempt = (error: unknown, errno: number) => { + console.error(`[kernel-worker] onClone failed: ${error}`); + if (!isCurrentGeneration()) return; + const ctidKey = `${channel.pid}:${tid}`; + const ctidState = this.threadCtidPtrs.get(ctidKey); + if (ctidState?.registration === processRegistration) { + this.threadCtidPtrs.delete(ctidKey); + } + this.notifyThreadExit(channel.pid, tid); + this.completeChannel(channel, SYS_CLONE, origArgs, undefined, -1, errno); + }; + + void (async () => { + let launchResult: CloneLaunchResult; + try { + launchResult = await this.callbacks.onClone!( + channel.pid, tid, fnPtr, argPtr, stackPtr, tlsPtr, ctidPtr, channel.memory, + ); + } catch (error) { + failCloneAttempt(error, 11); // EAGAIN + return; + } + + const abortBeforePublication = async (reason: unknown): Promise => { + try { + await launchResult.abort(); + } catch (abortError) { + console.error( + `[kernel-worker] clone abort failed after ${String(reason)}:`, + abortError, + ); } + }; + + if (!isCurrentGeneration()) { + await abortBeforePublication("process generation changed"); return; } - if (assignedTid !== tid && ctidPtr !== 0) { - this.threadCtidPtrs.delete(`${channel.pid}:${tid}`); - this.threadCtidPtrs.set(`${channel.pid}:${assignedTid}`, ctidPtr); + + if (launchResult.tid !== tid) { + const error = new Error( + `onClone returned tid ${launchResult.tid}, expected kernel tid ${tid}`, + ); + await abortBeforePublication(error); + failCloneAttempt(error, 11); // EAGAIN + return; } - if (flags & CLONE_PARENT_SETTID && ptidPtr !== 0) { - const procView = new DataView(channel.memory.buffer); - procView.setInt32(ptidPtr, assignedTid, true); + + if (!parentTidPointerIsValid()) { + const error = new Error("CLONE_PARENT_SETTID pointer became invalid before publication"); + await abortBeforePublication(error); + failCloneAttempt(error, 14); // EFAULT + return; } - this.completeChannel(channel, SYS_CLONE, origArgs, undefined, assignedTid, 0); - if (typeof launchResult !== "number") launchResult.start(); - }).catch((err) => { - if (ctidPtr !== 0) { - this.threadCtidPtrs.delete(`${channel.pid}:${tid}`); + + try { + this.completeChannel( + channel, + SYS_CLONE, + origArgs, + undefined, + tid, + 0, + () => { + if (flags & CLONE_PARENT_SETTID) { + new DataView(channel.memory.buffer).setInt32(ptidPtr, tid, true); + } + }, + ); + } catch (error) { + await abortBeforePublication(error); + failCloneAttempt(error, 11); // EAGAIN + return; } - console.error(`[kernel-worker] onClone failed: ${err}`); - // kernel_clone provisionally allocated ThreadInfo before the host could - // instantiate the backing Worker. Roll it back on launch failure so - // signals, /proc state, and future TID allocation reflect reality. - this.notifyThreadExit(channel.pid, tid); - if (this.processes.has(channel.pid)) { - this.completeChannel(channel, SYS_CLONE, origArgs, undefined, -1, 11); // EAGAIN + + // start() is an idempotent, non-throwing commit operation. Any failure to + // post thread_start is handled by the host as immediate thread death; + // clone success is already visible and must never be rolled back here. + try { + launchResult.start(); + } catch (error) { + console.error("[kernel-worker] CloneLaunchResult.start() threw after publication:", error); } - }); + })(); } /** @@ -6455,6 +6568,7 @@ export class CentralizedKernelWorker { // Main thread exit or exit_group: record exit status for waitpid, // queue SIGCHLD to parent, then notify the host callback. const exitingPid = channel.pid; + this.retireProcessGeneration(exitingPid, channel.memory); // Idempotency: this guard is shared with handleProcessTerminated so a // SYS_KILL that races a clean SYS_EXIT from the same process doesn't // produce two SIGCHLDs / two parent wake-ups. Cleared by @@ -6492,6 +6606,7 @@ export class CentralizedKernelWorker { */ private handleProcessTerminated(channel: ChannelInfo): void { const exitingPid = channel.pid; + this.retireProcessGeneration(exitingPid, channel.memory); // Idempotency guard — both handleExit and reapKilledProcessesAfterSyscall // can route here for the same pid; do the parent-wakeup work exactly // once per generation. Cleared by deactivateProcess + registerProcess @@ -7011,19 +7126,28 @@ export class CentralizedKernelWorker { * context, while `tid` addresses the kernel/libc thread state and clear-TID * futex word used by joiners. */ - finalizeThreadExit(pid: number, tid: number, channelOffset: number): void { + finalizeThreadExit( + pid: number, + tid: number, + channelOffset: number, + expectedMemory?: WebAssembly.Memory, + ): void { + const registration = this.processes.get(pid); + if ( + !registration || + !registration.active || + (expectedMemory && registration.memory !== expectedMemory) + ) return; const tidKey = `${pid}:${channelOffset}`; this.channelTids.delete(tidKey); this.threadForkContexts.delete(tidKey); const ctidKey = `${pid}:${tid}`; - const ctidPtr = this.threadCtidPtrs.get(ctidKey); - if (ctidPtr && ctidPtr !== 0) { + const ctidState = this.threadCtidPtrs.get(ctidKey); + if (ctidState && ctidState.registration === registration && ctidState.ptr !== 0) { this.threadCtidPtrs.delete(ctidKey); - const channel = this.activeChannels.find( - (ch) => ch.pid === pid && ch.channelOffset === channelOffset, - ); - const memory = channel?.memory ?? this.processes.get(pid)?.memory; + const ctidPtr = ctidState.ptr; + const memory = registration.memory; if (memory) { const procView = new DataView(memory.buffer); procView.setInt32(ctidPtr, 0, true); @@ -7033,7 +7157,7 @@ export class CentralizedKernelWorker { } this.notifyThreadExit(pid, tid); - this.removeChannel(pid, channelOffset); + this.removeChannel(pid, channelOffset, registration.memory); } /** diff --git a/host/src/node-kernel-worker-entry.ts b/host/src/node-kernel-worker-entry.ts index 9920b8a35a..81da67fb66 100644 --- a/host/src/node-kernel-worker-entry.ts +++ b/host/src/node-kernel-worker-entry.ts @@ -107,6 +107,7 @@ interface ForkReplayContext { } interface ProcessInfo { + active: boolean; memory: WebAssembly.Memory; programBytes: ArrayBuffer; programModule?: WebAssembly.Module; @@ -170,6 +171,7 @@ interface ThreadWorkerInfo { tid: number; basePage: number; termination?: Promise; + cancel?: (reason: string) => Promise; } const threadWorkers = new Map(); const threadExits = new ThreadExitCoordinator(); @@ -186,8 +188,8 @@ async function terminateThreadWorkers(pid: number): Promise { if (!threads) return; threadWorkers.delete(pid); for (const t of threads) { - await (t.termination ?? terminateTrackedWorker(t.worker)); - threadExits.release(pid, t.channelOffset); + await (t.cancel?.("process generation retired") ?? + t.termination ?? terminateTrackedWorker(t.worker)); } } @@ -233,6 +235,8 @@ async function finalizeProcessWorker( ): Promise { const cur = processes.get(pid); if (cur && cur.worker === worker) { + cur.active = false; + kernelWorker.retireProcessGeneration(pid, cur.memory); // Synthesize a signal-style reap *before* `deactivateProcess` in // case the worker died without sending SYS_EXIT_GROUP (uncaught // wasm trap, instantiation failure → `{type:"error"}` path). @@ -729,6 +733,7 @@ function handleSpawn(msg: SpawnMessage) { const worker = workerAdapter.createWorker(initData); processes.set(pid, { + active: true, memory, programBytes: msg.programBytes, programModule: msg.programModule, @@ -832,6 +837,7 @@ async function handleFork( const childWorker = workerAdapter.createWorker(childInitData); processes.set(childPid, { + active: true, memory: childMemory, programBytes: parentProgram, programModule: parentInfo.programModule, @@ -871,12 +877,18 @@ async function handleExec( const { programBytes, argv: launchArgv } = resolved; const newPtrWidth = detectPtrWidth(programBytes); + const oldInfo = processes.get(pid); + if (!oldInfo?.active) return -3; // ESRCH + oldInfo.active = false; const setupResult = kernelWorker.kernelExecSetup(pid); - if (setupResult < 0) return setupResult; + if (setupResult < 0) { + if (processes.get(pid) === oldInfo) oldInfo.active = true; + return setupResult; + } kernelWorker.prepareProcessForExec(pid); + await terminateThreadWorkers(pid); - const oldInfo = processes.get(pid); if (oldInfo?.worker) { intentionallyTerminated.add(oldInfo.worker as object); await oldInfo.worker.terminate().catch(() => {}); @@ -919,6 +931,7 @@ async function handleExec( const newWorker = workerAdapter.createWorker(initData); processes.set(pid, { + active: true, memory: newMemory, programBytes, worker: newWorker, @@ -1028,6 +1041,7 @@ async function handlePosixSpawn( const newWorker = workerAdapter.createWorker(initData); processes.set(childPid, { + active: true, memory, programBytes, worker: newWorker, @@ -1064,19 +1078,28 @@ async function handleClone( memory: WebAssembly.Memory, ): Promise { const processInfo = processes.get(pid); - if (!processInfo) throw new Error(`Unknown pid ${pid} for clone`); + if (!processInfo?.active || processInfo.memory !== memory) { + throw new Error(`Unknown or retired pid ${pid} for clone`); + } + const threadAllocator = processInfo.threadAllocator; // Auto-compile thread module if not already cached per-PID let threadModule = threadModuleCache.get(pid); if (!threadModule) { const patched = patchWasmForThread(processInfo.programBytes); threadModule = await WebAssembly.compile(patched); + if (!processInfo.active || processes.get(pid) !== processInfo || processInfo.memory !== memory) { + throw new Error(`Process ${pid} changed while compiling its thread module`); + } threadModuleCache.set(pid, threadModule); } + if (!processInfo.active || processes.get(pid) !== processInfo || processInfo.memory !== memory) { + throw new Error(`Process ${pid} changed before thread allocation`); + } let alloc: ReturnType; try { - alloc = processInfo.threadAllocator.allocate(memory); + alloc = threadAllocator.allocate(memory); } catch (e) { const message = e instanceof Error ? e.message : String(e); post({ @@ -1114,8 +1137,8 @@ async function handleClone( try { threadWorker = workerAdapter.createWorker(threadInitData); } catch (error) { - kernelWorker.removeChannel(pid, alloc.channelOffset); - processInfo.threadAllocator.free(alloc.basePage); + kernelWorker.removeChannel(pid, alloc.channelOffset, memory); + threadAllocator.free(alloc.basePage); throw error; } if (!threadWorkers.has(pid)) threadWorkers.set(pid, []); @@ -1128,27 +1151,27 @@ async function handleClone( threadWorkers.get(pid)!.push(threadEntry); let reclaimed = false; - const reclaimThread = () => { + function reclaimThread(): void { if (reclaimed) return; reclaimed = true; - processInfo.threadAllocator.free(alloc.basePage); - threadExits.release(pid, alloc.channelOffset); + threadAllocator.free(alloc.basePage); + threadExits.release(pid, alloc.channelOffset, terminateThreadEntry); const threads = threadWorkers.get(pid); if (threads) { const idx = threads.indexOf(threadEntry); if (idx >= 0) threads.splice(idx, 1); if (threads.length === 0) threadWorkers.delete(pid); } - }; - const terminateThreadEntry = (): Promise => { + } + function terminateThreadEntry(): Promise { if (!threadEntry.termination) { threadEntry.termination = terminateTrackedWorker(threadWorker).finally(reclaimThread); } return threadEntry.termination; - }; + } threadExits.register(pid, alloc.channelOffset, terminateThreadEntry); - let launchState: "pending" | "ready" | "failed" = "pending"; + let launchState: "pending" | "ready" | "started" | "failed" = "pending"; let finished = false; let resolveLaunch!: (result: CloneLaunchResult) => void; let rejectLaunch!: (error: Error) => void; @@ -1167,7 +1190,7 @@ async function handleClone( launchState = "failed"; finished = true; reportThreadFailure(reason); - kernelWorker.removeChannel(pid, alloc.channelOffset); + kernelWorker.removeChannel(pid, alloc.channelOffset, memory); void terminateThreadEntry().then( () => rejectLaunch(new Error(reason)), (error) => rejectLaunch(error instanceof Error ? error : new Error(String(error))), @@ -1175,12 +1198,30 @@ async function handleClone( return true; }; + const abortLaunch = async (): Promise => { + if (launchState === "failed" || finished) { + if (threadEntry.termination) await threadEntry.termination; + return; + } + launchState = "failed"; + finished = true; + kernelWorker.removeChannel(pid, alloc.channelOffset, memory); + await terminateThreadEntry(); + }; + threadEntry.cancel = async (reason: string): Promise => { + if (failLaunch(reason)) { + if (threadEntry.termination) await threadEntry.termination; + return; + } + await abortLaunch(); + }; + const failThread = (reason: string) => { if (failLaunch(reason) || finished) return; finished = true; reportThreadFailure(reason); const disposition = threadWorkerFailureDisposition(reason); - kernelWorker.finalizeThreadExit(pid, tid, alloc.channelOffset); + kernelWorker.finalizeThreadExit(pid, tid, alloc.channelOffset, memory); void terminateThreadEntry(); if (disposition.kind === "guest-fatal-trap") { try { kernelWorker.notifyHostProcessCrashed(pid, disposition.signum); } catch { /* best-effort */ } @@ -1189,22 +1230,38 @@ async function handleClone( }; threadWorker.on("message", (msg: unknown) => { const m = msg as WorkerToHostMessage; - if (m.type === "thread_ready" && m.pid === pid && m.tid === tid) { + if (m.type === "thread_ready") { + if (m.pid !== pid || m.tid !== tid) { + failLaunch( + `worker reported readiness for pid=${m.pid} tid=${m.tid}; expected pid=${pid} tid=${tid}`, + ); + return; + } if (launchState !== "pending") return; + if (!processInfo.active || processes.get(pid) !== processInfo || processInfo.memory !== memory) { + failLaunch(`process ${pid} changed before thread readiness`); + return; + } launchState = "ready"; resolveLaunch({ tid, start: () => { - if (finished) return; + if (finished || launchState !== "ready") return; + launchState = "started"; try { threadWorker.postMessage({ type: "thread_start", pid, tid }); } catch (error) { failThread(`unable to start initialized worker: ${error}`); } }, + abort: abortLaunch, }); } else if (m.type === "thread_exit") { if (failLaunch("worker exited before reporting thread readiness")) return; + if (launchState === "ready") { + failThread("worker exited before thread start"); + return; + } finished = true; void terminateThreadEntry(); } else if (m.type === "error") { @@ -1236,6 +1293,8 @@ async function finishProcessExit(pid: number, exitStatus: number): Promise } const info = processes.get(pid); + if (info) info.active = false; + kernelWorker.retireProcessGeneration(pid, info?.memory); const teardown = (async () => { // Keep the pid registered until the process worker is gone. musl's @@ -1273,23 +1332,13 @@ async function finishProcessExit(pid: number, exitStatus: number): Promise async function handleTerminate(msg: TerminateProcessMessage) { const pid = msg.pid; + const info = processes.get(pid); + if (info) info.active = false; + kernelWorker.retireProcessGeneration(pid, info?.memory); - // Terminate thread workers - const threads = threadWorkers.get(pid); - if (threads) { - for (const t of threads) { - intentionallyTerminated.add(t.worker as object); - await t.worker.terminate().catch(() => {}); - try { - kernelWorker.notifyThreadExit(pid, t.tid); - kernelWorker.removeChannel(pid, t.channelOffset); - } catch {} - } - threadWorkers.delete(pid); - } + await terminateThreadWorkers(pid); // Terminate main process worker - const info = processes.get(pid); if (info?.worker) { await terminateTrackedWorker(info.worker); } diff --git a/host/src/thread-exit-coordinator.ts b/host/src/thread-exit-coordinator.ts index c113f5d2d4..af971cd586 100644 --- a/host/src/thread-exit-coordinator.ts +++ b/host/src/thread-exit-coordinator.ts @@ -20,8 +20,18 @@ export class ThreadExitCoordinator { } } - release(pid: number, channelOffset: number): void { + release( + pid: number, + channelOffset: number, + expectedTerminator?: ThreadTerminator, + ): void { const key = this.key(pid, channelOffset); + if ( + expectedTerminator !== undefined && + this.terminators.get(key) !== expectedTerminator + ) { + return; + } this.terminators.delete(key); this.pendingExits.delete(key); } diff --git a/host/test/centralized-test-helper.ts b/host/test/centralized-test-helper.ts index ae886efb32..1217227162 100644 --- a/host/test/centralized-test-helper.ts +++ b/host/test/centralized-test-helper.ts @@ -8,7 +8,11 @@ import { readFileSync, existsSync } from "node:fs"; import { join, dirname } from "node:path"; import { fileURLToPath } from "node:url"; -import { CAPTURED_STDIO, CentralizedKernelWorker } from "../src/kernel-worker"; +import { + CAPTURED_STDIO, + CentralizedKernelWorker, + type CloneLaunchResult, +} from "../src/kernel-worker"; import { resolveBinary } from "../src/binary-resolver"; import { NodePlatformIO } from "../src/platform/node"; import { NodeWorkerAdapter } from "../src/worker-adapter"; @@ -292,6 +296,28 @@ async function runOnMainThread(options: RunProgramOptions): Promise>(); + const threadTeardowns = new Map Promise>>(); + + const registerThreadTeardown = (threadPid: number, teardown: () => Promise) => { + let teardowns = threadTeardowns.get(threadPid); + if (!teardowns) { + teardowns = new Set(); + threadTeardowns.set(threadPid, teardowns); + } + teardowns.add(teardown); + }; + + const unregisterThreadTeardown = (threadPid: number, teardown: () => Promise) => { + const teardowns = threadTeardowns.get(threadPid); + if (!teardowns) return; + teardowns.delete(teardown); + if (teardowns.size === 0) threadTeardowns.delete(threadPid); + }; + + const terminateThreadsForProcess = async (threadPid: number): Promise => { + const teardowns = [...(threadTeardowns.get(threadPid) ?? [])]; + await Promise.all(teardowns.map((teardown) => teardown())); + }; const io = options.io ?? new NodePlatformIO(); const workerAdapter = new NodeWorkerAdapter(); @@ -400,6 +426,7 @@ async function runOnMainThread(options: RunProgramOptions): Promise; + try { + threadWorker = workerAdapter.createWorker(threadInitData); + } catch (error) { + kernelWorker.removeChannel(clonePid, alloc.channelOffset, memory); + threadAllocator.free(alloc.basePage); + throw error; + } + + let state: "pending" | "ready" | "started" | "finished" | "failed" = "pending"; + let termination: Promise | undefined; + const terminateThread = (): Promise => { + if (!termination) { + termination = threadWorker.terminate().catch(() => {}).finally(() => { + kernelWorker.removeChannel(clonePid, alloc.channelOffset, memory); + threadAllocator.free(alloc.basePage); + unregisterThreadTeardown(clonePid, terminateThread); + }); + } + return termination; + }; + registerThreadTeardown(clonePid, terminateThread); + + let resolveLaunch!: (result: CloneLaunchResult) => void; + let rejectLaunch!: (error: Error) => void; + const launch = new Promise((resolve, reject) => { + resolveLaunch = resolve; + rejectLaunch = reject; + }); + + const failPendingLaunch = (reason: string): boolean => { + if (state !== "pending") return false; + state = "failed"; + void terminateThread().then( + () => rejectLaunch(new Error(reason)), + (error) => rejectLaunch(error instanceof Error ? error : new Error(String(error))), + ); + return true; + }; + + const abortLaunch = async (): Promise => { + if (state !== "finished") state = "failed"; + await terminateThread(); + }; + threadWorker.on("message", (msg: unknown) => { const m = msg as WorkerToHostMessage; - if (m.type === "thread_exit") { - threadAllocator.free(alloc.basePage); - threadWorker.terminate().catch(() => {}); + if (m.type === "thread_ready") { + if (m.pid !== clonePid || m.tid !== tid) { + failPendingLaunch( + `Worker reported readiness for pid=${m.pid} tid=${m.tid}; ` + + `expected pid=${clonePid} tid=${tid}`, + ); + return; + } + if (state !== "pending") return; + if ( + threadAllocators.get(clonePid) !== threadAllocator || + processProgramBytes.get(clonePid) !== cloneProgramBytes + ) { + failPendingLaunch(`Process ${clonePid} changed before thread readiness`); + return; + } + state = "ready"; + resolveLaunch({ + tid, + start: () => { + if (state !== "ready") return; + state = "started"; + try { + threadWorker.postMessage({ type: "thread_start", pid: clonePid, tid }); + } catch (error) { + state = "failed"; + kernelWorker.finalizeThreadExit(clonePid, tid, alloc.channelOffset, memory); + void terminateThread(); + } + }, + abort: abortLaunch, + }); + } else if (m.type === "thread_exit") { + if (failPendingLaunch("Worker exited before reporting thread readiness")) return; + state = "finished"; + void terminateThread(); + } else if (m.type === "error") { + if (failPendingLaunch(m.message)) return; + if (state === "finished" || state === "failed") return; + state = "failed"; + kernelWorker.finalizeThreadExit(clonePid, tid, alloc.channelOffset, memory); + void terminateThread(); } }); - threadWorker.on("error", () => { - kernelWorker.notifyThreadExit(clonePid, tid); - kernelWorker.removeChannel(clonePid, alloc.channelOffset); - threadAllocator.free(alloc.basePage); + threadWorker.on("error", (error: Error) => { + if (failPendingLaunch(`Thread worker failed: ${error.message}`)) return; + if (state === "finished" || state === "failed") return; + state = "failed"; + kernelWorker.finalizeThreadExit(clonePid, tid, alloc.channelOffset, memory); + void terminateThread(); }); - return tid; + return launch; }, onExit: (exitPid, exitStatus) => { if (exitPid === pid) { + void terminateThreadsForProcess(exitPid); kernelWorker.unregisterProcess(exitPid); processProgramBytes.delete(exitPid); processLayouts.delete(exitPid); @@ -508,6 +623,7 @@ async function runOnMainThread(options: RunProgramOptions): Promise { for (const [, w] of workers) w.terminate().catch(() => {}); + for (const threadPid of threadTeardowns.keys()) { + void terminateThreadsForProcess(threadPid); + } rejectExit(new Error(`Program timed out after ${timeout}ms`)); }, timeout); @@ -607,12 +726,18 @@ async function runOnMainThread(options: RunProgramOptions): Promise {}); + for (const threadPid of threadTeardowns.keys()) { + void terminateThreadsForProcess(threadPid); + } rejectExit(new Error(m.message)); } }); const exitCode = await exitPromise; clearTimeout(timer); + for (const threadPid of [...threadTeardowns.keys()]) { + await terminateThreadsForProcess(threadPid); + } const totalLen = stdoutChunks.reduce((sum, c) => sum + c.length, 0); const stdoutBytes = new Uint8Array(totalLen); diff --git a/host/test/fork-instrument-coverage.test.ts b/host/test/fork-instrument-coverage.test.ts index 8383dc39f5..089de05010 100644 --- a/host/test/fork-instrument-coverage.test.ts +++ b/host/test/fork-instrument-coverage.test.ts @@ -34,6 +34,7 @@ import { describe, it, expect } from "vitest"; import { runCentralizedProgram } from "./centralized-test-helper"; import { resolveBinary, tryResolveBinary } from "../src/binary-resolver"; +import { NodePlatformIO } from "../src/platform/node"; // --------------------------------------------------------------------------- // Helpers @@ -480,6 +481,20 @@ describe("fork_instrument_coverage / P-* process & threading", () => { timeout: 10_000, }); }); + + it("P-10 works through the custom PlatformIO clone callback", async () => { + const binary = tryResolveBinary("programs/p_10_fork_child_creates_thread.wasm"); + expect(binary, "missing P-10 fixture").toBeTruthy(); + const result = await runCentralizedProgram({ + programPath: binary!, + io: new NodePlatformIO(), + timeout: 10_000, + }); + + expect(result.exitCode).toBe(0); + expect(result.stdout).toContain("CHILD_THREAD: ok"); + expect(result.stdout).toContain("PASS: P-10"); + }); }); // --------------------------------------------------------------------------- diff --git a/host/test/multi-worker.test.ts b/host/test/multi-worker.test.ts index be99be99b3..56ed6ec621 100644 --- a/host/test/multi-worker.test.ts +++ b/host/test/multi-worker.test.ts @@ -4,7 +4,11 @@ // setNextChildPid, and fork flow. import { describe, it, expect, vi } from "vitest"; import { readFileSync } from "node:fs"; -import { CAPTURED_STDIO, CentralizedKernelWorker } from "../src/kernel-worker"; +import { + CAPTURED_STDIO, + CentralizedKernelWorker, + type CloneLaunchResult, +} from "../src/kernel-worker"; import { resolveBinary } from "../src/binary-resolver"; import { NodePlatformIO } from "../src/platform/node"; import { SharedLockTable } from "../src/shared-lock-table"; @@ -71,6 +75,7 @@ describe("CentralizedKernelWorker Process Management", () => { const kw = Object.assign(Object.create(CentralizedKernelWorker.prototype), { activeChannels: [{ pid }, { pid: peerPid }], processes: new Map([[pid, {}], [peerPid, {}]]), + threadCtidPtrs: new Map(), stdinFinite: new Set([pid]), stdinBuffers: new Map([[pid, new Uint8Array()]]), alarmTimers: new Map(), @@ -208,17 +213,23 @@ describe("CentralizedKernelWorker Process Management", () => { consecutiveSyscalls: 0, }; new DataView(memory.buffer).setInt32(ctidPtr, tid, true); + const registration = { + active: true, + memory, + channels: [{ channelOffset: mainChannelOffset }, channel], + }; const kw = Object.assign(Object.create(CentralizedKernelWorker.prototype), { - processes: new Map([ - [pid, { memory, channels: [{ channelOffset: mainChannelOffset }, channel] }], - ]), + processes: new Map([[pid, registration]]), activeChannels: [channel], channelTids: new Map([[`${pid}:${threadChannelOffset}`, tid]]), threadForkContexts: new Map([ [`${pid}:${threadChannelOffset}`, { fnPtr: 1, argPtr: 2 }], ]), - threadCtidPtrs: new Map([[`${pid}:${tid}`, ctidPtr]]), + threadCtidPtrs: new Map([[ + `${pid}:${tid}`, + { ptr: ctidPtr, registration }, + ]]), notifyThreadExit: vi.fn(), }) as CentralizedKernelWorker; @@ -253,11 +264,13 @@ describe("CentralizedKernelWorker Process Management", () => { maximum: 1, }); const kernelView = new DataView(kernelMemory.buffer); - const threadCtidPtrs = new Map(); - let resolveClone!: (value: number) => void; + const threadCtidPtrs = new Map(); + const channel = { pid, channelOffset: mainChannelOffset, memory }; + const registration = { active: true, channels: [channel], memory }; + let resolveClone!: (value: CloneLaunchResult) => void; const onClone = vi.fn(() => { - expect(threadCtidPtrs.get(`${pid}:${tid}`)).toBe(ctidPtr); - return new Promise((resolve) => { + expect(threadCtidPtrs.get(`${pid}:${tid}`)?.ptr).toBe(ctidPtr); + return new Promise((resolve) => { resolveClone = resolve; }); }); @@ -272,9 +285,7 @@ describe("CentralizedKernelWorker Process Management", () => { kernelMemory, scratchOffset: 0, currentHandlePid: 0, - processes: new Map([ - [pid, { channels: [{ channelOffset: mainChannelOffset }] }], - ]), + processes: new Map([[pid, registration]]), threadCtidPtrs, completeChannel: vi.fn(), bindKernelTidForChannel: vi.fn(), @@ -290,13 +301,13 @@ describe("CentralizedKernelWorker Process Management", () => { }); (kw as any).handleClone( - { pid, channelOffset: mainChannelOffset, memory }, + channel, [0, stackPtr, 0, tlsPtr, ctidPtr, 0], ); expect(onClone).toHaveBeenCalledTimes(1); - expect(threadCtidPtrs.get(`${pid}:${tid}`)).toBe(ctidPtr); - resolveClone(tid); + expect(threadCtidPtrs.get(`${pid}:${tid}`)?.ptr).toBe(ctidPtr); + resolveClone({ tid, start: vi.fn(), abort: vi.fn(async () => {}) }); await Promise.resolve(); expect((kw as any).completeChannel).toHaveBeenCalled(); }); @@ -314,12 +325,21 @@ describe("CentralizedKernelWorker Process Management", () => { const kernelMemory = new WebAssembly.Memory({ initial: 1, maximum: 1 }); const kernelView = new DataView(kernelMemory.buffer); - const completeChannel = vi.fn(); + const completeChannel = vi.fn((...args: unknown[]) => { + const beforePublish = args[6] as (() => void) | undefined; + beforePublish?.(); + }); + const channel = { pid, channelOffset: mainChannelOffset, memory }; + const registration = { active: true, channels: [channel], memory }; const start = vi.fn(() => { expect(completeChannel).toHaveBeenCalledTimes(1); expect(new DataView(memory.buffer).getInt32(ptidPtr, true)).toBe(tid); }); - const onClone = vi.fn(async () => ({ tid, start })); + const onClone = vi.fn(async () => ({ + tid, + start, + abort: vi.fn(async () => {}), + })); const kw = Object.assign(Object.create(CentralizedKernelWorker.prototype), { callbacks: { onClone }, @@ -327,7 +347,7 @@ describe("CentralizedKernelWorker Process Management", () => { kernelMemory, scratchOffset: 0, currentHandlePid: 0, - processes: new Map([[pid, { channels: [{ channelOffset: mainChannelOffset }] }]]), + processes: new Map([[pid, registration]]), threadCtidPtrs: new Map(), completeChannel, bindKernelTidForChannel: vi.fn(), @@ -344,7 +364,7 @@ describe("CentralizedKernelWorker Process Management", () => { const CLONE_PARENT_SETTID = 0x00100000; (kw as any).handleClone( - { pid, channelOffset: mainChannelOffset, memory }, + channel, [CLONE_PARENT_SETTID, 0x00800000, ptidPtr, 0x00900000, 0, 0], ); @@ -353,6 +373,251 @@ describe("CentralizedKernelWorker Process Management", () => { expect(start).toHaveBeenCalledTimes(1); }); + it.each([0, 16 * WASM_PAGE_SIZE - 3])( + "rejects invalid CLONE_PARENT_SETTID pointer %d before allocating a TID", + (ptidPtr) => { + const pid = 130; + const mainChannelOffset = WASM_PAGE_SIZE; + const memory = new WebAssembly.Memory({ initial: 16, maximum: 16, shared: true }); + const channel = { pid, channelOffset: mainChannelOffset, memory }; + const registration = { active: true, channels: [channel], memory }; + const processView = new DataView(memory.buffer, mainChannelOffset); + processView.setUint32(CH_DATA, 11, true); + processView.setUint32(CH_DATA + 4, 22, true); + + const kernelMemory = new WebAssembly.Memory({ initial: 1, maximum: 1 }); + const kernelHandle = vi.fn(); + const onClone = vi.fn(); + const completeChannel = vi.fn(); + const kw = Object.assign(Object.create(CentralizedKernelWorker.prototype), { + callbacks: { onClone }, + kernel: { toKernelPtr: (value: number | bigint) => Number(value) }, + kernelMemory, + scratchOffset: 0, + currentHandlePid: 0, + processes: new Map([[pid, registration]]), + threadCtidPtrs: new Map(), + completeChannel, + bindKernelTidForChannel: vi.fn(), + kernelInstance: { exports: { kernel_handle_channel: kernelHandle } }, + }); + + const CLONE_PARENT_SETTID = 0x00100000; + const args = [CLONE_PARENT_SETTID, 0x00800000, ptidPtr, 0x00900000, 0, 0]; + (kw as any).handleClone(channel, args); + + expect(kernelHandle).not.toHaveBeenCalled(); + expect(onClone).not.toHaveBeenCalled(); + expect(completeChannel).toHaveBeenCalledOnce(); + expect(completeChannel).toHaveBeenCalledWith( + channel, + ABI_SYSCALLS.Clone, + args, + undefined, + -1, + 14, + ); + }, + ); + + it("accepts an in-bounds unaligned CLONE_PARENT_SETTID pointer", async () => { + const pid = 131; + const tid = 82; + const mainChannelOffset = WASM_PAGE_SIZE; + const ptidPtr = 0x00030001; + const memory = new WebAssembly.Memory({ initial: 16, maximum: 16, shared: true }); + const channel = { pid, channelOffset: mainChannelOffset, memory }; + const registration = { active: true, channels: [channel], memory }; + const processView = new DataView(memory.buffer, mainChannelOffset); + processView.setUint32(CH_DATA, 11, true); + processView.setUint32(CH_DATA + 4, 22, true); + + const kernelMemory = new WebAssembly.Memory({ initial: 1, maximum: 1 }); + const kernelView = new DataView(kernelMemory.buffer); + const start = vi.fn(); + const abort = vi.fn(async () => {}); + const completeChannel = vi.fn((...args: unknown[]) => { + const beforePublish = args[6] as (() => void) | undefined; + beforePublish?.(); + }); + const kw = Object.assign(Object.create(CentralizedKernelWorker.prototype), { + callbacks: { onClone: vi.fn(async () => ({ tid, start, abort })) }, + kernel: { toKernelPtr: (value: number | bigint) => Number(value) }, + kernelMemory, + scratchOffset: 0, + currentHandlePid: 0, + processes: new Map([[pid, registration]]), + threadCtidPtrs: new Map(), + completeChannel, + bindKernelTidForChannel: vi.fn(), + kernelInstance: { + exports: { + kernel_handle_channel: vi.fn(() => { + kernelView.setBigInt64(CH_RETURN, BigInt(tid), true); + kernelView.setUint32(CH_ERRNO, 0, true); + return 0; + }), + }, + }, + }); + + const CLONE_PARENT_SETTID = 0x00100000; + (kw as any).handleClone( + channel, + [CLONE_PARENT_SETTID, 0x00800000, ptidPtr, 0x00900000, 0, 0], + ); + await Promise.resolve(); + await Promise.resolve(); + + expect(new DataView(memory.buffer).getInt32(ptidPtr, true)).toBe(tid); + expect(start).toHaveBeenCalledOnce(); + expect(abort).not.toHaveBeenCalled(); + }); + + it("aborts a ready Worker when clone publication fails", async () => { + const pid = 132; + const tid = 83; + const mainChannelOffset = WASM_PAGE_SIZE; + const ptidPtr = 0x00030000; + const ctidPtr = 0x00040000; + const memory = new WebAssembly.Memory({ initial: 16, maximum: 16, shared: true }); + const channel = { pid, channelOffset: mainChannelOffset, memory }; + const registration = { active: true, channels: [channel], memory }; + const processView = new DataView(memory.buffer, mainChannelOffset); + processView.setUint32(CH_DATA, 11, true); + processView.setUint32(CH_DATA + 4, 22, true); + new DataView(memory.buffer).setInt32(ptidPtr, -1, true); + + const kernelMemory = new WebAssembly.Memory({ initial: 1, maximum: 1 }); + const kernelView = new DataView(kernelMemory.buffer); + const start = vi.fn(); + const abort = vi.fn(async () => {}); + const notifyThreadExit = vi.fn(); + const publishedFailures: Array<[number, number]> = []; + const completeChannel = vi.fn((...args: unknown[]) => { + const retVal = args[4] as number; + const errVal = args[5] as number; + if (retVal === tid) throw new Error("pre-publication output drain failed"); + publishedFailures.push([retVal, errVal]); + }); + const threadCtidPtrs = new Map(); + const kw = Object.assign(Object.create(CentralizedKernelWorker.prototype), { + callbacks: { onClone: vi.fn(async () => ({ tid, start, abort })) }, + kernel: { toKernelPtr: (value: number | bigint) => Number(value) }, + kernelMemory, + scratchOffset: 0, + currentHandlePid: 0, + processes: new Map([[pid, registration]]), + threadCtidPtrs, + completeChannel, + notifyThreadExit, + bindKernelTidForChannel: vi.fn(), + kernelInstance: { + exports: { + kernel_handle_channel: vi.fn(() => { + kernelView.setBigInt64(CH_RETURN, BigInt(tid), true); + kernelView.setUint32(CH_ERRNO, 0, true); + return 0; + }), + }, + }, + }); + + const CLONE_PARENT_SETTID = 0x00100000; + (kw as any).handleClone( + channel, + [CLONE_PARENT_SETTID, 0x00800000, ptidPtr, 0x00900000, ctidPtr, 0], + ); + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + + expect(abort).toHaveBeenCalledOnce(); + expect(start).not.toHaveBeenCalled(); + expect(notifyThreadExit).toHaveBeenCalledWith(pid, tid); + expect(threadCtidPtrs.has(`${pid}:${tid}`)).toBe(false); + expect(new DataView(memory.buffer).getInt32(ptidPtr, true)).toBe(-1); + expect(publishedFailures).toEqual([[-1, 11]]); + }); + + it("does not let stale clone cleanup touch a replacement process generation", async () => { + const pid = 133; + const tid = 84; + const mainChannelOffset = WASM_PAGE_SIZE; + const ctidPtr = 0x00040000; + const memory = new WebAssembly.Memory({ initial: 16, maximum: 16, shared: true }); + const channel = { pid, channelOffset: mainChannelOffset, memory }; + const oldRegistration = { active: true, channels: [channel], memory }; + const processView = new DataView(memory.buffer, mainChannelOffset); + processView.setUint32(CH_DATA, 11, true); + processView.setUint32(CH_DATA + 4, 22, true); + + const kernelMemory = new WebAssembly.Memory({ initial: 1, maximum: 1 }); + const kernelView = new DataView(kernelMemory.buffer); + let resolveLaunch!: (result: CloneLaunchResult) => void; + const onClone = vi.fn(() => new Promise((resolve) => { + resolveLaunch = resolve; + })); + const abort = vi.fn(async () => {}); + const completeChannel = vi.fn(); + const notifyThreadExit = vi.fn(); + const threadCtidPtrs = new Map(); + const processes = new Map([[pid, oldRegistration]]); + const kw = Object.assign(Object.create(CentralizedKernelWorker.prototype), { + callbacks: { onClone }, + kernel: { toKernelPtr: (value: number | bigint) => Number(value) }, + kernelMemory, + scratchOffset: 0, + currentHandlePid: 0, + processes, + threadCtidPtrs, + completeChannel, + notifyThreadExit, + bindKernelTidForChannel: vi.fn(), + kernelInstance: { + exports: { + kernel_handle_channel: vi.fn(() => { + kernelView.setBigInt64(CH_RETURN, BigInt(tid), true); + kernelView.setUint32(CH_ERRNO, 0, true); + return 0; + }), + }, + }, + }); + + (kw as any).handleClone( + channel, + [0, 0x00800000, 0, 0x00900000, ctidPtr, 0], + ); + expect(threadCtidPtrs.get(`${pid}:${tid}`)?.registration).toBe(oldRegistration); + + oldRegistration.active = false; + const replacementMemory = new WebAssembly.Memory({ + initial: 16, + maximum: 16, + shared: true, + }); + const replacementRegistration = { + active: true, + channels: [], + memory: replacementMemory, + }; + processes.set(pid, replacementRegistration); + const replacementCtidState = { ptr: ctidPtr, registration: replacementRegistration }; + threadCtidPtrs.set(`${pid}:${tid}`, replacementCtidState); + + resolveLaunch({ tid, start: vi.fn(), abort }); + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + + expect(abort).toHaveBeenCalledOnce(); + expect(notifyThreadExit).not.toHaveBeenCalled(); + expect(completeChannel).not.toHaveBeenCalled(); + expect(threadCtidPtrs.get(`${pid}:${tid}`)).toBe(replacementCtidState); + }); + it("rolls back the provisional kernel thread when its Worker cannot launch", async () => { const pid = 128; const mainChannelOffset = WASM_PAGE_SIZE; @@ -369,10 +634,12 @@ describe("CentralizedKernelWorker Process Management", () => { const kernelView = new DataView(kernelMemory.buffer); const completeChannel = vi.fn(); const notifyThreadExit = vi.fn(); + const channel = { pid, channelOffset: mainChannelOffset, memory }; + const registration = { active: true, channels: [channel], memory }; const onClone = vi.fn(async () => { throw new WebAssembly.CompileError("invalid thread module"); }); - const threadCtidPtrs = new Map(); + const threadCtidPtrs = new Map(); const kw = Object.assign(Object.create(CentralizedKernelWorker.prototype), { callbacks: { onClone }, @@ -380,7 +647,7 @@ describe("CentralizedKernelWorker Process Management", () => { kernelMemory, scratchOffset: 0, currentHandlePid: 0, - processes: new Map([[pid, { channels: [{ channelOffset: mainChannelOffset }] }]]), + processes: new Map([[pid, registration]]), threadCtidPtrs, completeChannel, notifyThreadExit, @@ -398,7 +665,7 @@ describe("CentralizedKernelWorker Process Management", () => { const CLONE_PARENT_SETTID = 0x00100000; const args = [CLONE_PARENT_SETTID, 0x00800000, ptidPtr, 0x00900000, ctidPtr, 0]; - (kw as any).handleClone({ pid, channelOffset: mainChannelOffset, memory }, args); + (kw as any).handleClone(channel, args); await Promise.resolve(); await Promise.resolve(); diff --git a/host/test/thread-exit-coordinator.test.ts b/host/test/thread-exit-coordinator.test.ts index 9390d90edf..23035d253d 100644 --- a/host/test/thread-exit-coordinator.test.ts +++ b/host/test/thread-exit-coordinator.test.ts @@ -41,4 +41,18 @@ describe("ThreadExitCoordinator", () => { expect(firstTerminate).toHaveBeenCalledTimes(1); expect(secondTerminate).not.toHaveBeenCalled(); }); + + it("does not let stale cleanup release a reused channel's terminator", () => { + const exits = new ThreadExitCoordinator(); + const firstTerminate = vi.fn(async () => {}); + const secondTerminate = vi.fn(async () => {}); + + exits.register(123, 0x20000, firstTerminate); + exits.register(123, 0x20000, secondTerminate); + exits.release(123, 0x20000, firstTerminate); + + expect(exits.requestExit(123, 0x20000)).toBe(true); + expect(firstTerminate).not.toHaveBeenCalled(); + expect(secondTerminate).toHaveBeenCalledOnce(); + }); }); diff --git a/packages/registry/mariadb/test/run-tests.ts b/packages/registry/mariadb/test/run-tests.ts index a42b7a6d73..fdf429e33c 100644 --- a/packages/registry/mariadb/test/run-tests.ts +++ b/packages/registry/mariadb/test/run-tests.ts @@ -16,7 +16,11 @@ import { readFileSync, existsSync, mkdirSync, readdirSync, writeFileSync } from "fs"; import { resolve, dirname } from "path"; import { createConnection, createServer, type Socket } from "net"; -import { CAPTURED_STDIO, CentralizedKernelWorker } from "../../../../host/src/kernel-worker"; +import { + CAPTURED_STDIO, + CentralizedKernelWorker, + type CloneLaunchResult, +} from "../../../../host/src/kernel-worker"; import { NodePlatformIO } from "../../../../host/src/platform/node"; import { NodeWorkerAdapter } from "../../../../host/src/worker-adapter"; import { patchWasmForThread } from "../../../../host/src/worker-main"; @@ -120,7 +124,10 @@ function nextPid(): number { return _nextPid++; } // Server mid-test restart state let autoRestartOnServerExit = false; let serverRestartPromise: Promise | null = null; -const serverThreadWorkers = new Set>(); +const serverThreadTeardowns = new Map< + ReturnType, + (reason: string) => Promise +>(); // Track current running test for thread crash abort let currentTestReject: ((err: Error) => void) | null = null; @@ -236,7 +243,8 @@ async function main() { }, onClone: async (pid, tid, fnPtr, argPtr, stackPtr, tlsPtr, ctidPtr, memory) => { - const alloc = threadAllocator.allocate(memory); + const allocator = threadAllocator; + const alloc = allocator.allocate(memory); kernelWorker.addChannel(pid, alloc.channelOffset, tid); const threadInitData: CentralizedThreadInitMessage = { @@ -247,25 +255,101 @@ async function main() { memory, channelOffset: alloc.channelOffset, fnPtr, argPtr, stackPtr, tlsPtr, ctidPtr, + tlsOffset: alloc.tlsOffset, tlsAllocAddr: alloc.tlsAllocAddr, }; - const threadWorker = workerAdapter.createWorker(threadInitData); - serverThreadWorkers.add(threadWorker); + let threadWorker: ReturnType; + try { + threadWorker = workerAdapter.createWorker(threadInitData); + } catch (error) { + kernelWorker.removeChannel(pid, alloc.channelOffset, memory); + allocator.free(alloc.basePage); + throw error; + } + let state: "pending" | "ready" | "started" | "finished" | "failed" = "pending"; + let termination: Promise | undefined; + const terminateThread = (): Promise => { + if (!termination) { + termination = threadWorker.terminate().catch(() => {}).finally(() => { + serverThreadTeardowns.delete(threadWorker); + allocator.free(alloc.basePage); + }); + } + return termination; + }; + let resolveLaunch!: (result: CloneLaunchResult) => void; + let rejectLaunch!: (error: Error) => void; + const launch = new Promise((resolveLaunchPromise, rejectLaunchPromise) => { + resolveLaunch = resolveLaunchPromise; + rejectLaunch = rejectLaunchPromise; + }); + const failPendingLaunch = (reason: string): boolean => { + if (state !== "pending") return false; + state = "failed"; + kernelWorker.removeChannel(pid, alloc.channelOffset, memory); + void terminateThread().then( + () => rejectLaunch(new Error(reason)), + (error) => rejectLaunch(error instanceof Error ? error : new Error(String(error))), + ); + return true; + }; + const abortLaunch = async (): Promise => { + state = "failed"; + kernelWorker.removeChannel(pid, alloc.channelOffset, memory); + await terminateThread(); + }; + const cancelThread = async (reason: string): Promise => { + if (failPendingLaunch(reason)) { + if (termination) await termination; + return; + } + await abortLaunch(); + }; + serverThreadTeardowns.set(threadWorker, cancelThread); threadWorker.on("message", (msg: unknown) => { const m = msg as WorkerToHostMessage; - if (m.type === "thread_exit") { - serverThreadWorkers.delete(threadWorker); - kernelWorker.notifyThreadExit(pid, tid); - kernelWorker.removeChannel(pid, alloc.channelOffset); - threadAllocator.free(alloc.basePage); - threadWorker.terminate().catch(() => {}); + if (m.type === "thread_ready") { + if (m.pid !== pid || m.tid !== tid) { + failPendingLaunch( + `Worker reported readiness for pid=${m.pid} tid=${m.tid}; expected pid=${pid} tid=${tid}`, + ); + return; + } + if (state !== "pending") return; + state = "ready"; + resolveLaunch({ + tid, + start: () => { + if (state !== "ready") return; + state = "started"; + try { + threadWorker.postMessage({ type: "thread_start", pid, tid }); + } catch (error) { + state = "failed"; + kernelWorker.finalizeThreadExit(pid, tid, alloc.channelOffset, memory); + void terminateThread(); + } + }, + abort: abortLaunch, + }); + } else if (m.type === "thread_exit") { + if (failPendingLaunch("Worker exited before reporting thread readiness")) return; + state = "finished"; + void terminateThread(); + } else if (m.type === "error") { + if (failPendingLaunch(m.message)) return; + if (state === "finished" || state === "failed") return; + state = "failed"; + kernelWorker.finalizeThreadExit(pid, tid, alloc.channelOffset, memory); + void terminateThread(); } }); threadWorker.on("error", (err) => { - serverThreadWorkers.delete(threadWorker); - try { kernelWorker.notifyThreadExit(pid, tid); } catch {} - try { kernelWorker.removeChannel(pid, alloc.channelOffset); } catch {} - threadAllocator.free(alloc.basePage); + if (failPendingLaunch(`Server thread failed: ${err?.message ?? err}`)) return; + if (state === "finished" || state === "failed") return; + state = "failed"; + kernelWorker.finalizeThreadExit(pid, tid, alloc.channelOffset, memory); + void terminateThread(); // Server thread crashed — mark for restart and abort current test const msg = err?.message || "server thread crashed"; console.error(`[thread-worker] tid=${tid} CAUGHT ERROR: ${msg}`); @@ -277,7 +361,7 @@ async function main() { currentTestWorker = null; } }); - return tid; + return launch; }, onExec: async () => -38, // ENOSYS @@ -337,10 +421,10 @@ async function main() { port: number, ): Promise { // Terminate remaining server thread workers - for (const tw of serverThreadWorkers) { - await tw.terminate().catch(() => {}); + for (const teardown of [...serverThreadTeardowns.values()]) { + await teardown("server generation retired"); } - serverThreadWorkers.clear(); + serverThreadTeardowns.clear(); threadAllocator = new ThreadPageAllocator(MAX_PAGES); // Clean Aria control/log files @@ -385,10 +469,10 @@ async function main() { /** Kill all workers and start a fresh server instance. */ async function restartServer(): Promise { // Terminate all workers (including server threads) - for (const tw of serverThreadWorkers) { - await tw.terminate().catch(() => {}); + for (const teardown of [...serverThreadTeardowns.values()]) { + await teardown("server generation retired"); } - serverThreadWorkers.clear(); + serverThreadTeardowns.clear(); for (const [pid, w] of workers) { await w.terminate().catch(() => {}); try { kernelWorker.unregisterProcess(pid); } catch {} From f678e098e10b2c563ae0eecd3e3088e862695086 Mon Sep 17 00:00:00 2001 From: Brandon Payton Date: Sat, 11 Jul 2026 19:48:27 -0400 Subject: [PATCH 3/3] host: ignore debug names when patching thread modules --- host/src/worker-main.ts | 33 ++------------------------- host/test/thread-wasm-patch.test.ts | 35 +++++++++++++++++++++++++++++ 2 files changed, 37 insertions(+), 31 deletions(-) diff --git a/host/src/worker-main.ts b/host/src/worker-main.ts index 98c9f5c25e..1acd5dfaec 100644 --- a/host/src/worker-main.ts +++ b/host/src/worker-main.ts @@ -1537,39 +1537,10 @@ export function patchWasmForThread(bytes: ArrayBuffer): ArrayBuffer { ctorCandidates.set(index, sources); }; + // Custom name sections are optional debug metadata and cannot authorize a + // function-body rewrite. Use only executable linker semantics as evidence. addCtorCandidate(exportFuncIndicesByName.get("__wasm_call_ctors"), "function export"); - // Unstripped modules can retain the exact synthetic function name even when - // it is not exported. Fork instrumentation appends to this same name map. - for (const sec of sections) { - if (sec.id !== 0) continue; - let pos = sec.contentOffset; - let customName: string; - [customName, pos] = readName(pos); - if (customName !== "name") continue; - const sectionEnd = sec.contentOffset + sec.contentSize; - while (pos < sectionEnd) { - const subsectionId = src[pos++]; - const [subsectionSize, sizeBytes] = readLEB128(src, pos); - pos += sizeBytes; - const subsectionEnd = pos + subsectionSize; - if (subsectionId === 1) { - const [nameCount, countBytes] = readLEB128(src, pos); - pos += countBytes; - for (let i = 0; i < nameCount; i++) { - const [funcIndex, indexBytes] = readLEB128(src, pos); - pos += indexBytes; - let functionName: string; - [functionName, pos] = readName(pos); - if (functionName === "__wasm_call_ctors") { - addCtorCandidate(funcIndex, "name section"); - } - } - } - pos = subsectionEnd; - } - } - const abiMarkerIndex = exportFuncIndicesByName.get("__abi_version"); if (abiMarkerIndex !== undefined && extractAbiVersion(bytes) !== null) { const bounds = getInstructionBounds(abiMarkerIndex); diff --git a/host/test/thread-wasm-patch.test.ts b/host/test/thread-wasm-patch.test.ts index ed419360f1..8b4266557a 100644 --- a/host/test/thread-wasm-patch.test.ts +++ b/host/test/thread-wasm-patch.test.ts @@ -26,6 +26,7 @@ function moduleBytes(options: { functionTypes: number[]; bodies: number[][]; exports: Array<{ name: string; index: number }>; + functionNames?: Array<{ name: string; index: number }>; start?: number; }): ArrayBuffer { const bytes = [0x00, 0x61, 0x73, 0x6d, 0x01, 0x00, 0x00, 0x00]; @@ -61,6 +62,21 @@ function moduleBytes(options: { codeContent.push(...uleb(body.length), ...body); } bytes.push(...section(10, codeContent)); + if (options.functionNames) { + const functionNameMap = [ + ...uleb(options.functionNames.length), + ...options.functionNames.flatMap((entry) => [ + ...uleb(entry.index), + ...name(entry.name), + ]), + ]; + bytes.push(...section(0, [ + ...name("name"), + 0x01, + ...uleb(functionNameMap.length), + ...functionNameMap, + ])); + } return new Uint8Array(bytes).buffer; } @@ -110,6 +126,25 @@ describe("patchWasmForThread", () => { expect((instance.exports.__abi_version as () => number)()).toBe(18); }); + it("does not rewrite a function identified only by spoofed name metadata", async () => { + const original = moduleBytes({ + types: [VOID, I32_RESULT], + functionTypes: [0, 0, 1], + bodies: [ + [], + [0x00], + [0x10, 0x01, 0x41, 0x12], + ], + exports: [{ name: "invoke_spoof", index: 2 }], + functionNames: [{ name: "__wasm_call_ctors", index: 1 }], + }); + + const patched = patchWasmForThread(original); + expect(WebAssembly.validate(patched)).toBe(true); + const { instance } = await WebAssembly.instantiate(patched); + expect(() => (instance.exports.invoke_spoof as () => number)()).toThrow(/unreachable/); + }); + it("rejects constructor evidence that does not point to a () -> () function", () => { const original = moduleBytes({ types: [VOID, I32_RESULT],