From acc83bd4bb049b74ff98b1e5591dfc558d9ac022 Mon Sep 17 00:00:00 2001 From: Vance Ingalls Date: Wed, 13 May 2026 21:06:54 -0700 Subject: [PATCH] perf(producer): pipeline capture and shader-blend per-frame (hf#732 PR 5/5) (#760) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary PR 5 of 5 in the hf#732 decomposition stack. Adds a per-worker K-deep ring of transition buffer-triples to the hybrid layered path. Capture-N+1 on the DOM worker now runs concurrently with the shader-blend pool's work on frames N-K+1..N instead of being serialized behind each blend. ### Mechanism - Each worker carries a ring of K buffer triples (`bufferA` / `bufferB` / `output`), default K=4. - The DOM worker round-robins through slots; on ring wrap, it awaits any still-in-flight blend on that slot before reusing its buffers. - The shader-blend dispatch is no longer awaited inline. It returns the pool's promise (or the inline-fallback promise), which is stored in `ringInFlight[slot]`. The blend, buffer-reattach, and ordered encoder write all run inside that promise. - The encoder reorder buffer (from PR 4) fences final output order — out-of-order blend completion is fine. ### Why K=4 The optimal K is `blend_per_frame / capture_per_frame`. For 854×480 rgb48le with complex shaders this is ~910ms / ~175ms ≈ 5. K=4 balances perf vs. memory: | K | Pool concurrency | Wall (hf#677 fixture) | |---|---|---| | 1 (PR 4) | ≤1 task/worker | ~135s | | 2 | 2–4 tasks | ~135s | | 4 | saturated | ~100s — **chosen** | | 10 | saturated + idle slots | ~100s | Memory: 6 workers × 4 slots × 3 buffers × 854×480×6 bytes ≈ 180MB peak. Override at runtime via `HF_TRANSITION_RING_DEPTH`. ### Failure modes - Pool spawn failed in PR 3 → inline blend fallback still works (each slot just resolves quickly). - Slot rejection caught onto a separate handle so unhandled-rejection can't fire; the error surfaces on next slot-await OR on end-of-task drain. - End-of-task drain awaits every remaining in-flight slot — worker success guarantees all blends hit the encoder. ## Stack Top of the hf#732 decomposition stack. Stacked on top of #759 (PR 4: hybrid path). ## Test plan - [x] Producer typecheck clean - [x] oxlint clean - [x] oxfmt clean ### Empirical validation Mark Witt fixture (Mac, Apple Silicon, hardware GPU, no beginframe): - Published CLI (pre-stack): 2m 12.2s - Cascade CLI (full hf#732 stack): 1m 07.7s - **Measured speedup: ~2× on Mac (1.95× exact).** (Earlier "2.22×" wording was a per-component projection; the empirical end-to-end number is 1.95× on the validated fixture.) Linux CI confirmation pending — top-of-stack regression run will surface the Linux number. — Vai --- .../render/stages/captureHdrHybridLoop.ts | 129 +++++++++++++----- 1 file changed, 97 insertions(+), 32 deletions(-) diff --git a/packages/producer/src/services/render/stages/captureHdrHybridLoop.ts b/packages/producer/src/services/render/stages/captureHdrHybridLoop.ts index f88aa6158..ba8251832 100644 --- a/packages/producer/src/services/render/stages/captureHdrHybridLoop.ts +++ b/packages/producer/src/services/render/stages/captureHdrHybridLoop.ts @@ -147,15 +147,37 @@ export async function runHybridLayeredFrameLoop(input: HybridLoopInput): Promise } const workerCanvases: Buffer[] = sessions.map(() => Buffer.alloc(bufSize)); - const workerTransitionBuffers: Array = sessions.map(() => - hasTransitions - ? { - bufferA: Buffer.alloc(bufSize), - bufferB: Buffer.alloc(bufSize), - output: Buffer.alloc(bufSize), - } - : null, + // hf#732 PR 5: K-deep ring of transition buffer-triples per worker. The + // ring lets capture-N+1 proceed on the DOM worker while the shader-blend + // pool is still working on frames N-K+1..N. Without the ring (PR 4), each + // worker awaited its own blend before the next capture, capping the pool + // at <=1 task per worker. With K=4, the pool sees up to min(N_workers * K, + // poolSize) concurrent blends, which empirically pushes shader-render + // wall time another ~10-20% past PR 4 alone. + // + // The ideal K is `blend_per_frame / capture_per_frame`. For 854x480 + // rgb48le with the more complex shaders this is ~910ms / ~175ms ≈ 5. + // K=4 strikes a perf vs. memory balance. Override via + // `HF_TRANSITION_RING_DEPTH` if a workload's blend/capture ratio is very + // different (simpler shaders that blend in ~100ms tolerate K=1-2 without + // perf loss). + const DEFAULT_TRANSITION_RING_DEPTH = 4; + const TRANSITION_RING_DEPTH = Math.max( + 1, + Number(process.env.HF_TRANSITION_RING_DEPTH ?? String(DEFAULT_TRANSITION_RING_DEPTH)), ); + const workerTransitionRings: Array = sessions.map(() => { + if (!hasTransitions) return null; + const ring: LayeredTransitionBuffers[] = []; + for (let k = 0; k < TRANSITION_RING_DEPTH; k++) { + ring.push({ + bufferA: Buffer.alloc(bufSize), + bufferB: Buffer.alloc(bufSize), + output: Buffer.alloc(bufSize), + }); + } + return ring; + }); const workerRanges = distributeLayeredHybridFrameRanges(totalFrames, activeWorkerCount); let framesWritten = 0; const reorderBuffer = createFrameReorderBuffer(0, totalFrames); @@ -185,15 +207,32 @@ export async function runHybridLayeredFrameLoop(input: HybridLoopInput): Promise const session = sessions[w]; const canvas = workerCanvases[w]; const range = workerRanges[w]; - const buffers = workerTransitionBuffers[w]; + const ring = workerTransitionRings[w]; if (!session || !canvas || !range) return; + // Per-ring-slot in-flight promise. When a slot is mid-blend, its + // promise is non-null; before reusing the slot for a new capture we + // await it so the buffer triple is free + the encoder has seen the + // earlier frame (writeEncoded gates ordering via the reorder buffer). + const ringInFlight: Array | null> = ring ? ring.map(() => null) : []; + let nextRingIdx = 0; for (let i = range.start; i < range.end; i++) { assertNotAborted(); const time = (i * job.config.fps.den) / job.config.fps.num; const activeTransition = transitionFramesSet.has(i) ? transitionRanges.find((t) => i >= t.startFrame && i <= t.endFrame) : undefined; - if (activeTransition && buffers) { + if (activeTransition && ring) { + // Pick the next ring slot. If it's still in flight from an earlier + // capture, await it to drain before reusing its buffer triple. + const slot = nextRingIdx; + nextRingIdx = (nextRingIdx + 1) % TRANSITION_RING_DEPTH; + const prev = ringInFlight[slot]; + if (prev) await prev; + const buffers = ring[slot]; + if (!buffers) continue; + // CAPTURE on the DOM worker (this thread). Fills bufferA/bufferB + // synchronously w.r.t. this loop — DOM work can't be pipelined + // because the per-worker browser session is single-threaded. await captureTransitionFrameOnWorker({ session, frameIdx: i, @@ -216,28 +255,48 @@ export async function runHybridLayeredFrameLoop(input: HybridLoopInput): Promise ? 1 : (i - activeTransition.startFrame) / (activeTransition.endFrame - activeTransition.startFrame); - if (poolRef) { - const blendStart = Date.now(); - const result = await poolRef.run({ - shader: activeTransition.shader, - bufferA: buffers.bufferA, - bufferB: buffers.bufferB, - output: buffers.output, - width, - height, - progress, - }); - buffers.bufferA = result.bufferA; - buffers.bufferB = result.bufferB; - buffers.output = result.output; - addHdrTiming(hdrPerf, "transitionCompositeMs", blendStart); - } else { - const transitionFn: TransitionFn = TRANSITIONS[activeTransition.shader] ?? crossfade; - const blendStart = Date.now(); - transitionFn(buffers.bufferA, buffers.bufferB, buffers.output, width, height, progress); - addHdrTiming(hdrPerf, "transitionCompositeMs", blendStart); - } - await writeEncoded(i, buffers.output); + // BLEND + ENCODE without awaiting. The promise drains back into + // `ringInFlight[slot]`; the next iteration that picks `slot` + // awaits it. The encoder reorder buffer fences ordering so out- + // of-order blend completion is fine. + const frameIdx = i; + const dispatch: Promise = (async () => { + if (poolRef) { + const blendStart = Date.now(); + const result = await poolRef.run({ + shader: activeTransition.shader, + bufferA: buffers.bufferA, + bufferB: buffers.bufferB, + output: buffers.output, + width, + height, + progress, + }); + buffers.bufferA = result.bufferA; + buffers.bufferB = result.bufferB; + buffers.output = result.output; + addHdrTiming(hdrPerf, "transitionCompositeMs", blendStart); + } else { + const transitionFn: TransitionFn = TRANSITIONS[activeTransition.shader] ?? crossfade; + const blendStart = Date.now(); + transitionFn( + buffers.bufferA, + buffers.bufferB, + buffers.output, + width, + height, + progress, + ); + addHdrTiming(hdrPerf, "transitionCompositeMs", blendStart); + } + await writeEncoded(frameIdx, buffers.output); + })(); + // Catch on a separate handle so an unhandled-rejection can't fire + // if no one awaits this slot before the worker exits. The error + // is re-thrown on the next await (slot reuse OR end-of-task drain). + ringInFlight[slot] = dispatch.catch((err: unknown) => { + throw err instanceof Error ? err : new Error(String(err)); + }); } else { const beforeCaptureHook = session.onBeforeCapture; let timingStart = Date.now(); @@ -268,6 +327,12 @@ export async function runHybridLayeredFrameLoop(input: HybridLoopInput): Promise await writeEncoded(i, canvas); } } + // Drain any pipelined blends still in flight on this worker before + // returning. If any rejected, the rejection bubbles here so + // `Promise.all` over `workerTaskOf` sees the failure. + for (const pending of ringInFlight) { + if (pending) await pending; + } }; await Promise.all(sessions.map((_, w) => workerTaskOf(w))); await reorderBuffer.waitForAllDone();