mirror of
https://github.com/heygen-com/hyperframes.git
synced 2026-09-10 22:20:14 +00:00
feat(engine,producer): verified interleaved parallel drawElement streaming (opt-in)
Step 2 of the DE engagement plan: multi-worker drawElement capture through the streaming encoder, with the full runtime self-verification net riding along — the confinement rule that kept the parallel clamp in place is now satisfied on this path. Opt-in via HF_DE_PARALLEL_STREAM=true; default routing (including the #2026 single-worker inversion) is unchanged. Mechanism: - distributeFramesInterleaved + WorkerTask.frameStride: worker i captures frames i, i+N, i+2N... — seek-based capture makes stride free and the ordered writer's reorder window shrinks from totalFrames/N to N (contiguous chunks serialize workers behind the writer). - Depth-2 pipelined worker-encode produce in the parallel worker loop (the same shape as the sequential loop; frame k's in-page encode overlaps k+stride's produce). HF_DE_PAR_DEBUG=1 traces the first frames per worker. - Drain guard extracted to createDrainFrameGuard (session-parameterized): every parallel frame gets the SAME blank-guard + PSNR self-verify as the sequential drain, against its owning worker's pre-injection ground truth (all sessions arm identical sample indices from CaptureOptions.compositionDurationSeconds). - FrameReorderBuffer.abort(err): a failed worker (e.g. verification error) rejects all parked and future waiters — without this, peers park forever in waitForFrame and the pool (which awaits ALL workers before surfacing errors) deadlocks. Found by the verify-trip test; unit-tested. - The typed DrawElementVerificationError is preserved past the pool's error-string flattening so the orchestrator's verify-retry recognizes it. - Static-dedup stride hazard fixed: lastEncodeResult reuse now requires EVERY frame in (lastEncodeResultFrame, i] to be predicted-static (sequential capture reduces to the old has(i) check). - Workers get separate browser PROCESSES under the flag: pages co-tenant in one browser starve non-active pages of BeginFrames on the paint-wait path (measured 86s vs 30s on a 3,245-frame rAF comp). Validation: - Happy path W3: verify samples pass across workers (4x inf on the 2,381f comp), output vs single-worker DE = 59.3dB (encode noise floor) — the interleave + dedup-stride produce identical pixels. - Verify-trip (marginal comp + HF_DE_VERIFY_MIN_DB=45): fails at frame 649 (32.2dB < 45), peers abort instead of deadlocking, whole render retries via parallel screenshot, RENDER_OK in 42.6s. - Canary suite 7/7 with the flag off (default paths untouched); producer orchestrator tests 99/99; engine suite 909 passed (14 pre-existing main failures, stash-A/B verified); reorder-buffer abort unit tests. Perf note: capture-only parallel speedup measured 1.38x (W2) / 1.52x (W3) over single-worker DE in the spike; end-to-end numbers on this machine are currently noisy (separate-browser init overhead + bench load) — clean benchmarks before any default routing change. The flag stays explicit opt-in; promoting it into the router replaces the #2026 W=1 pin for the same cohort. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Vance Ingalls
co-authored by
Claude Fable 5
Claude Fable 5
parent
df1d20b765
commit
b3493a7b61
@@ -15,6 +15,7 @@ import {
|
||||
initializeSession,
|
||||
closeCaptureSession,
|
||||
captureFrame,
|
||||
captureFrameToBufferPipelined,
|
||||
captureFrameToBuffer,
|
||||
getCapturePerfSummary,
|
||||
type CaptureSession,
|
||||
@@ -42,6 +43,13 @@ export interface WorkerTask {
|
||||
* calculation still uses the absolute frame index.
|
||||
*/
|
||||
outputFrameOffset?: number;
|
||||
/**
|
||||
* Frame stride for interleaved distribution (HF_DE_PARALLEL_STREAM spike):
|
||||
* the worker captures startFrame, startFrame+stride, … < endFrame. Default 1
|
||||
* (contiguous range). Interleaving keeps the ordered streaming writer's
|
||||
* reorder window at O(workerCount) frames instead of O(totalFrames/N).
|
||||
*/
|
||||
frameStride?: number;
|
||||
}
|
||||
|
||||
export interface WorkerResult {
|
||||
@@ -234,6 +242,34 @@ export function distributeFrames(
|
||||
return tasks;
|
||||
}
|
||||
|
||||
/**
|
||||
* Interleaved (round-robin) distribution: worker i captures frames
|
||||
* i, i+N, i+2N, …. Seek-based capture makes stride access free (every frame
|
||||
* is an absolute seek), and the streaming reorder window shrinks from
|
||||
* totalFrames/N to N — contiguous chunks serialize workers behind the
|
||||
* ordered writer (worker 1's first frame waits for ALL of worker 0's).
|
||||
* HF_DE_PARALLEL_STREAM spike; disk-path capture keeps contiguous chunks.
|
||||
*/
|
||||
export function distributeFramesInterleaved(
|
||||
totalFrames: number,
|
||||
workerCount: number,
|
||||
workDir: string,
|
||||
rangeStart: number = 0,
|
||||
): WorkerTask[] {
|
||||
const tasks: WorkerTask[] = [];
|
||||
for (let i = 0; i < workerCount && i < totalFrames; i++) {
|
||||
tasks.push({
|
||||
workerId: i,
|
||||
startFrame: rangeStart + i,
|
||||
endFrame: rangeStart + totalFrames,
|
||||
frameStride: workerCount,
|
||||
outputDir: join(workDir, `worker-${i}`),
|
||||
outputFrameOffset: rangeStart,
|
||||
});
|
||||
}
|
||||
return tasks;
|
||||
}
|
||||
|
||||
/**
|
||||
* Decide whether a parallel worker should run the per-worker SwiftShader
|
||||
* assertion. Gated to worker 0 only: workers within a chunk share the same
|
||||
@@ -244,24 +280,77 @@ export function shouldVerifyWorkerGpu(workerId: number, config?: Partial<EngineC
|
||||
return config?.browserGpuMode === "software" && workerId === 0;
|
||||
}
|
||||
|
||||
// fallow-ignore-next-line complexity
|
||||
async function captureFrameRange(
|
||||
session: CaptureSession,
|
||||
task: WorkerTask,
|
||||
captureOptions: CaptureOptions,
|
||||
signal: AbortSignal | undefined,
|
||||
onFrameCaptured: ((workerId: number, frameIndex: number) => void) | undefined,
|
||||
onFrameBuffer: ((frameIndex: number, buffer: Buffer) => Promise<void>) | undefined,
|
||||
onFrameBuffer:
|
||||
| ((frameIndex: number, buffer: Buffer, session: CaptureSession) => Promise<void>)
|
||||
| undefined,
|
||||
): Promise<number> {
|
||||
let framesCaptured = 0;
|
||||
const outputOffset = task.outputFrameOffset ?? 0;
|
||||
for (let i = task.startFrame; i < task.endFrame; i++) {
|
||||
const stride = task.frameStride ?? 1;
|
||||
// Depth-2 pipelined drawElement produce (HF_DE_PARALLEL_STREAM spike): frame
|
||||
// k's in-page worker encode overlaps frame k+stride's produce phase — the
|
||||
// same shape as the sequential worker-encode loop. Only engaged when the
|
||||
// session's encode worker initialized (drawElement mode) and frames stream
|
||||
// back via onFrameBuffer; the ordered writer's waitForFrame provides the
|
||||
// cross-worker backpressure (each worker runs at most `stride` frames ahead).
|
||||
if (onFrameBuffer && session.workerEncodeEnabled) {
|
||||
const dbg = process.env.HF_DE_PAR_DEBUG === "1";
|
||||
const dbgT0 = Date.now();
|
||||
const dbgWin = 40 * stride;
|
||||
let prev: { idx: number; encodeResult: Promise<Buffer> } | null = null;
|
||||
for (let i = task.startFrame; i < task.endFrame; i += stride) {
|
||||
if (signal?.aborted) throw new Error("Parallel worker cancelled");
|
||||
const time = (i * captureOptions.fps.den) / captureOptions.fps.num;
|
||||
if (dbg && i < task.startFrame + dbgWin) {
|
||||
console.log(`[par:w${task.workerId}] +${Date.now() - dbgT0}ms produce ${i} start`);
|
||||
}
|
||||
const { encodeResult } = await captureFrameToBufferPipelined(session, i - outputOffset, time);
|
||||
if (dbg && i < task.startFrame + dbgWin) {
|
||||
console.log(`[par:w${task.workerId}] +${Date.now() - dbgT0}ms produce ${i} kicked`);
|
||||
}
|
||||
if (prev) {
|
||||
if (dbg && prev.idx < task.startFrame + dbgWin) {
|
||||
console.log(
|
||||
`[par:w${task.workerId}] +${Date.now() - dbgT0}ms drain ${prev.idx} await-encode`,
|
||||
);
|
||||
}
|
||||
const buf = await prev.encodeResult;
|
||||
if (dbg && prev.idx < task.startFrame + dbgWin) {
|
||||
console.log(
|
||||
`[par:w${task.workerId}] +${Date.now() - dbgT0}ms drain ${prev.idx} encoded ${buf.length}B`,
|
||||
);
|
||||
}
|
||||
await onFrameBuffer(prev.idx, buf, session);
|
||||
if (dbg && prev.idx < task.startFrame + dbgWin) {
|
||||
console.log(`[par:w${task.workerId}] +${Date.now() - dbgT0}ms drain ${prev.idx} written`);
|
||||
}
|
||||
framesCaptured++;
|
||||
if (onFrameCaptured) onFrameCaptured(task.workerId, prev.idx);
|
||||
}
|
||||
prev = { idx: i, encodeResult };
|
||||
}
|
||||
if (prev) {
|
||||
await onFrameBuffer(prev.idx, await prev.encodeResult, session);
|
||||
framesCaptured++;
|
||||
if (onFrameCaptured) onFrameCaptured(task.workerId, prev.idx);
|
||||
}
|
||||
return framesCaptured;
|
||||
}
|
||||
for (let i = task.startFrame; i < task.endFrame; i += stride) {
|
||||
if (signal?.aborted) throw new Error("Parallel worker cancelled");
|
||||
const time = (i * captureOptions.fps.den) / captureOptions.fps.num;
|
||||
const fileFrameIdx = i - outputOffset;
|
||||
|
||||
if (onFrameBuffer) {
|
||||
const { buffer } = await captureFrameToBuffer(session, fileFrameIdx, time);
|
||||
await onFrameBuffer(i, buffer);
|
||||
await onFrameBuffer(i, buffer, session);
|
||||
} else {
|
||||
await captureFrame(session, fileFrameIdx, time);
|
||||
}
|
||||
@@ -278,7 +367,7 @@ async function executeWorkerTask(
|
||||
createBeforeCaptureHook: () => BeforeCaptureHook | null,
|
||||
signal?: AbortSignal,
|
||||
onFrameCaptured?: (workerId: number, frameIndex: number) => void,
|
||||
onFrameBuffer?: (frameIndex: number, buffer: Buffer) => Promise<void>,
|
||||
onFrameBuffer?: (frameIndex: number, buffer: Buffer, session: CaptureSession) => Promise<void>,
|
||||
config?: Partial<EngineConfig>,
|
||||
parallel?: boolean,
|
||||
): Promise<WorkerResult> {
|
||||
@@ -309,11 +398,19 @@ async function executeWorkerTask(
|
||||
createBeforeCaptureHook(),
|
||||
workerConfig,
|
||||
);
|
||||
if (process.env.HF_DE_PAR_DEBUG === "1") {
|
||||
console.log(`[par:w${task.workerId}] session created`);
|
||||
}
|
||||
// Worker-0-only SwiftShader assertion — see `shouldVerifyWorkerGpu` and #955.
|
||||
if (shouldVerifyWorkerGpu(task.workerId, workerConfig)) {
|
||||
await assertSwiftShader(session.page, readWebGlVendorInfoFromCanvas);
|
||||
}
|
||||
await initializeSession(session);
|
||||
if (process.env.HF_DE_PAR_DEBUG === "1") {
|
||||
console.log(
|
||||
`[par:w${task.workerId}] init done (mode=${session.captureMode} workerEncode=${session.workerEncodeEnabled === true})`,
|
||||
);
|
||||
}
|
||||
framesCaptured = await captureFrameRange(
|
||||
session,
|
||||
task,
|
||||
@@ -358,7 +455,7 @@ export async function executeParallelCapture(
|
||||
createBeforeCaptureHook: () => BeforeCaptureHook | null,
|
||||
signal?: AbortSignal,
|
||||
onProgress?: (progress: ParallelProgress) => void,
|
||||
onFrameBuffer?: (frameIndex: number, buffer: Buffer) => Promise<void>,
|
||||
onFrameBuffer?: (frameIndex: number, buffer: Buffer, session: CaptureSession) => Promise<void>,
|
||||
config?: Partial<EngineConfig>,
|
||||
): Promise<WorkerResult[]> {
|
||||
const totalFrames = tasks.reduce((sum, t) => sum + (t.endFrame - t.startFrame), 0);
|
||||
|
||||
Reference in New Issue
Block a user