From f2be783c6bf57845776731d604cdda257fc21f5e Mon Sep 17 00:00:00 2001 From: Lann Martin Date: Sun, 13 Sep 2026 23:20:38 -0400 Subject: [PATCH] runtime: drive stores from coalesced scheduling events --- docs/architecture.md | 29 +- harness/src/wasmtime-expectations.ts | 2 +- harness/tests/wasmtime_subprocess_test.ts | 61 ++ runtime/src/embedder/streams.ts | 3 +- runtime/src/exec/boundary.ts | 904 ++++++++---------- runtime/src/exec/host_streams.ts | 82 +- runtime/src/intrinsics/async_builtins.ts | 1 + runtime/src/jspi/bridge.ts | 7 + runtime/src/task/mod.ts | 1 + runtime/src/task/scheduler.ts | 157 ++- runtime/src/task/streams.ts | 4 +- runtime/src/task/thread.ts | 2 + runtime/tests/conventions/future-gated.wasm | Bin 0 -> 1143 bytes runtime/tests/conventions/future-gated.wat | 85 ++ runtime/tests/conventions/lowering_test.ts | 41 +- runtime/tests/cross_store_driver_test.ts | 63 +- runtime/tests/driver_stale_settled_test.ts | 26 +- .../tests/driver_trap_exit_liveness_test.ts | 47 +- runtime/tests/embedder/streams_test.ts | 16 +- runtime/tests/event_driven_drain_test.ts | 435 +++++++++ runtime/tests/host_arm_test.ts | 2 + .../tests/host_boundary_preparation_test.ts | 2 +- .../tests/host_pump_reentrancy_probe_test.ts | 109 ++- runtime/tests/host_pump_test.ts | 7 +- runtime/tests/host_pump_trap_test.ts | 16 +- .../poison_before_background_listener_test.ts | 9 + .../tests/jspi/task_return_settlement_test.ts | 59 ++ runtime/tests/lift_background_return_test.ts | 6 + runtime/tests/lift_done_verdict_test.ts | 7 +- runtime/tests/parked_driver_host_call_test.ts | 117 --- runtime/tests/same_store_driver_test.ts | 302 ------ runtime/tests/settlement_pump_test.ts | 18 +- tools/wasmtime/worker.ts | 11 +- 33 files changed, 1375 insertions(+), 1256 deletions(-) create mode 100644 runtime/tests/conventions/future-gated.wasm create mode 100644 runtime/tests/conventions/future-gated.wat create mode 100644 runtime/tests/event_driven_drain_test.ts delete mode 100644 runtime/tests/parked_driver_host_call_test.ts delete mode 100644 runtime/tests/same_store_driver_test.ts diff --git a/docs/architecture.md b/docs/architecture.md index eba9bbd3..ce4fee9d 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -265,21 +265,20 @@ failed through its task/instance ownership channel. Cleanup and terminal notification are both attempted, and the first poison cause is retained even when cleanup fails. -**Overlapping drivers.** Concurrent exports may run overlapping -`driveAsync` loops on one store. The invariant is that an activation -consumes a settlement at most once and never resumes from an obsolete -settlement. `runtime/src/exec/boundary.ts` enforces this with synchronous -awaiting-membership removal, memoized Promise tags, Promise-identity -checks, and per-store pending-resumption bookkeeping. - -The asynchronous host-activity and host-settlement pumps are fallback -drivers: they stand down cooperatively when another driver is active. -This is not a ban on synchronous pump participation: `HostActivity.pump()` -services settled activations and ticks ready threads before its async -fallback checks driver depth, including while an export driver is live. -Arrival notifications wake parked drivers so they can yield or reconsider their -waits. New host-call registrations also wake incumbent drivers rather -than leaving them parked on an obsolete snapshot of pending work. +**Event-driven store service.** Each store has one coalescing runnable-work +coordinator. Host settlements and scheduler state transitions record their +result first, then request service; the coordinator stops when no work is +runnable and never remains parked racing outstanding host Promises. Host stream +and future retention remains liveness evidence, not a second pump. + +Engine-only promising-entry hops and pending resumption claims are mandatory +continuations of the current canonical transfer, not extra guest scheduling +points. Autonomous service therefore waits while an unsettled entry hop exists. +A hop's own queued settlement may still dispatch, and registration of a genuine +`SuspensionPoint` park requests service because it removes that barrier. Direct +call, cancellation, and synchronous entry paths remain distinct from ordinary +store draining. A bounded work quantum yields to platform timers and I/O only +between complete canonical steps. **Between-calls progress.** A host import settling can resume background guest work even with no export call in flight. A task waiting for the diff --git a/harness/src/wasmtime-expectations.ts b/harness/src/wasmtime-expectations.ts index e8419e4b..44e03e6c 100644 --- a/harness/src/wasmtime-expectations.ts +++ b/harness/src/wasmtime-expectations.ts @@ -180,7 +180,7 @@ export const WASMTIME_EXPECTATION_GROUPS: readonly WasmtimeExpectationGroup[] = rows: [{ lines: [65], cause: - 'Error: expected trap "wasm trap: cannot block a synchronous task before returning", got "wasm trap: deadlock detected: event loop cannot make further progress (export \'run\': every suspended activation is waiting on a suspension only this scheduler could resume, and none is ready)"', + 'Error: expected trap "wasm trap: cannot block a synchronous task before returning", got "wasm trap: deadlock detected: event loop cannot make further progress (export \'run\': no runnable work or host call is outstanding)"', status: "failed", }], }, diff --git a/harness/tests/wasmtime_subprocess_test.ts b/harness/tests/wasmtime_subprocess_test.ts index 4ea7e1c4..75be46d3 100644 --- a/harness/tests/wasmtime_subprocess_test.ts +++ b/harness/tests/wasmtime_subprocess_test.ts @@ -26,6 +26,67 @@ Deno.test({ }, }); +Deno.test({ + name: "worker-style final output is complete before prompt explicit exit", + ignore: !canRun, + fn: async () => { + const expected = { + source: "x.wast", + results: [{ line: 1, type: "module", status: "passed" }], + }; + const script = ` + const bytes = new TextEncoder().encode(JSON.stringify(${ + JSON.stringify(expected) + }) + "\\n"); + let written = 0; + while (written < bytes.length) written += Deno.stdout.writeSync(bytes.subarray(written)); + setInterval(() => {}, 10); + Deno.exit(0); + `; + const started = performance.now(); + const outcome = await runChild( + new Deno.Command(Deno.execPath(), { + args: ["eval", script], + stdout: "piped", + stderr: "piped", + }), + 1_000, + ); + if (outcome === "timeout" || !outcome.success) { + throw new Error("explicit worker exit did not complete"); + } + if (performance.now() - started > 900) { + throw new Error("worker did not exit promptly"); + } + const parsed = JSON.parse(new TextDecoder().decode(outcome.stdout)); + const malformed = validateWorkerResult( + { source_filename: "x.wast", commands: [{ line: 1, type: "module" }] }, + parsed, + ); + if (malformed !== undefined) throw new Error(malformed); + }, +}); + +Deno.test({ + name: "worker failure before final output remains non-green", + ignore: !canRun, + fn: async () => { + const outcome = await runChild( + new Deno.Command(Deno.execPath(), { + args: ["eval", 'throw new Error("worker failed before result")'], + stdout: "piped", + stderr: "piped", + }), + 1_000, + ); + if (outcome === "timeout") throw new Error("failed worker timed out"); + if (outcome.success) throw new Error("failed worker exited successfully"); + if (outcome.stdout.length !== 0) { + throw new Error("failed worker wrote a partial result"); + } + }, +}); + Deno.test("malformed worker result is rejected", () => { const doc = { source_filename: "x.wast", diff --git a/runtime/src/embedder/streams.ts b/runtime/src/embedder/streams.ts index b787e993..e694e831 100644 --- a/runtime/src/embedder/streams.ts +++ b/runtime/src/embedder/streams.ts @@ -102,12 +102,13 @@ function reportProducerFailure( ? cause : new StreamProducerError(where, cause); const shared = host.value as unknown as { - boundStore?: { hostFailure?: unknown } | null; + boundStore?: { hostFailure?: unknown; requestService?: () => void } | null; }; producerFailures.set(host.value as object, err); const store = shared.boundStore; if (store != null && typeof store === "object") { if (store.hostFailure === undefined) store.hostFailure = err; + store.requestService?.(); } return err; } diff --git a/runtime/src/exec/boundary.ts b/runtime/src/exec/boundary.ts index bf1b9f34..76aa85ef 100644 --- a/runtime/src/exec/boundary.ts +++ b/runtime/src/exec/boundary.ts @@ -29,7 +29,7 @@ import { type PreparedTransfer, prepareRawValues, } from "../cabi/values.ts"; -import { assert_, AssertionError } from "../cabi/trap.ts"; +import { assert_, AssertionError, Trap } from "../cabi/trap.ts"; import { type BlockRequest, type Cancelled, @@ -39,18 +39,17 @@ import { entryRefusal, EventCode, type EventTuple, + guestActivationLive, + hasHostRetention, hasRealHostCall, instancePoisonCause, isInstancePoisoned, NeedsJspi, needsJspi, notifyInstancePoisoned, - OriginatedSchedulerFailure, packSubtaskResult, PendingCapability, - realHostCalls, type Store, - storeQuiescent, Subtask, SubtaskState, SyncEntryBusy, @@ -414,12 +413,13 @@ function drive( done: () => boolean, what: string, idle: IdlePolicy = "trap", + onBackgroundFailure?: (cause: unknown) => boolean, ): DriveExit | Promise { try { - return driveLoop(store, done, what, idle); + return driveLoop(store, done, what, idle, onBackgroundFailure); } catch (e) { // Sibling host calls and activation tails survive this driver's failure. - ensureSettlementPump(store); + requestStoreService(store); throw e; } } @@ -430,577 +430,416 @@ function driveLoop( done: () => boolean, what: string, idle: IdlePolicy, + onBackgroundFailure?: (cause: unknown) => boolean, ): DriveExit | Promise { - for (;;) { - traceDrive("drive", store, done, "top"); - // Promise-parked threads need microtasks; a synchronous YIELD loop would - // starve them. Hand those stores to the interleaved async drain. - while (store.awaiting.size === 0 && store.tick()) { - traceDrive("drive", store, done, "ticked"); - if (store.hostFailure !== undefined) { - throw takeHostFailure(store); - } - } - if (store.hostFailure !== undefined) { - throw takeHostFailure(store); - } + traceDrive("drive", store, done, "top"); + if (store.hostFailure !== undefined) throw takeHostFailure(store); + if (done()) { + traceDrive("drive", store, done, "EXIT-done"); + requestStoreService(store); + return "done"; + } + const drainState = stateFor(store); + // Use the same ordinary-service authority as the asynchronous coordinator. + // Nested entry cannot pass its live-activation gate; top-level synchronous + // entry may still drain ready work before deciding its deadlock verdict. + while (store.awaiting.size === 0 && serviceOrdinaryStep(store)) { + if (store.hostFailure !== undefined) throw takeHostFailure(store); if (done()) { - traceDrive("drive", store, done, "EXIT-done"); - // No async finally will run: hand off any background work here. - ensureSettlementPump(store); + requestStoreService(store); return "done"; } - // Awaiting activations and pending resumptions need an event-loop turn. - if (store.awaiting.size > 0 || store.hasPendingResumptions()) { - traceDrive("drive", store, done, "->async(awaiting/pending)"); - return driveAsync(store, done, what, idle); - } - if (store.pendingHostCalls.size === 0) { - // An idle async task remains live for a later driver. - if (idle === "exit") { - traceDrive("drive", store, done, "EXIT-idle"); - ensureSettlementPump(store); - return "idle"; - } - traceDrive("drive", store, done, "DEADLOCK-TRAP"); - trapIf( - true, - `wasm trap: deadlock detected: event loop cannot make further ` + - `progress (${what}: no thread is ready and no host call is ` + - `outstanding)`, + if (chargeWorkQuantum(drainState)) { + // CONTRACT: This is ordinary async scheduling, not definitions.py's + // direct driveSyncLift loop. Share the coordinator's quantum and cross a + // platform-task boundary before continuing so guest YIELD cannot starve + // timers (scheduler-cycle2.md:9-11). + void handoffWorkQuantum(store, drainState); + return driveAsync( + store, + done, + what, + idle, + onBackgroundFailure, + false, ); } - traceDrive("drive", store, done, "->async(hostcalls)"); - return driveAsync(store, done, what, idle); } + traceDrive("drive", store, done, "->store-service"); + return driveAsync(store, done, what, idle, onBackgroundFailure); } -/** A settled parked-thread promise, tagged with the thread that owns it. */ -type AwaitWinner = { - t: { - awaiting: Promise | null; - resumeWith(v: unknown, f?: { error: unknown }): void; - }; - /** Park identity: membership alone cannot distinguish a later re-park. */ - p: Promise; - value: unknown; - failure: { error: unknown } | undefined; -}; - /** - * Tagged promises, memoized by the *promise* (not the thread) so re-racing on - * every turn does not attach a fresh continuation to the same promise, and so - * a thread that parks again later can never pick up a stale tag. + * Register real host work. Its own reaction requests service after the host + * settlement's earlier reaction has committed readiness and removed the call. + * The coordinator never races or polls the Promise itself. */ -const taggedAwaits = new WeakMap, Promise>(); - -function tagAwait(t: AwaitWinner["t"]): Promise { - const p = t.awaiting!; - let tag = taggedAwaits.get(p); - if (tag === undefined) { - tag = p.then( - (value): AwaitWinner => ({ t, p, value, failure: undefined }), - (e): AwaitWinner => ({ t, p, value: undefined, failure: { error: e } }), - ); - taggedAwaits.set(p, tag); - } - return tag; +export function registerHostCall( + store: Store, + promise: Promise, +): void { + store.pendingHostCalls.add(promise); + promise.then( + () => store.requestService(), + () => store.requestService(), + ); } -/** - * Shared async driver for host activity and settlement pumps. Service tails, - * tick, then race awaiting activations and host calls. A pump that must stay - * pending rather than trap on idle must make `done()` true whenever - * `pendingHostCalls` is empty; idle traps require the opposite exit decision. - */ -export async function driveStoreAsync( - store: Store, - done: () => boolean, - what: string, -): Promise { - // The exit verdict is for `drive`'s lift caller (see `DriveExit`); the - // pumps drive to quiescence and have nothing to decide on it. - for (;;) { - try { - await driveAsync(store, done, what); - return; - } catch (e) { - if (consumeSchedulerFailure(store, e)) continue; - if (e instanceof HostFailureReport) throw e.cause; - throw e; - } +type DrainWaiter = { + promise: Promise; + done: () => boolean; + idle: IdlePolicy; + what: string; + resolve: (exit: DriveExit) => void; + reject: (cause: unknown) => void; + onBackgroundFailure?: (cause: unknown) => boolean; + idleReported: boolean; + idleProbeArmed: boolean; +}; + +type DrainState = { + scheduled: boolean; + running: boolean; + requested: boolean; + yielding: boolean; + waiters: Set; + budget: number; + hopProbe: { hops: Set; elapsed: boolean } | null; +}; + +const drainStates = new WeakMap(); +const DRAIN_TICK = Promise.resolve(); +const WORK_QUANTUM = 8; + +function stateFor(store: Store): DrainState { + let state = drainStates.get(store); + if (state === undefined) { + state = { + scheduled: false, + running: false, + requested: false, + yielding: false, + waiters: new Set(), + budget: WORK_QUANTUM, + hopProbe: null, + }; + drainStates.set(store, state); + store.serviceRequested = () => requestStoreService(store); } + return state; } -/** - * Live async drivers per store. Concurrent exports may overlap; fallback - * pumps stand down cooperatively when another driver arrives. There is no - * single-driver invariant. - * - * Each settlement must be delivered once to its original park. `resumeWith` - * deletes awaiting membership synchronously; race winners check both that - * membership and promise identity, then remove queued copies before resuming. - * Per-promise tags share settlement reactions across racers. Per-store - * pending-resumption gates give engine continuations a turn before more ticks. - */ -const driverDepth = new WeakMap(); -const driverIdle = new WeakMap; r: () => void }>(); +/** Request the store's single event-driven drain. Requests coalesce, but every + * settlement/event remains recorded in its owning state object. */ +export function requestStoreService(store: Store): void { + const state = stateFor(store); + state.requested = true; + if (state.running || state.scheduled || state.yielding) return; + state.scheduled = true; + DRAIN_TICK.then(() => runStoreDrain(store, state)); +} -export function storeDriverDepth(store: Store): number { - return driverDepth.get(store) ?? 0; +/** @internal Test-only visibility into call/service lifecycle ownership. */ +export function drainWaiterCountForTesting(store: Store): number { + return drainStates.get(store)?.waiters.size ?? 0; } -/** Resolves once no `driveAsync` loop is live on `store`. */ -export function whenStoreDriverIdle(store: Store): Promise { - if (storeDriverDepth(store) === 0) return Promise.resolve(); - let w = driverIdle.get(store); - if (w === undefined) { - let r!: () => void; - const p = new Promise((res) => (r = res)); - w = { p, r }; - driverIdle.set(store, w); - } - return w.p; +function ordinaryServiceAllowed(store: Store): boolean { + return !guestActivationLive(store) && !store.hasPendingResumptions() && + unsettledEntryHops(store).length === 0; } -// --------------------------------------------------------------------------- -// Driver arrival -// --------------------------------------------------------------------------- -// -// Wake incumbents so they release speculative gates and re-evaluate `done` -// without waiting for a possibly unbounded host call to settle. -const driverArrivals = new WeakMap< - Store, - { p: Promise; r: () => void } ->(); - -/** A one-shot that resolves (to `null`, the race's "nothing settled" value) - * when another driver starts on `store`. */ -function armDriverArrival(store: Store): Promise { - let n = driverArrivals.get(store); - if (n === undefined) { - let r!: () => void; - const p = new Promise((res) => (r = () => res(null))); - n = { p, r }; - driverArrivals.set(store, n); - } - return n.p; +function hasRunnable(store: Store): boolean { + return ordinaryServiceAllowed(store) && + (store.hasServiceableSettled() || store.readyCandidates().length > 0); } -function fireDriverArrival(store: Store): void { - const n = driverArrivals.get(store); - if (n === undefined) return; - // Deleted before resolving so the next `armDriverArrival` mints a fresh, - // unresolved one-shot: a driver that wakes on this and re-parks must not - // pick the settled promise back up and spin. - driverArrivals.delete(store); - n.r(); +/** The sole authority for ordinary store-wide progress. Direct canonical + * switches and driveSyncLift remain separate. */ +function serviceOrdinaryStep(store: Store): boolean { + if (!ordinaryServiceAllowed(store)) return false; + if (store.hasServiceableSettled()) return store.serviceSettledStep(); + if (hasEntryHop(store)) return false; + return store.tick(); } -// --------------------------------------------------------------------------- -// Host-call arrival -// --------------------------------------------------------------------------- -// -// Promise races watch snapshots. Synchronous export entry or host-activity -// draining can register a call without starting a new async driver. Announce -// every registration so parked drivers refresh their snapshots independently -// of the driver-arrival stand-down signal. -const hostCallArrivals = new WeakMap< - Store, - { p: Promise; r: () => void } ->(); - -/** A one-shot that resolves (to `null`, the race's "nothing settled" value) - * when a new host call is registered on `store`. */ -function armHostCallArrival(store: Store): Promise { - let n = hostCallArrivals.get(store); - if (n === undefined) { - let r!: () => void; - const p = new Promise((res) => (r = () => res(null))); - n = { p, r }; - hostCallArrivals.set(store, n); - } - return n.p; +/** Admission helpers may finish one queued tail, but may neither tick a ready + * sibling nor cross a live/pending/unsettled activation boundary. */ +function serviceAdmissionTailStep(store: Store): boolean { + if (!ordinaryServiceAllowed(store)) return false; + return store.serviceSettledStep(); } -function fireHostCallArrival(store: Store): void { - const n = hostCallArrivals.get(store); - if (n === undefined) return; - // Deleted before resolving, exactly as `fireDriverArrival`: a racer that - // wakes on this and re-parks must mint a fresh, unresolved one-shot rather - // than pick the settled promise back up and spin. - hostCallArrivals.delete(store); - n.r(); +function chargeWorkQuantum(state: DrainState): boolean { + return --state.budget <= 0; } -/** - * Register real host work and wake parked racers. Use this rather than adding - * directly to `pendingHostCalls`. HostActivity arms are different: they mean - * the embedder may act, not that an external result is outstanding. - */ -export function registerHostCall( +function handoffWorkQuantum( store: Store, - promise: Promise, -): void { - store.pendingHostCalls.add(promise); - fireHostCallArrival(store); + state: DrainState, +): Promise { + state.yielding = true; + return new Promise((resolve) => { + setTimeout(() => { + state.budget = WORK_QUANTUM; + state.yielding = false; + resolve(); + requestStoreService(store); + }, 0); + }); } -// --------------------------------------------------------------------------- -// The settlement pump: liveness between export calls -// --------------------------------------------------------------------------- -// -// Every driver exit, including exceptions, hands off real host calls, queued -// tails and hop-parked activations. The keeper drives their wakeups without -// requiring another export call. It stands down for live drivers, stops at -// quiescence rather than task completion, and parks failures on hostFailure. -// Activity arms are excluded. Re-arming a live keeper nudges it to refresh its -// snapshot, including calls registered by a drive it performed itself. - -const settlementPumps = new WeakSet(); -const settlementNudges = new WeakMap< - Store, - { p: Promise; r: () => void } ->(); - -function armSettlementNudge(store: Store): Promise { - let n = settlementNudges.get(store); - if (n === undefined) { - let r!: () => void; - const p = new Promise((res) => (r = res)); - n = { p, r }; - settlementNudges.set(store, n); - } - return n.p; +/** Awaiting engine entries that have not reached a genuine SuspensionPoint. + * Their Wasm continuation is a mandatory part of the current canonical + * transfer and must run before ordinary Store.tick scheduling. */ +function hasEntryHop(store: Store): boolean { + return entryHopThreads(store).length > 0; } -function fireSettlementNudge(store: Store): void { - const n = settlementNudges.get(store); - if (n !== undefined) { - settlementNudges.delete(store); - n.r(); - } +/** An engine-only hop whose activation has not yet produced its own settled + * tail. Autonomous service must not dispatch another tail across this window: + * the engine continuation may still register a genuine SuspensionPoint park. + * A queued settlement is excluded so its own tail can close the hop rather + * than deadlocking behind itself. */ +function unsettledEntryHops(store: Store): unknown[] { + if (store.awaiting.size === 0) return []; + const queued = new Set(store.settled.map((s) => s.t)); + return entryHopThreads(store).filter((t) => !queued.has(t)); } -/** - * Work needing an owner after driver exit, including exception exits. - */ -function pumpWork(store: Store): boolean { - return hasRealHostCall(store) || store.settled.length > 0 || - entryHopThreads(store).length > 0; +function sameIdentities(a: Set, b: readonly unknown[]): boolean { + return a.size === b.length && b.every((value) => a.has(value)); } -/** - * Ensure a settlement pump owns outstanding host calls, tails and entry hops. - * Idempotent and cheap; called at every driver exit. Never throws. - */ -export function ensureSettlementPump(store: Store): void { - if (settlementPumps.has(store)) { - // Already parked (or driving): wake it so it re-snapshots the race — - // this call may be reporting host calls registered after it parked. - fireSettlementNudge(store); - return; +function removeDrainWaiter(store: Store, promise: Promise): void { + const state = drainStates.get(store); + if (state === undefined) return; + for (const waiter of state.waiters) { + if (waiter.promise === promise) { + state.waiters.delete(waiter); + // The caller-owned terminal channel has already settled. Resolve this + // now-obsolete service observation so its `.then` closure is not left on + // an unreachable pending Promise; publishIfEligible is terminal-guarded. + waiter.resolve("done"); + return; + } } - if (store.hostFailure !== undefined) return; - if (!pumpWork(store)) return; - settlementPumps.add(store); - void settlementPumpLoop(store); } -async function settlementPumpLoop(store: Store): Promise { - let failed = false; - try { - for (;;) { - // Stand down while any driver is live: it races `pendingHostCalls` - // itself and services settlements on the guest's behalf. - while (storeDriverDepth(store) > 0) { - await whenStoreDriverIdle(store); - } - // A parked failure belongs to the next embedder call (the only place - // it can surface); driving into it here would just consume and re-park - // it in a loop. - if (store.hostFailure !== undefined) return; - const real = realHostCalls(store); - // Queued tails need no await; orphaned entry hops are raced with host work. - const hops = store.settled.length > 0 - ? [] - : entryHopThreads(store).map((t) => t.awaiting).filter(( - p, - ): p is Promise => p !== null); - if (store.settled.length === 0) { - if (real.length === 0 && hops.length === 0) return; - const nudge = armSettlementNudge(store); - // Driver exits nudge this snapshot; active drivers own new work meanwhile. - // Registration continuations, not this race, report host rejections. - await Promise.race([ - ...real.map((p) => p.then(() => {}, () => {})), - ...hops.map((p) => p.then(() => {}, () => {})), - nudge, - ]); - if (storeDriverDepth(store) > 0) continue; - } - // Drain after every wake: storeQuiescent does not count ready waiters - // left by a host settlement that already removed its call registration. - await driveStoreAsync( - store, - // Stop before idle traps, at quiescence, or when another driver arrives. - () => - store.pendingHostCalls.size === 0 || - storeQuiescent(store) || - storeDriverDepth(store) > 1, - "settlement pump", - ); +function rejectOneWaiter(state: DrainState, cause: unknown): boolean { + for (const waiter of state.waiters) { + if ( + waiter.onBackgroundFailure !== undefined && + !waiter.onBackgroundFailure(cause) + ) { + continue; + } + state.waiters.delete(waiter); + if (waiter.onBackgroundFailure === undefined) waiter.reject(cause); + return true; + } + return false; +} + +function settleWaiters(store: Store, state: DrainState): void { + for (const waiter of [...state.waiters]) { + if (waiter.done()) { + state.waiters.delete(waiter); + waiter.resolve("done"); + continue; } - } catch (e) { - failed = true; - store.hostFailure ??= e; - } finally { - settlementPumps.delete(store); - // Close the exit race: an `ensureSettlementPump` that saw us live and - // fired the nudge after our last snapshot check must not be lost. if ( - !failed && store.hostFailure === undefined && - storeDriverDepth(store) === 0 && pumpWork(store) + hasRunnable(store) || store.hasPendingResumptions() || + hasEntryHop(store) || hasRealHostCall(store) || hasHostRetention(store) ) { - ensureSettlementPump(store); + continue; + } + if (waiter.idle === "exit") { + // Resolving the driver's idle verdict must not discard the call's + // failure route. An async call remains active after going idle, and a + // later producer failure still belongs to that unresolved call. + if (!waiter.idleReported) { + waiter.idleReported = true; + waiter.resolve("idle"); + } + continue; + } + if (!waiter.idleProbeArmed) { + waiter.idleProbeArmed = true; + setTimeout(() => requestStoreService(store), 0); + continue; + } + state.waiters.delete(waiter); + try { + trapIf( + true, + `wasm trap: deadlock detected: event loop cannot make further ` + + `progress (${waiter.what}: no runnable work or host call is outstanding)`, + ); + } catch (e) { + waiter.reject(e); } } } -async function driveAsync( - store: Store, - done: () => boolean, - what: string, - idle: IdlePolicy = "trap", -): Promise { - const depth = storeDriverDepth(store) + 1; - driverDepth.set(store, depth); - // Wake incumbents to release speculative gates and let fallback pumps stand down. - if (depth > 1) fireDriverArrival(store); +/** Apply idle policy after an unsettled implementation hop has had its one + * event-loop opportunity to become a real park or queue its own settlement. */ +function settleUnprogressableHop(state: DrainState): void { + for (const waiter of [...state.waiters]) { + if (waiter.done()) { + state.waiters.delete(waiter); + waiter.resolve("done"); + } else if (waiter.idle === "exit") { + if (!waiter.idleReported) { + waiter.idleReported = true; + waiter.resolve("idle"); + } + } else { + state.waiters.delete(waiter); + waiter.reject( + new Trap( + `wasm trap: deadlock detected: event loop cannot make further ` + + `progress (${waiter.what}: no runnable work or host call is outstanding)`, + ), + ); + } + } +} + +async function runStoreDrain(store: Store, state: DrainState): Promise { + if (state.running) return; + state.scheduled = false; + // A synchronous prefix may have exhausted the shared budget after this + // drain's microtask was queued. Its platform-task handoff owns rescheduling. + if (state.yielding) return; + state.running = true; try { - let claimHops = 0; for (;;) { - traceDrive("driveAsync", store, done, "top"); - // Complete settled activation bookkeeping before any scheduling decision. - store.serviceSettled(); - if (store.hostFailure !== undefined) { - throw takeHostFailure(store); - } - // Yield for this store's engine resumptions. Only their execution/park - // or settlement may release them; never clear other owners' entries. - if (store.hasPendingResumptions()) { - traceDrive("driveAsync", store, done, "yield-pending"); - // Bound leaked claims, interleaving timer turns to avoid starving - // the event loop while diagnosing an internal scheduling failure. - claimHops++; - assert_( - claimHops < 10_000, - "driveAsync: a resumed-activation claim was never released " + - "(the activation neither parked, finished, nor trapped)", - ); - if (claimHops % 100 === 0) { - await new Promise((r) => setTimeout(r, 0)); - } else { - await Promise.resolve(); - } - continue; - } - claimHops = 0; - while (store.tick()) { - if (store.hostFailure !== undefined) { - throw takeHostFailure(store); - } - // A READY/YIELD loop must not starve promise settlements. Give engine - // continuations a microtask per tick and service any landed tails first. - if (store.awaiting.size > 0) { - await Promise.resolve(); - if (store.hasServiceableSettled()) break; - } + state.requested = false; + // A request made from consumePendingIfRunning is queued before the guest + // frame unwinds. Never turn that notification into reentrant scheduling. + if (guestActivationLive(store)) { + state.requested = true; + return; } if (store.hostFailure !== undefined) { - throw takeHostFailure(store); - } - if (done()) { - traceDrive("driveAsync", store, done, "EXIT-done"); - return "done"; - } - // Queued tails and pending resumptions take priority over parking. - if (store.hasServiceableSettled() || store.hasPendingResumptions()) { - continue; + const cause = store.hostFailure; + if (rejectOneWaiter(state, cause)) { + store.hostFailure = undefined; + continue; + } else return; } - // Race all activations and host calls. Awaiting one chosen activation - // alone could stop the scheduler that its nested suspension needs. - if (store.awaiting.size > 0) { - // With no external work or pending resumption, allow a timer turn for - // engine hops to settle before declaring idle. Internal activation - // promises alone do not establish that further progress is possible. - if ( - store.pendingHostCalls.size === 0 && !store.hasPendingResumptions() - ) { - traceDrive("driveAsync", store, done, "deadlock-probe"); - // Queued tails belong to serviceSettled; racing their settled tags - // repeatedly would create an unbounded microtask loop. - const queued = new Set(store.settled.map((s) => s.t)); - const parked = ([...store.awaiting] as AwaitWinner["t"][]).filter( - (t) => !queued.has(t), - ); - const progressed = await Promise.race([ - ...parked.map((t) => tagAwait(t).then(() => true)), - new Promise((r) => setTimeout(() => r(false), 0)), - ]); - traceDrive( - "driveAsync", - store, - done, - `deadlock-probe:progressed=${progressed}`, - ); - if (!progressed) { - // Revalidate the snapshot after awaiting. Apply the same queued-tail - // filter to both snapshots, using the current queue for the new one. - const freshQueued = new Set(store.settled.map((s) => s.t)); - const fresh = ([...store.awaiting] as AwaitWinner["t"][]).filter( - (t) => !freshQueued.has(t), - ); - const changed = fresh.length !== parked.length || - fresh.some((t, i) => t !== parked[i]); - if (changed) continue; - // Even unchanged awaiting membership can acquire external work, - // pending resumptions or queued tails during the probe. - if ( - store.pendingHostCalls.size > 0 || - store.hasPendingResumptions() || - store.hasServiceableSettled() - ) { - continue; - } - if (store.readyCandidates().length === 0) { - if (idle === "exit") { - traceDrive("driveAsync", store, done, "EXIT-idle"); - return "idle"; + try { + const unsettledHops = unsettledEntryHops(store); + if (unsettledHops.length > 0) { + // Host retention and real host calls both make this a valid wait. + // Neither is runnable work, so stop until their own notifications. + if (hasRealHostCall(store) || hasHostRetention(store)) { + state.hopProbe = null; + return; + } + if ( + state.hopProbe === null || + !sameIdentities(state.hopProbe.hops, unsettledHops) + ) { + const probe = { + hops: new Set(unsettledHops), + elapsed: false, + }; + state.hopProbe = probe; + setTimeout(() => { + if (state.hopProbe === probe) { + probe.elapsed = true; + requestStoreService(store); } - trapIf( - true, - `wasm trap: deadlock detected: event loop cannot make ` + - `further progress (${what}: every suspended activation is ` + - `waiting on a suspension only this scheduler could resume, ` + - `and none is ready)`, - ); - } - // A thread became ready: tick it rather than awaiting its dependents. - continue; + }, 0); + return; } - // Consume the settlement, either through the queue or the race below. - } - // The probe awaited: another driver may have consumed the park, or - // noteAwaiting may have queued its tail. Recheck before selecting one. - if (store.awaiting.size === 0 || store.hasServiceableSettled()) { - continue; - } - // Race only parks not already owned by the settled queue. - const queued = new Set(store.settled.map((s) => s.t)); - const parked = ([...store.awaiting] as AwaitWinner["t"][]).filter( - (t) => !queued.has(t), - ); - if (parked.length === 0) { - // Defensive fallback: the checks above imply a non-empty awaiting - // set and empty settled queue, so filtering cannot remove all parks. - if (store.pendingHostCalls.size > 0) { - await Promise.race([ - ...store.pendingHostCalls, - armDriverArrival(store), - armHostCallArrival(store), - ]).catch(() => {}); - continue; + if (!state.hopProbe.elapsed) { + return; } - traceDrive("driveAsync", store, done, "DEADLOCK-TRAP-deferred"); - trapIf( - true, - `wasm trap: deadlock detected: event loop cannot make further ` + - `progress (${what}: every settled activation tail is deferred ` + - `on a non-enterable instance and no host call is outstanding)`, - ); - } - const chosen = parked[0]; - const chosenTag = tagAwait(chosen); - const others: Promise[] = parked.slice(1).map( - tagAwait, - ); - for (const h of store.pendingHostCalls) { - others.push(h.then(() => null, () => null)); - } - // A sole driver may gate ticks speculatively while the engine runs - // chosen's activation. Driver arrival breaks the race and releases the - // gate, preventing an unbounded host wait from blocking a second loop. - // Remove only an identity this loop inserted, never clear the set. - // Genuine SuspensionPoint resumptions establish their own entries. - const sole = storeDriverDepth(store) === 1; - const added = sole && !store.pendingResumptions.has(chosen); - if (added) store.addPendingResumption(chosen); - let winner: AwaitWinner | null; - try { - // Arrivals trigger stand-down or snapshot refresh without host settlement. - winner = await Promise.race([ - chosenTag, - ...others, - armDriverArrival(store), - armHostCallArrival(store), - ]); - } finally { - if (added) store.removePendingResumption(chosen); + state.hopProbe = null; + settleUnprogressableHop(state); + return; } - // Deliver the actual winner only if its park is still current. - // Delete queued copies before resumeWith can synchronously re-park; - // otherwise serviceSettled could deliver this result to the new park. - if ( - winner !== null && store.awaiting.has(winner.t) && - winner.t.awaiting === winner.p - ) { - for (let i = store.settled.length - 1; i >= 0; i--) { - if (store.settled[i].t === winner.t) store.settled.splice(i, 1); + state.hopProbe = null; + while (!store.hasPendingResumptions()) { + if (!serviceOrdinaryStep(store)) break; + if (chargeWorkQuantum(state)) { + await handoffWorkQuantum(store, state); } - const task = (winner.t as { - task?: { failureOwner?: unknown }; - }).task; - const origin = task?.failureOwner ?? task; - try { - winner.t.resumeWith(winner.value, winner.failure); - } catch (e) { - throw new OriginatedSchedulerFailure(origin, e); + if ( + store.awaiting.size > 0 && !store.hasServiceableSettled() + ) { + // JSPI continuations and promise settlements queued by this tick + // precede the next ordinary scheduling choice. + await Promise.resolve(); } } + } catch (e) { + if (consumeSchedulerFailure(store, e)) continue; + const cause = e instanceof HostFailureReport ? e.cause : e; + if (!rejectOneWaiter(state, cause)) store.hostFailure ??= cause; continue; } - if (store.pendingHostCalls.size === 0) { - if (idle === "exit") { - traceDrive("driveAsync", store, done, "EXIT-idle"); - return "idle"; - } - traceDrive("driveAsync", store, done, "DEADLOCK-TRAP"); - trapIf( - true, - `wasm trap: deadlock detected: event loop cannot make further ` + - `progress (${what}: no thread is ready and no host call is ` + - `outstanding)`, - ); - } - traceDrive("driveAsync", store, done, "await-race"); - // Host settlement order is external. Arrivals must also wake this park - // so fallback drivers can stand down and all drivers refresh snapshots. - await Promise.race([ - ...store.pendingHostCalls, - armDriverArrival(store), - armHostCallArrival(store), - ]).catch(() => {}); + settleWaiters(store, state); + if (hasRunnable(store)) continue; + // A real claim is released only when its activation executes, parks, or + // settles. Those exact transitions request service; do not microtask-poll. + if (store.hasPendingResumptions()) return; + // Requests raised while this drain was executing belong to the next + // microtask. Do not fold them into the current synchronous drain: the + // notifying canonical mutation must return first. + return; } } finally { - const left = storeDriverDepth(store) - 1; - driverDepth.set(store, left); - if (left === 0) { - const w = driverIdle.get(store); - driverIdle.delete(store); - w?.r(); - // The last async driver hands off any remaining pump work. - ensureSettlementPump(store); + state.running = false; + if (state.requested) requestStoreService(store); + } +} + +function driveAsync( + store: Store, + done: () => boolean, + what: string, + idle: IdlePolicy = "trap", + onBackgroundFailure?: (cause: unknown) => boolean, + requestService = true, +): Promise { + let resolve!: (exit: DriveExit) => void; + let reject!: (cause: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + stateFor(store).waiters.add({ + promise, + done, + idle, + what, + resolve, + reject, + onBackgroundFailure, + idleReported: false, + idleProbeArmed: false, + }); + if (requestService) requestStoreService(store); + return promise; +} + +export async function driveStoreAsync( + store: Store, + done: () => boolean, + what: string, +): Promise { + for (;;) { + try { + await driveAsync(store, done, what); + return; + } catch (e) { + if (consumeSchedulerFailure(store, e)) continue; + if (e instanceof HostFailureReport) throw e.cause; + throw e; } } } @@ -1170,12 +1009,20 @@ export function createLiftedFunction(input: { let terminalCause: unknown; let invocationReturned = false; let futurePublicationQueued = false; + let serviceWaiter: Promise | null = null; let waiter: { promise: Promise; resolve: (value: unknown) => void; reject: (cause: unknown) => void; } | null = null; + const detachServiceWaiter = (): void => { + if (serviceWaiter !== null) { + removeDrainWaiter(store, serviceWaiter); + serviceWaiter = null; + } + }; + const terminalPromise = (): Promise => { if (waiter === null) { let resolve!: (value: unknown) => void; @@ -1226,12 +1073,14 @@ export function createLiftedFunction(input: { `${name}: task resolved as cancelled, but the host never requested cancellation`, ); terminal = true; + detachServiceWaiter(); task.detachCall(); waiter?.reject(terminalCause); return; } terminalValue = resultsToHost(resolved); terminal = true; + detachServiceWaiter(); task.detachCall(); // This runs only at a control-return boundary, outside canonical state // mutation, so Promise thenable inspection is safe here. @@ -1334,6 +1183,7 @@ export function createLiftedFunction(input: { const failCall = (e: unknown): boolean => { if (terminal) return false; terminal = true; + detachServiceWaiter(); task.detachCall(); terminalFailed = true; terminalCause = e; @@ -1402,6 +1252,7 @@ export function createLiftedFunction(input: { driveDone, `export '${name}'`, idlePolicy, + (cause) => failCall(cause), ); break; } catch (e) { @@ -1447,6 +1298,8 @@ export function createLiftedFunction(input: { throw e; } } + serviceWaiter = outcome; + if (terminal) detachServiceWaiter(); // The driver may remain parked on unrelated host work after this call's // result becomes eligible. Observe it for this call's own deadlock/fault, // but return the call-owned completion channel instead (#350/#357). @@ -1480,7 +1333,9 @@ export function createLiftedFunction(input: { // never bypass an unsettled hop, which would expose result memory. if (input.refuseOnEntryHops && store.hasServiceableSettled()) { try { - store.serviceSettled(); + while (serviceAdmissionTailStep(store)) { + // Recheck permission after every guest tail. + } } catch (e) { if (!consumeSchedulerFailure(store, e)) { prepared?.cleanup(e); @@ -1560,22 +1415,14 @@ function entryHopThreads( * gated callers recheck independently, with no FIFO admission guarantee. */ async function awaitHopQuiescence(store: Store, inst: unknown): Promise { + requestStoreService(store); + await DRAIN_TICK; for (;;) { const hops = entryHopThreads(store, inst); if (hops.length === 0) return; - await Promise.race( - hops.map((t) => - (t.awaiting ?? Promise.resolve()).then( - () => undefined, - () => undefined, - ) - ), - ); - try { - store.serviceSettled(); - } catch (e) { - if (!consumeSchedulerFailure(store, e)) throw e; - } + const revision = store.serviceProgressRevision(); + await store.waitForServiceProgress(revision); + await DRAIN_TICK; } } @@ -2094,6 +1941,7 @@ export function createLoweredImport(input: { store.pendingHostCalls.delete(registered); } outcome = done; + store.requestService(); }); registered = promise; // Mark this park externally wakeable for drivers and teardown. @@ -2155,6 +2003,8 @@ export function createLoweredImport(input: { ) { store.hostFailure = e; } + } finally { + store.requestService(); } }); registered = promise; diff --git a/runtime/src/exec/host_streams.ts b/runtime/src/exec/host_streams.ts index 5563fa9a..389e7663 100644 --- a/runtime/src/exec/host_streams.ts +++ b/runtime/src/exec/host_streams.ts @@ -16,17 +16,12 @@ import { assert_ } from "../cabi/trap.ts"; import { despecialize } from "../cabi/types.ts"; import type { ComponentValue, ValType } from "../cabi/types.ts"; -import { - driveStoreAsync, - storeDriverDepth, - whenStoreDriverIdle, -} from "./boundary.ts"; +import { requestStoreService } from "./boundary.ts"; import { abandonSharedFuture, BUFFER_MAX_LENGTH, type ByteWindow, type ComponentInstanceState, - consumeSchedulerFailure, CopyResult, type DirectBuffer, type DirectOutcome, @@ -36,8 +31,6 @@ import { SharedFutureImpl, SharedStreamImpl, type Store, - storeQuiescent as quiescent, - unwrapSchedulerFailure, } from "../task/mod.ts"; /** @@ -210,7 +203,6 @@ class HostActivity { #closed = false; /** No retained end; revivable via `rearm()`. */ #disarmed = false; - #pumping = false; bind(store: Store): void { if (this.#store !== null || this.#closed) return; @@ -236,79 +228,17 @@ class HostActivity { } r?.(); this.#arm(); + if (this.#store !== null) requestStoreService(this.#store); } /** - * Drain settled activations and ready threads synchronously, then drive - * asynchronous work if the store is not quiescent. This gives host - * operations progress between export calls, including guest dependencies - * on Promise-returning host imports. - * - * Record synchronous failures as well as throwing them: retirement may - * already have settled the host operation's Promise, whose executor would - * then discard the throw. The next driver can still report hostFailure. + * Request the store's coalesced drain. Host retention is liveness evidence, + * not runnable work and therefore never owns an independent pump. */ pump(): void { const store = this.#store; if (store === null) return; - // Settled activation tails gate `tick` (Store.settled); a driver that - // never services them wedges the store — this loop runs BETWEEN export - // calls, when no driveAsync exists to do it. An attributed failure is one - // completed step; keep draining healthy ready siblings. - for (;;) { - try { - const serviced = store.serviceSettled(); - const ticked = store.tick(); - if (!serviced && !ticked) break; - } catch (e) { - if (consumeSchedulerFailure(store, e)) continue; - const cause = unwrapSchedulerFailure(e); - store.hostFailure ??= cause; - throw cause; - } - } - if (this.#pumping) return; - // Retention alone is not work to drive; leave the host Promise pending. - if (quiescent(store)) return; - this.#pumping = true; - void this.#pumpAsync(store); - } - - async #pumpAsync(store: Store): Promise { - try { - // Yield to an existing driver. This is cooperative: another may enter - // while we await, so done() checks depth again. Resume sites recheck - // awaiting membership and Promise identity before consuming a result - // (boundary.ts storeDriverDepth), preventing double resumption. - while (!quiescent(store)) { - if (storeDriverDepth(store) > 0) { - await whenStoreDriverIdle(store); - continue; - } - await driveStoreAsync( - store, - // Stop on quiescence, before an idle deadlock verdict, or when - // another driver enters. Our own depth is 1 inside this loop. - // The caller awaits its operation, not this fallback pump. - () => - store.pendingHostCalls.size === 0 || - quiescent(store) || - storeDriverDepth(store) > 1, - "host stream/future activity", - ); - } - } catch (e) { - if (consumeSchedulerFailure(store, e)) return; - // Nothing is awaiting this pump, so park the failure where the next - // driving loop will surface it (same channel as a host-import - // rejection). - store.hostFailure ??= unwrapSchedulerFailure(e); - } finally { - this.#pumping = false; - } - // Guest progress may have settled a task without settling anything in - // another driver's pendingHostCalls race. Wake it to recheck done(). - this.notify(); + requestStoreService(store); } /** No further host activity is possible on this stream. */ @@ -321,6 +251,7 @@ class HostActivity { this.#store.pendingHostCalls.delete(p); } r?.(); + this.#store?.requestService(); } /** @@ -336,6 +267,7 @@ class HostActivity { this.#store.pendingHostCalls.delete(p); } r?.(); + this.#store?.requestService(); } /** diff --git a/runtime/src/intrinsics/async_builtins.ts b/runtime/src/intrinsics/async_builtins.ts index 8aa1f8f3..8349ee63 100644 --- a/runtime/src/intrinsics/async_builtins.ts +++ b/runtime/src/intrinsics/async_builtins.ts @@ -171,6 +171,7 @@ export function createBackpressureDec(inst: ComponentInstanceState): CoreFn { ); inst.backpressure -= 1; trapIf(inst.backpressure < 0, "backpressure counter underflow"); + inst.store.requestService(); }; } diff --git a/runtime/src/jspi/bridge.ts b/runtime/src/jspi/bridge.ts index 979e5f46..52205cb5 100644 --- a/runtime/src/jspi/bridge.ts +++ b/runtime/src/jspi/bridge.ts @@ -451,6 +451,7 @@ export class SuspensionPoint implements SchedulableThread { this.#fail = rej; }); store.startWaiting(this); + if (this.ready()) store.requestService(); } waiting(): boolean { @@ -681,6 +682,12 @@ export function blockCurrentActivation(input: { ) { input.task?.controlReturned?.(owner); } + // The owner is now represented by a genuine SuspensionPoint rather than + // an engine-only entry hop. Same-instance admission observes completed + // scheduler state changes, not coalesced requests, so publish this park + // transition before asking the coordinator to service newly exposed work. + input.store.noteServiceProgress(); + input.store.requestService(); }); return point.promise; } diff --git a/runtime/src/task/mod.ts b/runtime/src/task/mod.ts index 47db6f96..d28fc120 100644 --- a/runtime/src/task/mod.ts +++ b/runtime/src/task/mod.ts @@ -255,6 +255,7 @@ export class Task { "exit_implicit_thread without holding the exclusive thread", ); this.inst.exclusiveThread = null; + this.inst.store.requestService(); } } diff --git a/runtime/src/task/scheduler.ts b/runtime/src/task/scheduler.ts index e05aa39b..7cf4a5af 100644 --- a/runtime/src/task/scheduler.ts +++ b/runtime/src/task/scheduler.ts @@ -507,6 +507,18 @@ export function maybeCurrentThread(): CurrentThreadLike | undefined { return resolveAmbient(); } +/** Whether guest code is executing on the current JavaScript stack. Engine + * continuation claims deliberately do not count: they describe attribution + * across a hop, not live execution. Ordinary store service must never enter a + * second guest activation while this is true (contracts/intrinsics.md:118-126). */ +export function guestActivationLive(store?: Store): boolean { + if (store === undefined) { + return threadStack.length > 0 || entryStack.length > 0; + } + return threadStack.some((t) => t?.task?.inst?.store === store) || + entryStack.some((t) => t?.task?.inst?.store === store); +} + /** * Resolve using the declaring instance: a matching top synchronous bracket, * then the newest matching activation claim, then the unscoped fallback. @@ -644,12 +656,42 @@ export class Store { */ hostFailure: unknown = undefined; + /** Installed by exec/boundary.ts. Scheduler transitions only announce that + * work may now be runnable; one store coordinator owns asynchronous drain. */ + serviceRequested: (() => void) | null = null; + #serviceProgressRevision = 0; + #serviceProgressObservers = new Set<() => void>(); + + requestService(): void { + this.serviceRequested?.(); + } + + /** Record completed scheduler state change separately from a request to run + * the coordinator. Admission waiters use this generation so a coalesced + * request cannot wake them before the corresponding tail has executed. */ + noteServiceProgress(): void { + this.#serviceProgressRevision++; + const observers = [...this.#serviceProgressObservers]; + this.#serviceProgressObservers.clear(); + for (const resolve of observers) resolve(); + } + + serviceProgressRevision(): number { + return this.#serviceProgressRevision; + } + + waitForServiceProgress(after: number): Promise { + if (this.#serviceProgressRevision !== after) return Promise.resolve(); + return new Promise((resolve) => + this.#serviceProgressObservers.add(resolve) + ); + } + /** * Per-store scheduling gate, not an ambient source. Several resumptions * can be outstanding; `tick` waits until all entries are released, without * blocking independent stores. An entry ends when its activation runs or - * parks (`consumePendingIfRunning`), finishes (`noteAwaiting`), or the - * driver removes its own speculative entry. + * parks (`consumePendingIfRunning`) or finishes (`noteAwaiting`). */ readonly pendingResumptions: Set = new Set(); @@ -671,7 +713,7 @@ export class Store { /** Drop exactly `t` (the driver's own speculative entry). */ removePendingResumption(t: unknown): void { if (AMBIENT_TRACE) traceAmbient("pending-", t); - this.pendingResumptions.delete(t); + if (this.pendingResumptions.delete(t)) this.requestService(); } /** @@ -680,7 +722,14 @@ export class Store { */ consumePendingIfRunning(): void { const a = activationOf(); - if (a !== null && a !== undefined) this.pendingResumptions.delete(a); + if ( + a !== null && a !== undefined && this.pendingResumptions.delete(a) + ) { + // This request is queued while the activation is still live. The + // coordinator's live-execution guard delays service until the canonical + // park/return boundary has actually unwound. + this.requestService(); + } } /** @@ -690,17 +739,23 @@ export class Store { // deno-lint-ignore no-explicit-any releasePendingOf(t: any): void { releaseActivationAmbient(t); - this.pendingResumptions.delete(t); + let released = this.pendingResumptions.delete(t); const implicit = (t as { task?: { implicitThread?: unknown } })?.task ?.implicitThread; if (implicit !== undefined && implicit !== null) { - this.pendingResumptions.delete(implicit); + released = this.pendingResumptions.delete(implicit) || released; } + if (released) this.requestService(); } startWaiting(t: SchedulableThread): void { assert_(!this.waiting.includes(t), "thread already in the waiting list"); this.waiting.push(t); + // A newly registered SuspensionPoint can turn its owner's `awaiting` + // entry from an implementation-only hop into a genuine canonical park. + // Notify even when `t` is not ready: the transition removes the hop + // barrier and can expose already-queued tails or unrelated ready work. + this.requestService(); } stopWaiting(t: SchedulableThread): void { @@ -752,58 +807,60 @@ export class Store { (value) => { this.settled.push({ t, value, failure: undefined }); this.releasePendingOf(t); + this.requestService(); }, (e) => { this.settled.push({ t, value: undefined, failure: { error: e } }); this.releasePendingOf(t); + this.requestService(); }, ); } /** - * Dispatch tails in queue order without an entry-lock test. Every driver - * must service this queue before and between ticks; exceptions propagate - * to that driver. Stale entries are discarded, and `resumeWith` retires - * poisoned-instance tails without running their bodies. + * Dispatch at most one live tail in queue order. Stale entries are discarded + * until a live tail is found or the queue is empty. Autonomous service uses + * this step form so it can revalidate entry-hop permission after each guest + * activation. */ - serviceSettled(): boolean { - let did = false; - // Rescan from the head after every dispatch: a dispatched tail runs guest - // code synchronously, which can change lock/poison state and can re-enter - // `serviceSettled` (mutating the queue under us). - scan: for (;;) { - for (let i = 0; i < this.settled.length; i++) { - const s = this.settled[i]; - // Another driver already resumed this thread. - if (!this.awaiting.has(s.t)) { - this.settled.splice(i, 1); - continue scan; - } - this.settled.splice(i, 1); - const t = s.t as { - resumeWith(v: unknown, f?: { error: unknown }): void; - task?: { - inst?: object; - failureOwner?: unknown; - fail?(cause: unknown): boolean; - }; + serviceSettledStep(): boolean { + while (this.settled.length > 0) { + const s = this.settled.shift()!; + // Another driver already resumed this thread. + if (!this.awaiting.has(s.t)) continue; + const t = s.t as { + resumeWith(v: unknown, f?: { error: unknown }): void; + task?: { + inst?: object; + failureOwner?: unknown; + fail?(cause: unknown): boolean; }; - const origin = (t.task?.failureOwner ?? t.task) as - | { onFailure?: unknown } - | undefined; - try { - t.resumeWith(s.value, s.failure); - } catch (e) { - // Autonomous tails belong to their originating task, not whichever - // sibling happened to service the store queue (#357). - throw originatedFailure(origin, e); - } - did = true; - continue scan; + }; + const origin = (t.task?.failureOwner ?? t.task) as + | { onFailure?: unknown } + | undefined; + try { + t.resumeWith(s.value, s.failure); + } catch (e) { + // Autonomous tails belong to their originating task, not whichever + // sibling happened to service the store queue (#357). + throw originatedFailure(origin, e); + } finally { + // The awaiting identity may have been removed, re-parked, or retired. + // Notify state observers after that transition, not when service was + // merely requested. + this.noteServiceProgress(); } - // A full scan found nothing stale and nothing serviceable. - return did; + return true; } + return false; + } + + /** Explicit callers that need all currently queued tails drain stepwise. */ + serviceSettled(): boolean { + let did = false; + while (this.serviceSettledStep()) did = true; + return did; } /** @@ -906,13 +963,13 @@ export function hasRealHostCall(store: Store): boolean { return false; } -/** Every outstanding host call that is real work (not an activity arm). */ -export function realHostCalls(store: Store): Promise[] { - const out: Promise[] = []; +/** Host retention suppresses deadlock but is not runnable work or a host + * dependency for the coordinator to poll. */ +export function hasHostRetention(store: Store): boolean { for (const p of store.pendingHostCalls) { - if (!hostActivityArms.has(p)) out.push(p); + if (hostActivityArms.has(p)) return true; } - return out; + return false; } /** diff --git a/runtime/src/task/streams.ts b/runtime/src/task/streams.ts index 67acf92f..b6bdc801 100644 --- a/runtime/src/task/streams.ts +++ b/runtime/src/task/streams.ts @@ -338,7 +338,7 @@ export class SharedStreamImpl implements SharedBase { * between export calls (see exec/host_streams.ts `HostActivity.pump`); a * purely guest-to-guest stream never reads it. */ - boundStore: unknown = null; + boundStore: object | null = null; dropped = false; pendingInst: unknown = null; @@ -589,7 +589,7 @@ export class SharedFutureImpl implements SharedBase { * between export calls (see exec/host_streams.ts `HostActivity.pump`); a * purely guest-to-guest stream never reads it. */ - boundStore: unknown = null; + boundStore: object | null = null; dropped = false; /** diff --git a/runtime/src/task/thread.ts b/runtime/src/task/thread.ts index a93b5d2b..ed56b146 100644 --- a/runtime/src/task/thread.ts +++ b/runtime/src/task/thread.ts @@ -112,6 +112,7 @@ export class Thread implements SchedulableThread { resumeLater(): void { assert_(this.suspended(), "resume_later on a non-suspended thread"); this.#startWaiting(() => true); + this.#store.requestService(); } /** Pending `awaitValue` promise, if this thread is parked on one. */ @@ -229,6 +230,7 @@ export class Thread implements SchedulableThread { } else { this.#state = "suspended"; this.#startWaiting(req.readyFunc); + if (this.ready()) this.#store.requestService(); } this.task.controlReturned?.(this); } diff --git a/runtime/tests/conventions/future-gated.wasm b/runtime/tests/conventions/future-gated.wasm new file mode 100644 index 0000000000000000000000000000000000000000..56d9dffb95ddbbf86bc0f71a5b47c981b2d480c5 GIT binary patch literal 1143 zcmbVM!EVz)5PdVd_BvUGno2F)a7hjoQVs2y9NJv!3kUwB zU%5+rIGqZ2r&S>5)@_Eq%!D#Zdq&~za0Ffe)F{BD#Fktm0FFdq56ITa7 zqG$(khromYN(}9SW3>_M>1QEEFZvxx*+h4+B*pY2S{+FPa}K&c=c=MgtSVSIC;3!W z$4EB#b5ftjQnZ@*TjA9+pO3_iZfDn}5AnG#!O&l)*Apo8X1-j^Cc~?REYbwJZD8@{ zwX_sR9B~}QJGvqV`{-v5B3XkEka7f8jAN3Jzss>6TeU)m*edTmHvxmsoq2A8m^r=4T zBaK|*UcF`X$t5Yb*?n41+_KWeb2vB(K;aNz!{gW<5?4Z(rnJX_Uua#r&u~0WV}~s( z!BtJ6^o&AD8Pz3;E0r`Q{pDQCw)jPjQkc%@ov9~8^?bOQjtaBU^<<*Wxx{Nz;Sbs{ zqVM^5;YokVOP|O%FQ+5V<|bc`r}?O>z_U9FbR5siyj;xfV~p9?NNSRyKwGQ4V-pyM z<(*3EkqX6{H2D)#Qp;X7`fBCXs*7uJsYY9Rvjs$JJs<`Vw#h_v62!kN*NC{$k{ByB zC31yeI5tT@9s$d^ClLX(B1>X%J758v5M2epOBDbs?X?#$g#M{8v69tQvC_sWyQfee zEHSKjr8u+{B&8KkE~mxKbmAxFFx+)|FVO#;A6YzQgvWwJ2Nv7ngolI*svnlqKY4H2 A!2kdN literal 0 HcmV?d00001 diff --git a/runtime/tests/conventions/future-gated.wat b/runtime/tests/conventions/future-gated.wat new file mode 100644 index 00000000..f14a0790 --- /dev/null +++ b/runtime/tests/conventions/future-gated.wat @@ -0,0 +1,85 @@ +;; A same-store future whose producing call cannot reach task.return until a +;; separate export opens an explicit guest gate. This makes the deferred-handle +;; window observable without host timing assumptions. +;; Regenerate: wasm-tools parse future-gated.wat -o future-gated.wasm +(component + (type $F (future u32)) + (core module $Mem (memory (export "mem") 1)) + (core instance $mem (instantiate $Mem)) + (canon future.new $F (core func $new)) + (canon future.read $F async (memory $mem "mem") (core func $read)) + (canon future.write $F async (memory $mem "mem") (core func $write)) + (canon waitable-set.new (core func $set)) + (canon waitable.join (core func $join)) + (canon task.return (result $F) (core func $return-future)) + (canon task.return (result u32) (memory $mem "mem") (core func $return-u32)) + (core module $M + (import "" "mem" (memory 1)) + (import "" "new" (func $new (result i64))) + (import "" "read" (func $read (param i32 i32) (result i32))) + (import "" "write" (func $write (param i32 i32) (result i32))) + (import "" "set" (func $set (result i32))) + (import "" "join" (func $join (param i32 i32))) + (import "" "return-future" (func $return-future (param i32))) + (import "" "return-u32" (func $return-u32 (param i32))) + (global $value-rx (mut i32) (i32.const 0)) + (global $value-tx (mut i32) (i32.const 0)) + (global $gate-tx (mut i32) (i32.const 0)) + (global $make-set (mut i32) (i32.const 0)) + (func (export "make") (result i32) + (local $value i64) (local $gate i64) + (local.set $value (call $new)) + (local.set $gate (call $new)) + (global.set $value-rx (i32.wrap_i64 (local.get $value))) + (global.set $value-tx (i32.wrap_i64 (i64.shr_u (local.get $value) (i64.const 32)))) + (global.set $gate-tx (i32.wrap_i64 (i64.shr_u (local.get $gate) (i64.const 32)))) + (if (i32.ne (call $read (i32.wrap_i64 (local.get $gate)) (i32.const 0)) (i32.const -1)) + (then unreachable)) + (global.set $make-set (call $set)) + (call $join (i32.wrap_i64 (local.get $gate)) (global.get $make-set)) + (i32.or (i32.const 2) (i32.shl (global.get $make-set) (i32.const 4)))) + (func (export "make-cb") (param $code i32) (param i32) (param $payload i32) (result i32) + (if (i32.ne (local.get $payload) (i32.const 0)) (then unreachable)) + (if (i32.eq (local.get $code) (i32.const 4)) + (then + (call $return-future (global.get $value-rx)) + (i32.store (i32.const 8) (i32.const 42)) + (if (i32.ne (call $write (global.get $value-tx) (i32.const 8)) (i32.const -1)) + (then unreachable)) + (call $join (global.get $value-tx) (global.get $make-set)) + (return (i32.or (i32.const 2) (i32.shl (global.get $make-set) (i32.const 4)))))) + (i32.const 0)) + (func (export "release") (result i32) + (i32.store (i32.const 0) (i32.const 1)) + (if (i32.ne (call $write (global.get $gate-tx) (i32.const 0)) (i32.const 0)) + (then unreachable)) + (i32.const 2)) + (func (export "double") (param $rx i32) (result i32) (local $set i32) (local $status i32) + (local.set $status (call $read (local.get $rx) (i32.const 16))) + (if (i32.eq (local.get $status) (i32.const 0)) + (then + (call $return-u32 (i32.mul (i32.load (i32.const 16)) (i32.const 2))) + (return (i32.const 0)))) + (if (i32.ne (local.get $status) (i32.const -1)) (then unreachable)) + (local.set $set (call $set)) + (call $join (local.get $rx) (local.get $set)) + (i32.or (i32.const 2) (i32.shl (local.get $set) (i32.const 4)))) + (func (export "double-cb") (param $code i32) (param i32) (param $payload i32) (result i32) + (if (i32.ne (local.get $code) (i32.const 4)) (then unreachable)) + (if (i32.ne (local.get $payload) (i32.const 0)) (then unreachable)) + (call $return-u32 (i32.mul (i32.load (i32.const 16)) (i32.const 2))) + (i32.const 0))) + (core instance $m (instantiate $M (with "" (instance + (export "mem" (memory $mem "mem")) + (export "new" (func $new)) + (export "read" (func $read)) + (export "write" (func $write)) + (export "set" (func $set)) + (export "join" (func $join)) + (export "return-future" (func $return-future)) + (export "return-u32" (func $return-u32)))))) + (func (export "make") async (result $F) + (canon lift (core func $m "make") async (callback (core func $m "make-cb")))) + (func (export "release") (result u32) (canon lift (core func $m "release"))) + (func (export "double") async (param "f" $F) (result u32) + (canon lift (core func $m "double") async (memory $mem "mem") (callback (core func $m "double-cb"))))) diff --git a/runtime/tests/conventions/lowering_test.ts b/runtime/tests/conventions/lowering_test.ts index 819d92e1..1e7690c5 100644 --- a/runtime/tests/conventions/lowering_test.ts +++ b/runtime/tests/conventions/lowering_test.ts @@ -10,7 +10,7 @@ // completion. The transcripts record the values that crossed, never the order // in which independent tasks got scheduled. -import { guest, haveFixture, instantiateFixture } from "./harness.ts"; +import { guest, haveFixture, instantiateFixture, local } from "./harness.ts"; import { transcript } from "./support.ts"; import { asyncIterable, classify, readable, thenable } from "./probe.ts"; @@ -60,29 +60,46 @@ Deno.test({ }, }); -const futureUserReady = await haveFixture(guest("future-user")); +const futureUserReady = await haveFixture(local("future-gated")); Deno.test({ name: "conventions/b: a Future HANDLE is a lowering source (same store)", ignore: !futureUserReady, fn: async () => { await transcript("b-future-handle-source", async (t) => { - const c = await instantiateFixture(guest("future-user")); + const c = await instantiateFixture(local("future-gated")); // An export whose WIT result is `future` returns the handle // EAGERLY — call without awaiting to hold it. - const f = c.exports.makeFuture(41) as unknown; + const f = c.exports.make() as unknown; t.note("export-result", { classified: classify(f), value: f }); // handle disposal: such a handle is DEFERRED — its host end materializes when the // producing call completes. Lowering it before then is refused, loudly. - await t.attempt("lower-while-in-flight", () => c.exports.doubleFuture(f)); - - // Drive the instance to quiescence with an unrelated single task, so the - // producing call has completed. (Deterministic: the probe task's own - // completion is what is awaited, and the producer needs no further host - // action.) - await t.attempt("unrelated-call", () => c.exports.doubleFuture(1)); - await t.attempt("lower-after-settled", () => c.exports.doubleFuture(f)); + await t.attempt("lower-while-in-flight", () => c.exports.double(f)); + + await t.attempt("unrelated-call", () => c.exports.release()); + await t.attempt("lower-after-settled", async () => { + // release() makes the producer runnable; it does not mean the Future + // handle's host end has already been published. Retry only the exact + // deferred-handle refusal, which leaves the source unconsumed. + for (let attempt = 0; attempt < 20; attempt++) { + try { + return await c.exports.double(f); + } catch (e) { + if ( + !(e instanceof TypeError) || + e.message !== + "this Future is still in flight and cannot be passed to a guest yet" + ) { + throw e; + } + } + await new Promise((resolve) => setTimeout(resolve, 0)); + } + throw new Error( + "Future handle remained in flight after 20 platform turns", + ); + }); }); }, }); diff --git a/runtime/tests/cross_store_driver_test.ts b/runtime/tests/cross_store_driver_test.ts index 91c389c5..bc3628c6 100644 --- a/runtime/tests/cross_store_driver_test.ts +++ b/runtime/tests/cross_store_driver_test.ts @@ -1,28 +1,6 @@ -// Concurrent drivers on independent stores (issues #210, #158 mechanism B). -// -// The driver's speculative resume entry (exec/boundary.ts, `Promise.race` -// over the parked threads) is held for the entire duration of a guest's wait -// on a slow host import — the completely ordinary suspended-guest shape. -// The gate is `Store.pendingResumptions`, PER STORE: activations never cross -// stores, so A's pending resumption is none of B's business. Were it shared, -// every `driveAsync` loop would yield at its top while ANY store held an -// entry, against a bounded hop counter: -// -// assert_(claimHops < 10_000, "driveAsync: a resumed-activation claim was -// never released ...") -// -// so an idle, completely unrelated store B's `driveStoreAsync` would die at -// 10,000 hops in ~311ms while store A merely dwelt on its import — an -// internal AssertionError naming neither component (issue #210). -// -// This test pins the requirement: B's driver must return promptly. The -// control below pins that the gate still gates — A's OWN driver refuses to -// tick past A's own pending entry. -// -// Scaffolding style: settlement_pump_test.ts (fake inst/threads, real Store, -// real driveStoreAsync). +// Independent stores have independent event-driven coordinators (#210). -import { driveStoreAsync } from "../src/exec/mod.ts"; +import { driveStoreAsync, requestStoreService } from "../src/exec/mod.ts"; import { Store } from "../src/task/mod.ts"; function assert(cond: boolean, msg: string): asserts cond { @@ -51,35 +29,15 @@ function awaitingThread(store: Store, p: Promise) { return t; } -function hostImport(store: Store, settle: Promise): void { - const p: Promise = settle.then(() => { - store.pendingHostCalls.delete(p); - }); - store.pendingHostCalls.add(p); -} - -Deno.test("a store dwelling on a slow import does not stall another store's driver (#210)", async () => { +Deno.test("a pending resumption in one store does not stall another store (#210)", async () => { const storeA = new Store(); const storeB = new Store(); let settleAThread!: (v: unknown) => void; - let settleAHost!: (v: unknown) => void; const aThreadP = new Promise((r) => (settleAThread = r)); - const aHostP = new Promise((r) => (settleAHost = r)); - let aDone = false; try { - // Store A: guest suspended on a slow host import. - awaitingThread(storeA, aThreadP); - hostImport(storeA, aHostP); - const aDriver = driveStoreAsync(storeA, () => aDone, "A: dweller") - .catch((e) => e); - - // Wait until A's driver has taken its speculative entry (it needs a few - // turns to reach the race). - const t0 = Date.now(); - while (!storeA.hasPendingResumptions()) { - assert(Date.now() - t0 < 2000, "A's driver never took its entry"); - await new Promise((r) => setTimeout(r, 1)); - } + const thread = awaitingThread(storeA, aThreadP); + storeA.addPendingResumption(thread); + requestStoreService(storeA); // Store B: a COMPLETELY IDLE unrelated store; its driver has nothing to do // (done() is immediately true). It must not consult A's gate at all. @@ -103,19 +61,14 @@ Deno.test("a store dwelling on a slow import does not stall another store's driv `B's driver returned in ${elapsed}ms, expected < 2s`, ); - // CONTROL: the gate still gates its OWN store. A's driver is still in its - // race, holding A's entry, and A's own `tick` refuses. + // CONTROL: the real resumption claim still gates its own store. assert(storeA.hasPendingResumptions(), "A's entry is still pending"); assert(storeA.tick() === false, "A's own gate still refuses to schedule"); - // Cleanup: let A's driver exit. - aDone = true; settleAThread(0); - settleAHost(0); - await aDriver; + await Promise.resolve(); } finally { settleAThread(0); - settleAHost(0); storeA.pendingResumptions.clear(); storeB.pendingResumptions.clear(); } diff --git a/runtime/tests/driver_stale_settled_test.ts b/runtime/tests/driver_stale_settled_test.ts index 4d53b8ed..168385b5 100644 --- a/runtime/tests/driver_stale_settled_test.ts +++ b/runtime/tests/driver_stale_settled_test.ts @@ -18,7 +18,7 @@ // assertion below is exactly that — the value arrives once. import { assertEq } from "./support/asserts.ts"; -import { driveStoreAsync, registerHostCall } from "../src/exec/mod.ts"; +import { requestStoreService } from "../src/exec/mod.ts"; import { Store } from "../src/task/mod.ts"; function assert(cond: boolean, msg: string): asserts cond { @@ -66,18 +66,8 @@ Deno.test({ t.awaiting = p1; store.noteAwaiting(t, p1); - // A real, never-settling host call: it keeps `pendingHostCalls` non-empty - // so the driver takes the P5 servicing race (the winner path) instead of - // the deadlock probe. - const stuck = new Promise(() => {}); - registerHostCall(store, stuck); - - let finished = false; - const driving = driveStoreAsync(store, () => finished, "F1 driver"); - - // Let the driver reach the race. - for (let i = 0; i < 10; i++) await Promise.resolve(); - await new Promise((r) => setTimeout(r, 0)); + requestStoreService(store); + await Promise.resolve(); assertEq(t.received.length, 0); settleA("A"); @@ -97,16 +87,6 @@ Deno.test({ // NEW park. assertEq(store.settled.length, 0); - // Teardown: let the driver exit and leave `pendingHostCalls` empty so the - // settlement pump inherits nothing. - finished = true; - store.pendingHostCalls.delete(stuck); - const wake: Promise = Promise.resolve().then(() => { - store.pendingHostCalls.delete(wake); - }); - registerHostCall(store, wake); - await driving; - assertEq(store.pendingHostCalls.size, 0); assertEq(store.hostFailure, undefined); }, }); diff --git a/runtime/tests/driver_trap_exit_liveness_test.ts b/runtime/tests/driver_trap_exit_liveness_test.ts index 163f65f4..e9a13ebe 100644 --- a/runtime/tests/driver_trap_exit_liveness_test.ts +++ b/runtime/tests/driver_trap_exit_liveness_test.ts @@ -18,7 +18,11 @@ // exit; run_tests.py `lift_and_run` drains the whole store either way. import { assertEq } from "./support/asserts.ts"; -import { driveStoreAsync, registerHostCall } from "../src/exec/mod.ts"; +import { + driveStoreAsync, + registerHostCall, + requestStoreService, +} from "../src/exec/mod.ts"; import { createDtorEntry } from "../src/exec/boundary.ts"; import { ComponentInstanceState, Store } from "../src/task/mod.ts"; import { Trap } from "../src/cabi/mod.ts"; @@ -95,28 +99,32 @@ Deno.test({ const impl = new ComponentInstanceState(1, store); const lifted = createDtorEntry({ dtor: () => undefined, instance: impl }); - let thrown: unknown; - try { - lifted(0); - } catch (e) { - thrown = e; + assertEq(lifted(0), undefined); + for (let i = 0; i < 20 && store.hostFailure === undefined; i++) { + await Promise.resolve(); } + const thrown: unknown = store.hostFailure; assert(thrown instanceof Trap, `expected the sibling trap, got ${thrown}`); assertEq(sibling.resumed, 1); assertEq(store.pendingHostCalls.size, 1); - // The sibling's host call answers. Its continuation readies the sibling - // and deletes its own entry — readying is all it does; somebody has to - // tick the store, and after a throwing exit that is the settlement pump. + // The sibling's host call answers. The raw background trap remains parked + // for the next call, so service cannot resume healthy work until that + // compatibility channel has been observed. settle(); - for (let i = 0; i < 20 && sibling.resumed < 2; i++) { - await new Promise((r) => setTimeout(r, 0)); + await Promise.resolve(); + assertEq(sibling.resumed, 1); + try { + await driveStoreAsync(store, () => false, "observe raw trap"); + } catch (e) { + assertEq(e, thrown); } + requestStoreService(store); + for (let i = 0; i < 20 && sibling.resumed < 2; i++) await Promise.resolve(); assert( sibling.resumed >= 2, - "the healthy sibling was never resumed: `drive` threw without arming " + - "the settlement pump, so its in-flight host call had no keeper", + "healthy sibling did not resume after the parked raw failure was observed", ); assertEq(store.pendingHostCalls.size, 0); }, @@ -200,6 +208,19 @@ Deno.test({ store.startWaiting(sibling as any); // deno-lint-ignore no-explicit-any store.startWaiting(trapper as any); + // This synthetic promise park has no real SuspensionPoint owner. Mark its + // boundary as returned so the coordinator may service unrelated ready work; + // real JSPI parks acquire this evidence through blockCurrentActivation. + store.startWaiting( + { + owner: sibling, + boundaryReturned: true, + task: sibling.task, + ready: () => false, + waiting: () => true, + resume: () => {}, + } as unknown as import("../src/task/mod.ts").SchedulableThread, + ); const driving = driveStoreAsync(store, () => false, "export 'trapping'"); let thrown: unknown; diff --git a/runtime/tests/embedder/streams_test.ts b/runtime/tests/embedder/streams_test.ts index ebbe82b4..6cfffcb8 100644 --- a/runtime/tests/embedder/streams_test.ts +++ b/runtime/tests/embedder/streams_test.ts @@ -21,6 +21,7 @@ import { lowerStream, } from "../../src/cabi/async_values.ts"; import { + isInstancePoisoned, ReadableStreamEnd, SharedFutureImpl, SharedStreamImpl, @@ -522,6 +523,14 @@ const disposalFixture = "runtime/tests/embedder/future-disposal.wasm"; const disposalReady = await haveFixture(disposalFixture); const turn = () => new Promise((resolve) => setTimeout(resolve, 0)); +async function waitFor( + predicate: () => boolean, + message: string, +): Promise { + for (let i = 0; i < 100 && !predicate(); i++) await turn(); + assertEq(predicate(), true, message); +} + for (const jspi of [undefined, false, true]) { for (const deferred of [false, true]) { Deno.test({ @@ -531,10 +540,15 @@ for (const jspi of [undefined, false, true]) { for (const method of ["drop", Symbol.dispose] as const) { const c = await instantiateFixture(disposalFixture, {}, { jspi }); const f = c.exports.run() as Future; + // The fixture calls task.return before parking its writer. Dropping + // the reader wakes that writer and its callback deliberately traps. if (!deferred) await turn(); assertEq(f[method](), undefined); assertEq(f[method](), undefined); - await turn(); // Deno rejects any unhandled derived disposal promise. + await waitFor( + () => isInstancePoisoned(c.handle.componentInstances[0]), + "reader drop must drive the guest callback to its deliberate trap", + ); const error = await caught(() => Promise.resolve(f)); assertEq(error instanceof PeerTrappedError, true, String(error)); assertEq(await caught(() => Promise.resolve(f)), error); diff --git a/runtime/tests/event_driven_drain_test.ts b/runtime/tests/event_driven_drain_test.ts new file mode 100644 index 00000000..a76b6b2a --- /dev/null +++ b/runtime/tests/event_driven_drain_test.ts @@ -0,0 +1,435 @@ +import { + createLiftedFunction, + driveStoreAsync, + newStats, + requestStoreService, + type ResolvedOptions, +} from "../src/exec/mod.ts"; +import { + ComponentInstanceState, + currentThread, + markHostActivityArm, + Store, + Thread, +} from "../src/task/mod.ts"; +import type { FuncType } from "../src/cabi/types.ts"; +import { assert, assertEquals } from "./jspi/asserts.ts"; +import { isSupported } from "../src/jspi/mechanics.ts"; +import { instantiateActivation } from "./jspi/support.ts"; + +class YieldingThread { + resumed = 0; + stop = false; + readonly task = { inst: {} }; + + ready(): boolean { + return !this.stop; + } + + waiting(): boolean { + return !this.stop; + } + + resume(): void { + this.resumed++; + } +} + +function awaitingThread(store: Store, promise: Promise) { + const t = { + awaiting: promise as Promise | null, + task: { inst: {} }, + resumeWith() { + t.awaiting = null; + store.awaiting.delete(t); + }, + }; + store.noteAwaiting(t, promise); + return t; +} + +Deno.test("event-driven drain yields to timers during sustained runnable work", async () => { + const store = new Store(); + const thread = new YieldingThread(); + store.startWaiting(thread); + + let timerRan = false; + const timer = new Promise((resolve) => { + setTimeout(() => { + timerRan = true; + thread.stop = true; + resolve(); + }, 0); + }); + + requestStoreService(store); + await timer; + await Promise.resolve(); + + assert(timerRan, "the drain starved the platform timer queue"); + assert( + thread.resumed >= 8, + `expected one bounded work quantum, got ${thread.resumed}`, + ); + assertEquals(store.hostFailure, undefined); +}); + +Deno.test("callback ABI lift yields to a timer before task.return", async () => { + const store = new Store(); + const inst = new ComponentInstanceState(0, store); + const ft: FuncType = { params: [], results: [], async: true }; + let stopYielding = false; + let callbackCalls = 0; + const run = createLiftedFunction({ + name: "callback-fairness", + ft, + opts: { + stringEncoding: "utf8", + memory: null, + realloc: null, + postReturn: null, + callback: () => + (() => { + callbackCalls++; + if (!stopYielding) return 1; + const task = (currentThread() as unknown as { task: never }).task; + (task as unknown as { return_(result: never[]): void }).return_([]); + return 0; + }) as never, + async: true, + cancellable: false, + coreType: { params: [], results: ["i32"] }, + instance: inst, + }, + core: () => 1, + stats: newStats(), + }); + + setTimeout(() => stopYielding = true, 0); + const result = run(); + assert( + result instanceof Promise, + "fairness handoff must return a Promise before the timer runs", + ); + await result; + + assert(stopYielding, "callback loop completed before the timer ran"); + assert( + callbackCalls >= 8, + `expected a full shared work quantum, got ${callbackCalls} callbacks`, + ); + assertEquals(store.hostFailure, undefined); +}); + +Deno.test("mixed queued and unsettled entry hops do not spin on the settled hop", async () => { + const store = new Store(); + awaitingThread(store, Promise.resolve()); + awaitingThread(store, new Promise(() => {})); + + requestStoreService(store); + let timerRan = false; + await new Promise((resolve) => { + setTimeout(() => { + timerRan = true; + resolve(); + }, 20); + }); + + assert(timerRan, "a settled hop was repeatedly raced and starved timers"); + assertEquals( + store.settled.length, + 1, + "queued hop crossed the unsettled barrier", + ); + assertEquals(store.awaiting.size, 2, "an entry hop was disturbed"); + assertEquals(store.hostFailure, undefined); +}); + +Deno.test("one request drains every immediately settled activation tail", async () => { + const store = new Store(); + const resumed: number[] = []; + for (let i = 0; i < 3; i++) { + const tail = { + awaiting: Promise.resolve(), + task: { inst: {} }, + resumeWith() { + resumed.push(i); + tail.awaiting = null as unknown as Promise; + store.awaiting.delete(tail); + }, + }; + store.noteAwaiting(tail, tail.awaiting); + } + await Promise.resolve(); + + requestStoreService(store); + await Promise.resolve(); + + assertEquals(resumed.join(","), "0,1,2"); + assertEquals(store.settled.length, 0); + assertEquals(store.awaiting.size, 0); +}); + +Deno.test("a pending claim stops queued-tail service until its release", async () => { + const store = new Store(); + let resumes = 0; + const tail = awaitingThread(store, Promise.resolve()); + tail.resumeWith = () => { + resumes++; + tail.awaiting = null; + store.awaiting.delete(tail); + }; + const claim = {}; + await Promise.resolve(); + store.addPendingResumption(claim); + assert( + store.pendingResumptions.has(claim), + "test claim was released before the blocked service request", + ); + + requestStoreService(store); + let timerRan = false; + queueMicrotask(() => store.removePendingResumption(claim)); + await new Promise((resolve) => { + setTimeout(() => { + timerRan = true; + resolve(); + }, 20); + }); + + assert(timerRan, "blocked queued work starved the platform timer"); + assertEquals(resumes, 1, "released tail must execute exactly once"); + assertEquals(store.settled.length, 0); +}); + +Deno.test({ + name: "same-instance JSPI entry waits through more than one tail quantum", + ignore: !isSupported(), + fn: async () => { + const wasm = await instantiateActivation({ + block: new WebAssembly.Suspending((x: number) => x), + }); + const store = new Store(); + const inst = new ComponentInstanceState(0, store); + let tails = 0; + for (let i = 0; i < 9; i++) { + const awaiting = Promise.resolve(); + const tail = { + awaiting: awaiting as Promise | null, + task: { inst }, + resumeWith() { + tails++; + tail.awaiting = null; + store.awaiting.delete(tail); + }, + }; + store.noteAwaiting(tail, awaiting); + } + await Promise.resolve(); + + const run = createLiftedFunction({ + name: "post-hop-entry", + ft: { params: [{ kind: "u32" }], results: [{ kind: "u32" }] }, + opts: { + stringEncoding: "utf8", + memory: null, + realloc: null, + postReturn: null, + callback: null, + async: false, + cancellable: false, + coreType: { params: ["i32"], results: ["i32"] }, + instance: inst, + }, + core: wasm.other, + stats: newStats(), + suspensionMode: "jspi", + }); + + const result = run(42); + assert(result instanceof Promise, "gated JSPI entry must remain async"); + assertEquals(await result, 1042); + assertEquals(tails, 9, "entry resumed before every hop tail completed"); + assertEquals(store.awaiting.size, 0); + }, +}); + +Deno.test("ordinary service stops when the first tail creates an unqueued hop", async () => { + const store = new Store(); + let secondRan = false; + const hop = { + awaiting: new Promise(() => {}), + task: { inst: {} }, + }; + const first = { + awaiting: Promise.resolve(), + task: { inst: {} }, + resumeWith() { + first.awaiting = null as unknown as Promise; + store.awaiting.delete(first); + store.awaiting.add(hop); + }, + }; + const second = { + awaiting: Promise.resolve(), + task: { inst: {} }, + resumeWith() { + secondRan = true; + second.awaiting = null as unknown as Promise; + store.awaiting.delete(second); + }, + }; + store.awaiting.add(first); + store.awaiting.add(second); + store.settled.push( + { t: first, value: undefined, failure: undefined }, + { t: second, value: undefined, failure: undefined }, + ); + + requestStoreService(store); + await Promise.resolve(); + assertEquals(secondRan, false, "second tail crossed the new hop barrier"); + assertEquals(store.settled.length, 1); + + const park = { + owner: hop, + task: hop.task, + ready: () => false, + waiting: () => true, + resume() {}, + }; + store.startWaiting(park); + await Promise.resolve(); + assertEquals( + secondRan, + true, + "second tail stayed blocked after a genuine park", + ); +}); + +Deno.test("an existing unqueued entry hop blocks a global service attempt", async () => { + const store = new Store(); + awaitingThread(store, new Promise(() => {})); + const ready = new YieldingThread(); + store.startWaiting(ready); + + requestStoreService(store); + await Promise.resolve(); + assertEquals(ready.resumed, 0, "ordinary tick crossed an unqueued entry hop"); +}); + +Deno.test("host retention suppresses ordinary and hop-probe deadlock verdicts", async () => { + for (const withHop of [false, true]) { + const store = new Store(); + const arm = new Promise(() => {}); + markHostActivityArm(arm); + store.pendingHostCalls.add(arm); + if (withHop) awaitingThread(store, new Promise(() => {})); + + let settled = false; + driveStoreAsync(store, () => false, `retained ${withHop ? "hop" : "idle"}`) + .then(() => settled = true, () => settled = true); + requestStoreService(store); + await new Promise((resolve) => setTimeout(resolve, 20)); + + assertEquals(settled, false, "retained host capability falsely deadlocked"); + assertEquals(store.hostFailure, undefined); + } +}); + +Deno.test("continuation-driven drains share the fairness budget", async () => { + const store = new Store(); + let resumes = 0; + let timerRan = false; + const thread = { + task: { inst: {} }, + ready: () => !timerRan, + waiting: () => !timerRan, + resume() { + resumes++; + // Force the next quantum through another coordinator invocation rather + // than the same `while (tick())` loop. + store.addPendingResumption(thread); + queueMicrotask(() => store.removePendingResumption(thread)); + }, + }; + store.startWaiting(thread); + let keepAlive = false; + const driving = driveStoreAsync(store, () => keepAlive, "fairness test") + .catch(() => {}); + setTimeout(() => timerRan = true, 0); + requestStoreService(store); + await new Promise((resolve) => setTimeout(resolve, 20)); + + assert(timerRan, "continuation-driven drain invocations starved the timer"); + assert(resumes >= 8, `fairness budget reset between drains at ${resumes}`); + const stoppedAt = resumes; + await new Promise((resolve) => setTimeout(resolve, 10)); + assertEquals(resumes, stoppedAt, "quiescent coordinator kept spinning"); + keepAlive = true; + requestStoreService(store); + await driving; +}); + +Deno.test("finite guest work continues past task.return and a fairness yield", async () => { + const store = new Store(); + const inst = new ComponentInstanceState(0, store); + const ft: FuncType = { params: [], results: [], async: true }; + const opts: ResolvedOptions = { + stringEncoding: "utf8", + memory: null, + realloc: null, + postReturn: null, + callback: () => (() => 0) as never, + async: true, + cancellable: false, + coreType: { params: [], results: ["i32"] }, + instance: inst, + }; + let steps = 0; + let finished = false; + let timerRan = false; + const run = createLiftedFunction({ + name: "post-result-finite-work", + ft, + opts, + core: () => { + const task = (currentThread() as unknown as { task: never }).task; + const background = new Thread( + task, + (function* () { + for (let i = 0; i < 20; i++) { + steps++; + yield { readyFunc: () => true, cancellable: false }; + } + finished = true; + })(), + ); + (task as unknown as { registerThread(t: Thread): void }).registerThread( + background, + ); + background.resume(); + (task as unknown as { return_(result: never[]): void }).return_([]); + return 0; + }, + stats: newStats(), + }); + + setTimeout(() => timerRan = true, 0); + await run(); + assert( + !finished, + "background work finished before public task.return delivery", + ); + while (!finished) await new Promise((resolve) => setTimeout(resolve, 0)); + + assertEquals(steps, 20); + assert(timerRan, "post-result work starved the platform timer"); + const stoppedAt = steps; + await new Promise((resolve) => setTimeout(resolve, 10)); + assertEquals( + steps, + stoppedAt, + "finished background work left resident service", + ); +}); diff --git a/runtime/tests/host_arm_test.ts b/runtime/tests/host_arm_test.ts index bc145c46..96aa9abd 100644 --- a/runtime/tests/host_arm_test.ts +++ b/runtime/tests/host_arm_test.ts @@ -39,6 +39,7 @@ import { } from "../src/exec/mod.ts"; import { dropSharedForTeardown, + hasHostRetention, hasRealHostCall, SharedFutureImpl, SharedStreamImpl, @@ -134,6 +135,7 @@ Deno.test({ !hasRealHostCall(store), "the arm is a retention claim, not outstanding host work", ); + assert(hasHostRetention(store), "the lifted end did not retain capability"); // The identity return: the host's only end goes back to the guest. fireLowered(shared, store); diff --git a/runtime/tests/host_boundary_preparation_test.ts b/runtime/tests/host_boundary_preparation_test.ts index 81b20ccf..ebce3d54 100644 --- a/runtime/tests/host_boundary_preparation_test.ts +++ b/runtime/tests/host_boundary_preparation_test.ts @@ -337,7 +337,7 @@ for (const timing of ["before-listener", "after-listener"] as const) { assertEq( inst.numWaitingToEnter, timing === "before-listener" ? 0 : 1, - "an origin failure consumed by the starting driver retires admission immediately", + "top-level ordinary service retires an already-poisoned admission", ); if (timing === "after-listener") notifyInstancePoisoned(inst, poison); assertEq(await rejected(pending), poison); diff --git a/runtime/tests/host_pump_reentrancy_probe_test.ts b/runtime/tests/host_pump_reentrancy_probe_test.ts index 23f5d642..1db08e80 100644 --- a/runtime/tests/host_pump_reentrancy_probe_test.ts +++ b/runtime/tests/host_pump_reentrancy_probe_test.ts @@ -1,9 +1,8 @@ -// PROBE for issue #298: does `HostActivity.pump()`'s synchronous half run -// `Store.tick()` while a guest activation is live on the JS stack, and can -// that interleave with a concurrently parked `driveAsync`? +// Event-driven host activity must defer ordinary scheduling until the current +// guest activation returns; one shared store drain then services the work. // // The shape, store-level in the style of `host_pump_test.ts` / -// `parked_driver_host_call_test.ts` (no checked-in example guest has the +// `host_pump_test.ts` (no checked-in example guest has the // participants): // // * instance A's SYNC export is on the JS stack — modelled by @@ -15,17 +14,17 @@ // * a ready fake thread of instance B records, in its `resume()`, whether // A's activation is live (both A's own body flag and `currentThread()`). // -// P-1 answers the reentrancy question; P-2 the two-loops question. +// P-1 covers reentrancy; P-2 covers the absence of resident async drivers. import { assertEq } from "./support/asserts.ts"; import { + createLiftedFunction, createLoweredImport, driveStoreAsync, hostStreamFor, newStats, registerHostCall, type ResolvedOptions, - storeDriverDepth, } from "../src/exec/mod.ts"; import { ComponentInstanceState, @@ -117,14 +116,12 @@ class FakeThread { readonly witness: { aBodyLive: boolean; ambient: unknown; - driverDepth: number; }[] = []; readonly task = { inst: {} }; constructor( private readonly probe: () => { aBodyLive: boolean; ambient: unknown; - driverDepth: number; }, ) {} ready(): boolean { @@ -144,12 +141,11 @@ class FakeThread { } // --------------------------------------------------------------------------- -// P-1: reentrance +// P-1: no reentrance // --------------------------------------------------------------------------- Deno.test({ - name: - "P-1: the sync half of pump() ticks instance B while instance A's activation is live", + name: "P-1: host activity defers B until instance A's activation returns", fn: async () => { const store = new Store(); const { shared, host } = hostEndOn(store, U8); @@ -159,7 +155,6 @@ Deno.test({ const b = new FakeThread(() => ({ aBodyLive, ambient: ambientOrNull(), - driverDepth: storeDriverDepth(store), })); store.startWaiting(b); @@ -178,24 +173,12 @@ Deno.test({ aBodyLive = false; } - // THE OBSERVATION. If B was resumed at all, it happened synchronously - // inside `a.call()` — nothing else ran. - assert( - b.resumed > 0, - "NOT REACHABLE: the sync half of pump() did not tick B at all", - ); + assertEq(b.resumed, 0, "host activity must not reenter a live activation"); + await Promise.resolve(); + assert(b.resumed > 0, "the requested drain did not run B"); ambientDuringResume = b.witness[0].ambient; - assert( - b.witness[0].aBodyLive, - "NOT REACHABLE: B was resumed outside A's activation", - ); - assertEq( - ambientDuringResume === a.thread, - true, - // `currentThread()` naming A's thread is the machine-checkable form of - // "a guest activation is live on the JS stack". - ); - assertEq(b.witness[0].driverDepth, 0); + assertEq(b.witness[0].aBodyLive, false); + assertEq(ambientDuringResume, null); // Housekeeping: settle the read so no live pump is left behind. shared.drop(); @@ -205,13 +188,62 @@ Deno.test({ }, }); +Deno.test("nested lifted entry does not tick an unrelated sibling inside the outer guest frame", async () => { + const store = new Store(); + const inst = new ComponentInstanceState(0, store); + let outerLive = false; + const sibling = new FakeThread(() => ({ + aBodyLive: outerLive, + ambient: ambientOrNull(), + })); + store.startWaiting(sibling); + + const opts: ResolvedOptions = { + stringEncoding: "utf8", + memory: null, + realloc: null, + postReturn: null, + callback: null, + async: false, + cancellable: false, + coreType: { params: [], results: [] }, + instance: inst, + }; + const nested = createLiftedFunction({ + name: "nested-noop", + ft: FT, + opts, + core: () => {}, + stats: newStats(), + }); + const outer = new Task(FT, TASK_OPTS, inst, () => [], () => {}); + const outerThread = new Thread(outer, (function* () {})()); + + outerLive = true; + pushCurrentThread(outerThread); + try { + assertEq(nested(), undefined); + assertEq( + sibling.resumed, + 0, + "nested call performed an ordinary store tick", + ); + } finally { + popCurrentThread(outerThread); + outerLive = false; + } + + await Promise.resolve(); + assertEq(sibling.resumed, 1, "deferred ordinary service did not run sibling"); + assertEq(sibling.witness[0].aBodyLive, false); +}); + // --------------------------------------------------------------------------- -// P-2: the two-loops question — a parked driveAsync plus the sync half +// P-2: pending host work does not keep the coordinator running // --------------------------------------------------------------------------- Deno.test({ - name: - "P-2: with a driveAsync parked on the same store, the sync half still ticks B", + name: "P-2: a pending host call does not create a resident driver", fn: async () => { const store = new Store(); const { shared, host } = hostEndOn(store, U8); @@ -220,7 +252,6 @@ Deno.test({ const b = new FakeThread(() => ({ aBodyLive, ambient: ambientOrNull(), - driverDepth: storeDriverDepth(store), })); // The incumbent loop: a `driveAsync` parked on a never-settling host call. @@ -230,7 +261,7 @@ Deno.test({ const driving = driveStoreAsync(store, () => finished, "test driver"); for (let i = 0; i < 10; i++) await Promise.resolve(); await new Promise((r) => setTimeout(r, 1)); - assert(storeDriverDepth(store) > 0, "the driver should be live"); + // A pending host call does not own a resident loop. // Only now make B ready, so the parked driver's snapshot predates it. store.startWaiting(b); @@ -249,8 +280,9 @@ Deno.test({ aBodyLive = false; } + assertEq(b.resumed, 0, "host activity must not synchronously run B"); + await Promise.resolve(); const syncResumes = b.resumed; - const depthSeen = b.witness.map((w) => w.driverDepth); // Let the parked driver have its turns too, and see whether B is resumed // a SECOND time (the resume-once question) or the store is poisoned. @@ -270,11 +302,9 @@ Deno.test({ assert( syncResumes > 0, - `NOT REACHABLE under a live driver: sync half did not tick B ` + - `(driver depth seen: ${JSON.stringify(depthSeen)})`, + `event-driven drain did not tick B ` + + `(no resident driver is available to rescue it)`, ); - // Reported, not asserted-away: what the probe is here to measure. - assertEq(depthSeen[0] > 0, true); assertEq(store.hostFailure, undefined); assertEq(b.resumed, syncResumes); }, @@ -295,7 +325,6 @@ Deno.test({ const b = new FakeThread(() => ({ aBodyLive, ambient: ambientOrNull(), - driverDepth: storeDriverDepth(store), })); store.startWaiting(b); diff --git a/runtime/tests/host_pump_test.ts b/runtime/tests/host_pump_test.ts index 883e7433..d30d94ce 100644 --- a/runtime/tests/host_pump_test.ts +++ b/runtime/tests/host_pump_test.ts @@ -30,7 +30,7 @@ // `TypeError: Cannot read properties of undefined (reading 'awaiting')`. import { assertEq } from "./support/asserts.ts"; -import { hostStreamFor } from "../src/exec/mod.ts"; +import { hostStreamFor, registerHostCall } from "../src/exec/mod.ts"; import { SharedStreamImpl, Store } from "../src/task/mod.ts"; import type { ComponentValue, ValType } from "../src/cabi/types.ts"; @@ -163,7 +163,7 @@ Deno.test({ guest.wake(); }, ); - store.pendingHostCalls.add(p); + registerHostCall(store, p); }); store.startWaiting(guest); @@ -293,6 +293,9 @@ Deno.test({ () => (settled = "resolved"), (e) => (settled = `rejected: ${(e as Error).message}`), ); + // The event-driven request runs on the next microtask rather than inside + // the host operation. Let that one bounded drain establish quiescence. + await Promise.resolve(); const afterFirstPump = ticks; // Bounded probe: many macrotask turns must pass with no further ticking diff --git a/runtime/tests/host_pump_trap_test.ts b/runtime/tests/host_pump_trap_test.ts index b3ecc091..7338d668 100644 --- a/runtime/tests/host_pump_trap_test.ts +++ b/runtime/tests/host_pump_trap_test.ts @@ -1,4 +1,4 @@ -// A trap raised by `HostActivity.pump()`'s SYNCHRONOUS half is mishandled. +// A trap raised by event-driven host activity must not wedge the host end. // // Every host stream/future op sets `parked.write`/`parked.read` and then calls // `activity.pump()` from INSIDE its `new Promise(executor)`. `pump()`'s @@ -105,10 +105,12 @@ Deno.test({ let firstOk = false; first.then(() => (firstOk = true), (e) => (firstErr = e)); await settleTurns(3); - assert( - firstOk || firstErr !== undefined, - "the first write neither resolved nor rejected", - ); + // The unrelated trap is parked on the store channel. This operation may + // remain pending until its peer acts; cancellation below must still work. + assertEq(firstOk, false); + assertEq(firstErr, undefined); + host.writable.cancelWrite(); + await settleTurns(1); // The fault has been delivered (or recorded). The END, however, belongs // to the embedder and must still be usable: the trap was raised by an @@ -151,7 +153,9 @@ Deno.test({ let settled = false; first.then(() => (settled = true), () => (settled = true)); await settleTurns(3); - assert(settled, "the first read neither resolved nor rejected"); + assertEq(settled, false); + host.readable.cancelRead(); + await settleTurns(1); let second: Promise | undefined; let secondThrow: unknown = undefined; diff --git a/runtime/tests/jspi/poison_before_background_listener_test.ts b/runtime/tests/jspi/poison_before_background_listener_test.ts index 39c59ff3..a70e1fef 100644 --- a/runtime/tests/jspi/poison_before_background_listener_test.ts +++ b/runtime/tests/jspi/poison_before_background_listener_test.ts @@ -2,6 +2,7 @@ import { instantiate } from "../../src/embedder/mod.ts"; import { Translator } from "../../src/shim/mod.ts"; import { isTrap, suspending } from "@polyengine/protocol"; import { assert, assertEquals } from "./asserts.ts"; +import { drainWaiterCountForTesting } from "../../src/exec/mod.ts"; const root = new URL("../../../", import.meta.url); @@ -99,6 +100,14 @@ Deno.test({ ); assertEquals(callbackEntered, true, "E callback did not reach its trap"); assertEquals(fEntered, false, "F entered the poisoned guest"); + await Promise.resolve(); + assertEquals( + drainWaiterCountForTesting( + component.handle.componentInstances[0].store, + ), + 0, + "poisoned idle call retained its coordinator waiter", + ); } finally { gate.resolve(); } diff --git a/runtime/tests/jspi/task_return_settlement_test.ts b/runtime/tests/jspi/task_return_settlement_test.ts index 06fdf5f8..54b32993 100644 --- a/runtime/tests/jspi/task_return_settlement_test.ts +++ b/runtime/tests/jspi/task_return_settlement_test.ts @@ -92,6 +92,65 @@ Deno.test({ }, }); +Deno.test({ + name: + "#323: a genuine second suspension releases same-instance entry admission", + ignore: shimWasm === null, + fn: async () => { + const gate = Promise.withResolvers(); + const gateEntered = Promise.withResolvers(); + const continued = Promise.withResolvers(); + const probeCall: { result: Promise | null } = { result: null }; + const translator = await Translator.create(shimWasm!); + const component = await instantiate({ + componentBytes, + ...translator.translate(componentBytes), + }, { + "before-gate": () => Promise.resolve(), + hop: suspending(() => { + probeCall.result = component.exports.probe() as Promise; + }), + gate: () => { + gateEntered.resolve(); + return gate.promise; + }, + continued: () => continued.resolve(), + "probe-entered": () => {}, + }); + + const result = component.exports.run() as Promise; + try { + await gateEntered.promise; + const probeResult = probeCall.result; + assert(probeResult !== null, "hop callback did not start the probe"); + const probe = await Promise.race([ + probeResult.then((value) => ({ state: "resolved", value })), + new Promise<{ state: "timeout" }>((resolve) => + setTimeout(() => resolve({ state: "timeout" }), 50) + ), + ]); + assertEquals( + JSON.stringify(probe), + JSON.stringify({ state: "resolved", value: 7 }), + "same-instance probe did not resolve after the producer genuinely parked", + ); + assertEquals(await result, 42); + } finally { + gate.resolve(); + } + + await continued.promise; + const later = await assertRejects( + () => component.exports.probe() as Promise, + "post-delivery producer failure must remain observable", + ); + assert( + isTrap(later) && String(later).includes("unreachable"), + `expected the retained producer trap, got ${later}`, + ); + }, +}); + Deno.test({ name: "#323: a late host rejection remains observable after result delivery", ignore: shimWasm === null, diff --git a/runtime/tests/lift_background_return_test.ts b/runtime/tests/lift_background_return_test.ts index d6dda609..88142816 100644 --- a/runtime/tests/lift_background_return_test.ts +++ b/runtime/tests/lift_background_return_test.ts @@ -18,6 +18,7 @@ import { assertEq } from "./support/asserts.ts"; import { createLiftedFunction, + drainWaiterCountForTesting, newStats, type ResolvedOptions, } from "../src/exec/mod.ts"; @@ -152,5 +153,10 @@ Deno.test({ ); assertEq(store.hostFailure, undefined); assertEq(store.pendingHostCalls.size, 0); + assertEq( + drainWaiterCountForTesting(store), + 0, + "terminal idle call retained its coordinator waiter", + ); }, }); diff --git a/runtime/tests/lift_done_verdict_test.ts b/runtime/tests/lift_done_verdict_test.ts index cbf4b732..4c967897 100644 --- a/runtime/tests/lift_done_verdict_test.ts +++ b/runtime/tests/lift_done_verdict_test.ts @@ -161,7 +161,7 @@ Deno.test({ }, }); -Deno.test("sync entry keeps a synchronous result after a foreign routed fault", () => { +Deno.test("sync entry returns before ordinary foreign work is serviced", async () => { const store = new Store(); const healthyInst = new ComponentInstanceState(0, store); const failedInst = new ComponentInstanceState(1, store); @@ -216,10 +216,13 @@ Deno.test("sync entry keeps a synchronous result after a foreign routed fault", const result = fn(); assertEq(result, 42); assertEq(result instanceof Promise, false); + assertEq(originFailed, false); + assertEq(healthyReadyRan, false); + await Promise.resolve(); assertEq(originFailed, true); assertEq( healthyReadyRan, true, - "the export driver must continue to the healthy ready thread", + "ordinary service must continue to the healthy ready thread after entry", ); }); diff --git a/runtime/tests/parked_driver_host_call_test.ts b/runtime/tests/parked_driver_host_call_test.ts deleted file mode 100644 index 396343d6..00000000 --- a/runtime/tests/parked_driver_host_call_test.ts +++ /dev/null @@ -1,117 +0,0 @@ -// A host call registered while a driver is already parked still wakes the -// guest — the stale-snapshot half of the #239 class. -// -// THE SHAPE (store-level, in the style of host_pump_test.ts and for the same -// reason: no checked-in example guest has the participants) -// =========================================================================== -// -// * a driver is live and parked on `Promise.race([...pendingHostCalls, -// ...])` because one outstanding host call never settles (the end-to-end -// original: a `readDirect` stream session keeping a `HostActivity` driver -// alive while the guest holds a long-poll import open); -// * a SECOND host call is then registered with no new driver entered — end -// to end that is an export entered through the synchronous `drive` path, -// which fires no driver arrival; here it is `registerHostCall` called -// directly, which is exactly what that path reduces to; -// * that call settles. Its continuation deletes itself from -// `pendingHostCalls` and readies the guest thread — and readying is all -// it does. Somebody has to tick the store. -// -// Pre-fix nobody did: the parked driver's snapshot predates the second call, -// and the settlement pump stands down because `storeDriverDepth > 0` (the -// parked driver counts). The guest sat until the next unrelated export call. -// Verified to fail on the pre-fix runtime: `resumed` stays 0 for the whole -// probe window. - -import { assertEq } from "./support/asserts.ts"; -import { driveStoreAsync, registerHostCall } from "../src/exec/mod.ts"; -import { Store } from "../src/task/mod.ts"; - -function assert(cond: boolean, msg: string): asserts cond { - if (!cond) throw new Error(`assertion failed: ${msg}`); -} - -/** The stand-in guest thread of host_pump_test.ts, trimmed to what a - * settlement-resumption needs: `Store.tick` resumes it whenever `ready()`. */ -class FakeThread { - #ready = false; - resumed = 0; - readonly task = { inst: {} }; - ready(): boolean { - return this.#ready; - } - waiting(): boolean { - return !this.#ready; - } - wake(): void { - this.#ready = true; - } - resume(): void { - this.#ready = false; - this.resumed++; - } -} - -Deno.test({ - name: - "a host call registered while a driver is parked wakes the guest (stale snapshot, #239 class)", - fn: async () => { - const store = new Store(); - const guest = new FakeThread(); - store.startWaiting(guest); - - // The never-settling call that keeps the incumbent driver parked, and - // the driver itself: `done` never fires on its own, so it exits only via - // the flag below (the test's stand-in for the stream session ending). - const stuck = new Promise(() => {}); - registerHostCall(store, stuck); - let finished = false; - const driving = driveStoreAsync(store, () => finished, "test driver"); - - // Let it reach the park. - for (let i = 0; i < 10; i++) await Promise.resolve(); - await new Promise((r) => setTimeout(r, 1)); - assertEq(guest.resumed, 0); - - // The late registration, modelled exactly as the async-lower site does - // it (exec/boundary.ts `createLoweredImport`): a promise whose - // continuation deletes its own entry and readies the guest, registered - // via `registerHostCall`. No driver is entered. - let settle!: () => void; - const raw = new Promise((r) => (settle = r)); - const call: Promise = raw.then(() => { - store.pendingHostCalls.delete(call); - guest.wake(); - }); - registerHostCall(store, call); - - settle(); - - // A few microtask/timer turns is all a woken driver needs: it drops out - // of the race, re-snapshots, and ticks the ready thread. - for (let i = 0; i < 20 && guest.resumed === 0; i++) { - await Promise.resolve(); - } - for (let i = 0; i < 5 && guest.resumed === 0; i++) { - await new Promise((r) => setTimeout(r, 1)); - } - assert( - guest.resumed > 0, - "the guest was never resumed: the parked driver's host-call snapshot " + - "was stale and nothing else drove the store", - ); - assertEq(store.hostFailure, undefined); - - // Release the driver so the test leaves no live loop behind — and leave - // `pendingHostCalls` EMPTY on the way out, or the settlement pump inherits - // an entry that can never settle again and spins. - finished = true; - store.pendingHostCalls.delete(stuck); - const wake: Promise = Promise.resolve().then(() => { - store.pendingHostCalls.delete(wake); - }); - registerHostCall(store, wake); - await driving; - assertEq(store.pendingHostCalls.size, 0); - }, -}); diff --git a/runtime/tests/same_store_driver_test.ts b/runtime/tests/same_store_driver_test.ts deleted file mode 100644 index 5f9151a8..00000000 --- a/runtime/tests/same_store_driver_test.ts +++ /dev/null @@ -1,302 +0,0 @@ -// Concurrent drivers on the SAME store (issue #239). -// -// The same-store half of #210. #210 made the driver's speculative resume -// entry per-store (`Store.pendingResumptions`), which stopped an unrelated -// store's driver from spinning on it — see `tests/cross_store_driver_test.ts`, -// whose header describes the entry as "held for the entire duration of a -// guest's wait on a slow host import". It left the same wedge intact for a -// SECOND driver on the store actually doing the waiting, which is the ordinary -// shape (two overlapping export calls; a detached guest task cancelling an -// in-flight import while the settlement pump holds the entry). -// -// Mechanism: `driveAsync` (exec/boundary.ts) takes the speculative entry and -// holds it across `Promise.race([chosenTag, ...others])`, where `others` -// includes `store.pendingHostCalls` — a window bounded only by when the HOST -// answers, i.e. possibly never. `Store.pendingResumptions` is a store-wide -// scheduling gate: `Store.tick()` returns false while it is non-empty -// (task/scheduler.ts), and every `driveAsync` yields at its top while -// `store.hasPendingResumptions()` holds, under a 10,000-hop bound that fires -// -// assert_(claimHops < 10_000, "driveAsync: a resumed-activation claim was -// never released (the activation neither parked, finished, nor trapped)") -// -// So the second driver died in ~311ms — an internal-bug detector firing on a -// perfectly ordinary suspended guest. -// -// The fix (boundary.ts, "Driver arrival: closing the overlap window"): the -// entry is taken only by the SOLE driver (`storeDriverDepth(store) === 1`), -// and `armDriverArrival(store)` rides the race so an incumbent wakes within a -// microtask when another driver arrives, drops the entry, and re-evaluates -// `done()`. -// -// These tests pin all three halves: the second driver returns promptly (1), -// the gate still gates when nobody else wants the store (2), and the entry is -// actually dropped while a second driver is live (3). -// -// Scaffolding style: cross_store_driver_test.ts / settlement_pump_test.ts -// (fake inst/threads, real Store, real driveStoreAsync). - -import { driveStoreAsync } from "../src/exec/mod.ts"; -import { Store } from "../src/task/mod.ts"; - -function assert(cond: boolean, msg: string): asserts cond { - if (!cond) throw new Error(`assertion failed: ${msg}`); -} - -function fakeInst() { - return {}; -} - -/** A thread parked on an awaitValue promise, as a promising-wrapped guest - * activation suspended on a host import is. */ -function awaitingThread(store: Store, p: Promise) { - const t = { - awaiting: p as Promise | null, - task: { inst: fakeInst() }, - ready: () => false, - waiting: () => false, - resume: () => {}, - resumeWith(_v: unknown, _f?: { error: unknown }) { - t.awaiting = null; - store.awaiting.delete(t); - }, - }; - store.noteAwaiting(t, p); - return t; -} - -function hostImport(store: Store, settle: Promise): void { - const p: Promise = settle.then(() => { - store.pendingHostCalls.delete(p); - }); - store.pendingHostCalls.add(p); -} - -/** Poll `cond` until true or the deadline passes; returns whether it held. - * Macrotask hops, so the drivers under observation get to run. */ -async function waitFor(cond: () => boolean, ms: number): Promise { - const t0 = Date.now(); - while (!cond()) { - if (Date.now() - t0 >= ms) return false; - await new Promise((r) => setTimeout(r, 1)); - } - return true; -} - -// The regression. Driver A dwells on a slow host import; driver B arrives on -// the SAME store with nothing to do. Pre-fix B spun at the top of its own loop -// against A's speculative entry and died at the hop bound in ~311ms. -Deno.test("a second driver on the same store is not wedged by the incumbent's speculative entry (#239)", async () => { - const store = new Store(); - let settleThread!: (v: unknown) => void; - let settleHost!: (v: unknown) => void; - const threadP = new Promise((r) => (settleThread = r)); - const hostP = new Promise((r) => (settleHost = r)); - let aDone = false; - let aDriver: Promise = Promise.resolve(); - try { - // A guest suspended on a host import that never answers. - awaitingThread(store, threadP); - hostImport(store, hostP); - aDriver = driveStoreAsync(store, () => aDone, "A: dweller").catch((e) => e); - - // Wait until A has reached its race holding the entry. A is the sole - // driver here, so `sole` is true and the entry IS taken — the predicate is - // as good a "A is parked in the race" signal as it is in the cross-store - // test (and it is the exact state the bug needs). - assert( - await waitFor(() => store.hasPendingResumptions(), 2000), - "A's driver never took its entry", - ); - - // Driver B: same store, immediately done. It must not consult A's entry. - const tB = Date.now(); - let threw: unknown = null; - try { - await driveStoreAsync(store, () => true, "B: arriving"); - } catch (e) { - threw = e; - } - const elapsed = Date.now() - tB; - // Name the bug in the failure report: pre-fix this is the claimHops - // AssertionError, not some generic rejection. - assert( - threw === null, - `B's driver must not throw; got: ${String(threw)}` + - (String(threw).includes("resumed-activation claim") - ? " -- this is the #239 wedge: the incumbent's speculative " + - "Store.pendingResumptions entry gated B's whole loop" - : ""), - ); - // Pre-fix: ~311ms (10,000 hops) ending in that assert. The bound is loose - // on purpose — the property is "promptly, not gated on A's host" — and it - // is the throw above that carries the regression; this only pins that B - // cannot instead be made to dwell for the host's own duration. - assert( - elapsed < 2000, - `B's driver returned in ${elapsed}ms, expected < 2s`, - ); - } finally { - aDone = true; - settleThread(0); - settleHost(0); - await aDriver; - store.pendingResumptions.clear(); - } -}); - -// CONTROL, mirroring the tail of cross_store_driver_test.ts: the fix narrows -// WHO may hold the speculative entry, it does not delete it. With A the sole -// driver parked in its race, the gate is held and its own store refuses to -// schedule past it. -Deno.test("the speculative entry still gates when the incumbent is the sole driver (#239 control)", async () => { - const store = new Store(); - let settleThread!: (v: unknown) => void; - let settleHost!: (v: unknown) => void; - const threadP = new Promise((r) => (settleThread = r)); - const hostP = new Promise((r) => (settleHost = r)); - let aDone = false; - let aDriver: Promise = Promise.resolve(); - try { - awaitingThread(store, threadP); - hostImport(store, hostP); - aDriver = driveStoreAsync(store, () => aDone, "A: dweller").catch((e) => e); - - assert( - await waitFor(() => store.hasPendingResumptions(), 2000), - "A's driver never took its entry", - ); - assert(store.hasPendingResumptions(), "A's entry is still pending"); - assert(store.tick() === false, "the gate still refuses to schedule"); - } finally { - aDone = true; - settleThread(0); - settleHost(0); - await aDriver; - store.pendingResumptions.clear(); - } -}); - -// The mechanism of the fix, observed directly: while a second driver is live, -// nobody holds the store-wide gate. A wakes on `armDriverArrival`, drops the -// entry in the race's `finally`, and re-parks with `sole === false`. -// -// B is kept ALIVE across the observation (its `done` stays false until we have -// looked) deliberately: once B exits, A is sole again and legitimately retakes -// the entry, so a post-hoc read would be a race against A's next lap. -Deno.test("the speculative entry is dropped while a second driver is live on the store (#239)", async () => { - const store = new Store(); - let settleThread!: (v: unknown) => void; - let settleHost!: (v: unknown) => void; - const threadP = new Promise((r) => (settleThread = r)); - const hostP = new Promise((r) => (settleHost = r)); - let aDone = false; - let bDone = false; - let aDriver: Promise = Promise.resolve(); - let bDriver: Promise = Promise.resolve(); - try { - awaitingThread(store, threadP); - hostImport(store, hostP); - aDriver = driveStoreAsync(store, () => aDone, "A: dweller").catch((e) => e); - - assert( - await waitFor(() => store.hasPendingResumptions(), 2000), - "A's driver never took its entry", - ); - - bDriver = driveStoreAsync(store, () => bDone, "B: co-resident") - .catch((e) => e); - - // A must stand out of its race and release the entry. Bounded wait, not a - // bare read: the handoff is a few microtask hops, not synchronous. - assert( - await waitFor(() => !store.hasPendingResumptions(), 2000), - "the incumbent never released the store-wide gate for the arriving " + - "driver (#239: `armDriverArrival` did not reach A's race, or A " + - "retook the entry while a second driver was live)", - ); - } finally { - aDone = true; - bDone = true; - settleThread(0); - settleHost(0); - await aDriver; - await bDriver; - store.pendingResumptions.clear(); - } -}); - -// The release path removes ONLY the driver's own entry (issue #158's lesson, -// re-asserted under #239's new wake path). -// -// When another driver arrives, the awaiting driver stands down and removes -// only its own speculative pending entry. An entry added mid-await by -// `SuspensionPoint.resume` must survive (#158). Two entries are needed to -// distinguish identity-scoped removal from clearing the whole set. -// -// The foreign entry is a bare sentinel object rather than a second -// `awaitingThread`. `pendingResumptions` is `Set` and the property -// under test is purely identity-scoped removal — the set's members are never -// dereferenced by the driver, only added/removed/counted — so a sentinel says -// exactly what is being tested and cannot accidentally participate in -// scheduling. A second parked thread would add awaiting-set membership the -// property does not involve. -// -// Delete the sentinel first during teardown: it holds the store-wide gate, -// so neither driver can reach `done()` until it is removed. -Deno.test("a driver releases only its own speculative entry, not the whole gate (#158 under #239's stand-down)", async () => { - const store = new Store(); - let settleThread!: (v: unknown) => void; - let settleHost!: (v: unknown) => void; - const threadP = new Promise((r) => (settleThread = r)); - const hostP = new Promise((r) => (settleHost = r)); - const foreign = { what: "an entry minted by SuspensionPoint.resume" }; - let aDone = false; - let bDone = false; - let aDriver: Promise = Promise.resolve(); - let bDriver: Promise = Promise.resolve(); - try { - const thread = awaitingThread(store, threadP); - hostImport(store, hostP); - aDriver = driveStoreAsync(store, () => aDone, "A: dweller").catch((e) => e); - - // A is sole, so the entry it takes in the race is `thread` itself. - assert( - await waitFor(() => store.pendingResumptions.has(thread), 2000), - "A's driver never took its entry", - ); - - // The delivery-in-flight entry, minted while A is parked in the race. - store.addPendingResumption(foreign); - - // B's arrival wakes A out of the race and runs its `finally`. B stays live - // across the observation for the same reason as the previous test. - bDriver = driveStoreAsync(store, () => bDone, "B: co-resident") - .catch((e) => e); - - assert( - await waitFor(() => !store.pendingResumptions.has(thread), 2000), - "A never released its own entry on the stand-down path", - ); - // THE PROPERTY: A named what it added. A blanket clear here would take the - // in-flight delivery's entry with it — the #158 regression, invisible to - // every other test in this file because they only ever have one entry. - assert( - store.pendingResumptions.has(foreign), - "A's release path cleared an entry it did not add (#158: the " + - "`finally` must remove only `chosen`, not clear the set — an entry " + - "minted mid-await by a guest-synchronous delivery must survive)", - ); - } finally { - // Sentinel first: it is the only thing holding the gate now, and both - // drivers are yielding against the hop bound until it goes. - store.pendingResumptions.delete(foreign); - aDone = true; - bDone = true; - settleThread(0); - settleHost(0); - await aDriver; - await bDriver; - store.pendingResumptions.clear(); - } -}); diff --git a/runtime/tests/settlement_pump_test.ts b/runtime/tests/settlement_pump_test.ts index 0c0a52d4..1e23a21a 100644 --- a/runtime/tests/settlement_pump_test.ts +++ b/runtime/tests/settlement_pump_test.ts @@ -1,12 +1,10 @@ -// The settlement pump (exec/boundary.ts): liveness between export calls. +// Event-driven store service: liveness between export calls. // -// A host-import promise that settles while NO driver is live only mutates -// scheduler state — the registration site's continuation readies the guest -// thread but nothing ticks the store. Before the settlement pump, that work -// sat queued until the next export call or host stream/future operation; a +// A host-import promise that settles while no call is driving only mutates +// scheduler state — its reaction must request the shared store coordinator. A // guest whose wakeup is a host clock (the componentize-go keep-alive-ticker // shape: a task parked WAIT whose pending host call is a wasi:clocks -// `wait-for`) was frozen between embedder calls. These tests pin the pump's +// `wait-for`) must progress between embedder calls. These tests pin the // contract at store level, in the style of host_pump_test.ts: // // * a `Store` and a fake guest thread (the `SchedulableThread` surface); @@ -15,13 +13,13 @@ // settle continuation deletes itself and readies the guest, and does NOT // tick the store; // * "an export call just returned" modelled as one `driveStoreAsync` round -// with an immediately-true `done` — the pump is armed at driver exit. +// with an immediately-true `done`. // // Verified against the pre-pump runtime: T-1 and T-2 time out (the guest is // never resumed), T-3 and T-4 pass vacuously/identically. import { assertEq } from "./support/asserts.ts"; -import { driveStoreAsync } from "../src/exec/mod.ts"; +import { driveStoreAsync, registerHostCall } from "../src/exec/mod.ts"; import { markHostActivityArm, Store } from "../src/task/mod.ts"; /** The slice of `ComponentInstance` that `Store.tick` touches. */ @@ -66,7 +64,7 @@ function hostImport( store.pendingHostCalls.delete(p); onSettle(); }); - store.pendingHostCalls.add(p); + registerHostCall(store, p); } function withTimeout(p: Promise, label: string, ms = 4000): Promise { @@ -188,7 +186,7 @@ Deno.test({ store.pendingHostCalls.delete(p); store.hostFailure = e; }); - store.pendingHostCalls.add(p); + registerHostCall(store, p); await exportCallReturns(store); // Wait out the settlement plus pump unwind. diff --git a/tools/wasmtime/worker.ts b/tools/wasmtime/worker.ts index debda862..a85139dc 100644 --- a/tools/wasmtime/worker.ts +++ b/tools/wasmtime/worker.ts @@ -20,4 +20,13 @@ const result = await runWastJson( (name) => Deno.readFile(join(generated, dir, name)), await RuntimeExecutor.create(shim, wasmtimeSpectest(file).imports), ); -console.log(JSON.stringify(result)); +const output = new TextEncoder().encode(`${JSON.stringify(result)}\n`); +let written = 0; +while (written < output.length) { + written += Deno.stdout.writeSync(output.subarray(written)); +} +// This worker is a bounded test scope, not the runtime's process-lifetime +// owner. Some valid upstream cases intentionally leave background guest work +// runnable after their final assertion. Exit only after the complete validated +// result has been written; pre-result failures still reject and exit nonzero. +Deno.exit(0);