diff --git a/src/browser/WebBundleAnalysisWorker.ts b/src/browser/WebBundleAnalysisWorker.ts index 9022229d26..83c3a1a644 100644 --- a/src/browser/WebBundleAnalysisWorker.ts +++ b/src/browser/WebBundleAnalysisWorker.ts @@ -1,3 +1,4 @@ +import { WorkerFrameSignal } from "../process/WorkerFrameSignal.js"; import { createHash, randomUUID } from "node:crypto"; import { open, rm } from "node:fs/promises"; import { join } from "node:path"; @@ -139,7 +140,7 @@ export class WebBundleAnalysisWorker { #activeDeadline: ProviderStartupDeadline | undefined; #activeJob: WebBundleActiveJob | undefined; #activeTerminalResponseObserved = false; - readonly #frameWaiters = new Set<(outcome: "frame" | "failure") => void>(); + readonly #frameSignal = new WorkerFrameSignal(); #retainedResultBytes = 0; #ready = false; readonly #stdout = (chunk: Buffer | string): void => { @@ -172,12 +173,12 @@ export class WebBundleAnalysisWorker { this.#observeFrame(this.#frames.at(-1)!); this.#pending = this.#pending.slice(end + 1); } - this.#wakeFrameWaiters("frame"); + this.#frameSignal.wake("frame"); } catch (cause) { this.#pending = ""; this.#protocolFailure = cause; this.#latchFailure({ kind: "protocol", cause }); - this.#wakeFrameWaiters("failure"); + this.#frameSignal.wake("failure"); } }; @@ -200,7 +201,7 @@ export class WebBundleAnalysisWorker { onInputFailure: (cause) => { if (this.#activeDeadline?.interruption !== undefined) return; this.#latchFailure({ kind: "input", cause }); - this.#wakeFrameWaiters("failure"); + this.#frameSignal.wake("failure"); }, supervisorOptions: { captureStdout: false, @@ -232,7 +233,7 @@ export class WebBundleAnalysisWorker { !this.#activeTerminalResponseObserved) ) this.#latchFailure({ kind: "incomplete-output" }); - this.#wakeFrameWaiters("failure"); + this.#frameSignal.wake("failure"); } else if (event.type === "output") { const failure = this.#observedFailure; if (failure?.kind !== "exit") return; @@ -780,41 +781,12 @@ export class WebBundleAnalysisWorker { }; } - #wakeFrameWaiters(outcome: "frame" | "failure"): void { - for (const wake of this.#frameWaiters) wake(outcome); - this.#frameWaiters.clear(); - } - async #waitForFrameOrOutputClose( supervisor: NonNullable, deadline: ProviderStartupDeadline, ): Promise<"frame" | "failure" | "closed" | "interrupted"> { if (this.#frames.length > 0) return "frame"; - let wakeFrame!: (outcome: "frame" | "failure") => void; - const frameAvailable = new Promise<"frame" | "failure">((resolve) => { - wakeFrame = resolve; - this.#frameWaiters.add(wakeFrame); - }); - const outputCloseController = new AbortController(); - const interruptOutputClose = () => - outputCloseController.abort(deadline.signal.reason); - if (deadline.signal.aborted) interruptOutputClose(); - else - deadline.signal.addEventListener("abort", interruptOutputClose, { - once: true, - }); - try { - const outputClose = supervisor - .waitForOutputClose(undefined, outputCloseController.signal) - .then((closed): "closed" | "interrupted" => - closed ? "closed" : "interrupted", - ); - return await Promise.race([frameAvailable, outputClose]); - } finally { - outputCloseController.abort(); - deadline.signal.removeEventListener("abort", interruptOutputClose); - this.#frameWaiters.delete(wakeFrame); - } + return this.#frameSignal.waitForOutputClose(supervisor, deadline.signal); } #completedSnapshot(cause: unknown): StableArtifactSnapshot | undefined { @@ -875,7 +847,7 @@ export class WebBundleAnalysisWorker { this.#activeDeadline = undefined; this.#activeJob = undefined; this.#activeTerminalResponseObserved = false; - this.#wakeFrameWaiters("failure"); + this.#frameSignal.wake("failure"); } #cleanupFailure(cause: unknown, primary?: AnalysisError): AnalysisError { diff --git a/src/javascript/analysis/JavaScriptAnalysisWorker.ts b/src/javascript/analysis/JavaScriptAnalysisWorker.ts index 3eab48aefe..f4e1440076 100644 --- a/src/javascript/analysis/JavaScriptAnalysisWorker.ts +++ b/src/javascript/analysis/JavaScriptAnalysisWorker.ts @@ -1,3 +1,4 @@ +import { WorkerFrameSignal } from "../../process/WorkerFrameSignal.js"; import type { JavaScriptSemanticProjectionCapacity } from "../../domain/javascript/javascriptSemanticFileProjection.js"; import { createHash, randomUUID } from "node:crypto"; import { readFile, rm, stat, writeFile } from "node:fs/promises"; @@ -100,7 +101,7 @@ export class JavaScriptAnalysisWorker { #ready = false; #heapLimitBytes = 0; #frames: JavaScriptWorkerMessage[] = []; - #frameWaiters = new Set<(outcome: "frame" | "failure") => void>(); + readonly #frameSignal = new WorkerFrameSignal(); #pending = ""; #decoder = new StringDecoder("utf8"); #primaryFailure: WorkerPrimaryFailure | undefined; @@ -141,7 +142,7 @@ export class JavaScriptAnalysisWorker { const frame = javaScriptWorkerMessageSchema.parse(raw); this.#frames.push(frame); this.#observeJobFrame(frame); - this.#wakeFrameWaiters("frame"); + this.#frameSignal.wake("frame"); } if (Buffer.byteLength(this.#pending) > DEFAULTS.protocolBytes) this.#latchOutputFailure( @@ -837,7 +838,7 @@ export class JavaScriptAnalysisWorker { ) ) { this.#primaryFailure = { ...this.#primaryFailure, heapExhaustion: true }; - this.#wakeFrameWaiters("failure"); + this.#frameSignal.wake("failure"); } if (event.type === "exit" && (event.code !== 0 || event.signal !== null)) { if (this.#primaryFailure === undefined) @@ -860,7 +861,7 @@ export class JavaScriptAnalysisWorker { ...this.#primaryFailure, heapExhaustion: true, }; - this.#wakeFrameWaiters("failure"); + this.#frameSignal.wake("failure"); } } } @@ -880,7 +881,7 @@ export class JavaScriptAnalysisWorker { this.#primaryFailure?.kind === "memory" || this.#primaryFailure?.kind === "exit" ) - this.#wakeFrameWaiters("failure"); + this.#frameSignal.wake("failure"); } } #observeJobFrame(frame: JavaScriptWorkerMessage): void { @@ -906,7 +907,7 @@ export class JavaScriptAnalysisWorker { kind: "output", reason: "JavaScript worker completed before a static checkpoint", }; - this.#wakeFrameWaiters("failure"); + this.#frameSignal.wake("failure"); return; } this.#terminalResponseId = frame.id; @@ -925,48 +926,20 @@ export class JavaScriptAnalysisWorker { if (this.#deadline?.interruption !== undefined) return; if (this.#primaryFailure === undefined) this.#primaryFailure = { kind: "output", reason }; - this.#wakeFrameWaiters("failure"); + this.#frameSignal.wake("failure"); } #latchInputFailure(cause: Error): void { if (this.#deadline?.interruption !== undefined) return; if (this.#primaryFailure === undefined) this.#primaryFailure = { kind: "input", cause }; - this.#wakeFrameWaiters("failure"); + this.#frameSignal.wake("failure"); } async #waitForFrameOrOutputClose( supervisor: ProviderProcessSupervisor, deadline: ProviderStartupDeadline, ): Promise<"frame" | "failure" | "closed" | "interrupted"> { if (this.#frames.length !== 0) return "frame"; - let wakeFrame!: (outcome: "frame" | "failure") => void; - const frameArrival = new Promise<"frame" | "failure">((resolve) => { - wakeFrame = resolve; - this.#frameWaiters.add(wakeFrame); - }); - const outputCloseController = new AbortController(); - const interruptOutputClose = () => - outputCloseController.abort(deadline.signal.reason); - if (deadline.signal.aborted) interruptOutputClose(); - else - deadline.signal.addEventListener("abort", interruptOutputClose, { - once: true, - }); - try { - const outputClose = supervisor - .waitForOutputClose(undefined, outputCloseController.signal) - .then((closed): "closed" | "interrupted" => - closed ? "closed" : "interrupted", - ); - return await Promise.race([frameArrival, outputClose]); - } finally { - outputCloseController.abort(); - deadline.signal.removeEventListener("abort", interruptOutputClose); - this.#frameWaiters.delete(wakeFrame); - } - } - #wakeFrameWaiters(outcome: "frame" | "failure"): void { - for (const wake of this.#frameWaiters) wake(outcome); - this.#frameWaiters.clear(); + return this.#frameSignal.waitForOutputClose(supervisor, deadline.signal); } #selectedWorkerFailure(path: string): AnalysisError { if (this.#primaryFailure !== undefined) @@ -1231,7 +1204,7 @@ export class JavaScriptAnalysisWorker { #resetState(): void { this.#ready = false; this.#frames = []; - this.#frameWaiters.clear(); + this.#frameSignal.clear(); this.#pending = ""; this.#decoder = new StringDecoder("utf8"); this.#primaryFailure = undefined; diff --git a/src/process/WorkerFrameSignal.ts b/src/process/WorkerFrameSignal.ts new file mode 100644 index 0000000000..dae2a2ee01 --- /dev/null +++ b/src/process/WorkerFrameSignal.ts @@ -0,0 +1,45 @@ +import type { ProviderProcessSupervisor } from "./ProviderProcess.js"; + +type FrameOutcome = "frame" | "failure"; + +/** Own frame waiters and release the losing output-close wait after each race. */ +export class WorkerFrameSignal { + readonly #waiters = new Set<(outcome: FrameOutcome) => void>(); + + wake(outcome: FrameOutcome): void { + for (const wake of this.#waiters) wake(outcome); + this.clear(); + } + + clear(): void { + this.#waiters.clear(); + } + + async waitForOutputClose( + supervisor: ProviderProcessSupervisor, + signal: AbortSignal, + ): Promise { + let wakeFrame!: (outcome: FrameOutcome) => void; + const frameAvailable = new Promise((resolve) => { + wakeFrame = resolve; + this.#waiters.add(wakeFrame); + }); + const outputCloseController = new AbortController(); + const interruptOutputClose = () => + outputCloseController.abort(signal.reason); + if (signal.aborted) interruptOutputClose(); + else signal.addEventListener("abort", interruptOutputClose, { once: true }); + try { + const outputClose = supervisor + .waitForOutputClose(undefined, outputCloseController.signal) + .then((closed): "closed" | "interrupted" => + closed ? "closed" : "interrupted", + ); + return await Promise.race([frameAvailable, outputClose]); + } finally { + outputCloseController.abort(); + signal.removeEventListener("abort", interruptOutputClose); + this.#waiters.delete(wakeFrame); + } + } +}