mirror of
https://github.com/heygen-com/hyperframes.git
synced 2026-09-11 23:00:03 +00:00
fix(hyperframes-media): cap TTS synthesis concurrency instead of firing every line at once (#1862)
Two independent post-release feedback reports of the same mechanism from two different skills (both delegate to this one shared engine): audio.mjs fired every line's Kokoro TTS + whisper-transcribe subprocess concurrently via a bare Promise.all with no cap. - One OOM'd 12/13 lines on a resource-constrained laptop (32GB total, ~7GB free), requiring a manual patch to a sequential for-loop. - The other saw 7/8 lines fail on first run, then pass on retry once the model was cached — concurrent cold-start model loads overwhelming the machine, not a real synthesis failure. Kokoro/Whisper each load their own local model per subprocess, so firing every line at once multiplies that cost by the line count. Extracted the concurrency cap into lib/concurrency.mjs (audio.mjs is a script — it runs CLI/exit side effects on import, so it can't be unit-tested directly; the cap is small enough to pull out and test in isolation). Default 4, overridable via HYPERFRAMES_TTS_CONCURRENCY, floored at 1 (matching one report's own manual workaround). hyperframes-media/scripts/audio.mjs is the single canonical engine per its own header comment; product-launch-video, faceless-explainer, and pr-to-video each carry a thin wrapper that spawns this file as a subprocess (confirmed via their DEFAULT_ENGINE path), so this one fix covers all four skills without touching the other three. Tests: 4 new cases for mapWithConcurrency (order preserved regardless of completion order, cap actually enforced, limit > item count doesn't hang, empty input). Full skills test suite (514 tests) shows no new failures — the 444 pre-existing failures are environment-dependent and reproduce identically on unmodified main.
This commit is contained in:
@@ -38,8 +38,8 @@
|
|||||||
"files": 3
|
"files": 3
|
||||||
},
|
},
|
||||||
"hyperframes-media": {
|
"hyperframes-media": {
|
||||||
"hash": "096b2ab0a43b05dd",
|
"hash": "c991b6d3575e0f17",
|
||||||
"files": 40
|
"files": 42
|
||||||
},
|
},
|
||||||
"hyperframes-registry": {
|
"hyperframes-registry": {
|
||||||
"hash": "e3b389526834109d",
|
"hash": "e3b389526834109d",
|
||||||
|
|||||||
@@ -53,6 +53,7 @@ import {
|
|||||||
} from "./lib/tts.mjs";
|
} from "./lib/tts.mjs";
|
||||||
import { generateBgmDetached, inferBgmPrompt, retrieveBgm } from "./lib/bgm.mjs";
|
import { generateBgmDetached, inferBgmPrompt, retrieveBgm } from "./lib/bgm.mjs";
|
||||||
import { resolveSfx } from "./lib/sfx.mjs";
|
import { resolveSfx } from "./lib/sfx.mjs";
|
||||||
|
import { mapWithConcurrency } from "./lib/concurrency.mjs";
|
||||||
|
|
||||||
const HERE = dirname(fileURLToPath(import.meta.url));
|
const HERE = dirname(fileURLToPath(import.meta.url));
|
||||||
const argv = process.argv.slice(2);
|
const argv = process.argv.slice(2);
|
||||||
@@ -67,6 +68,16 @@ const die = (m) => {
|
|||||||
};
|
};
|
||||||
const r3 = (x) => Number(x.toFixed(3));
|
const r3 = (x) => Number(x.toFixed(3));
|
||||||
|
|
||||||
|
// Two independent reports of an unbounded Promise.all over TTS lines
|
||||||
|
// overwhelming a machine: one OOM'd 12/13 concurrent Kokoro TTS +
|
||||||
|
// whisper-transcribe lines on a resource-constrained laptop, the other saw
|
||||||
|
// 7/8 lines fail on first run (concurrent cold-start model loads) and pass on
|
||||||
|
// retry once the model was cached. Kokoro/Whisper each load their own local
|
||||||
|
// model per subprocess, so firing every line at once multiplies that cost by
|
||||||
|
// the line count. mapWithConcurrency caps how many run at once — still
|
||||||
|
// parallel, just bounded.
|
||||||
|
const ttsConcurrency = Math.max(1, Number(process.env.HYPERFRAMES_TTS_CONCURRENCY) || 4);
|
||||||
|
|
||||||
const hyperframesDir = resolve(flag("hyperframes", "."));
|
const hyperframesDir = resolve(flag("hyperframes", "."));
|
||||||
const requestPath = resolve(flag("request", join(hyperframesDir, "audio_request.json")));
|
const requestPath = resolve(flag("request", join(hyperframesDir, "audio_request.json")));
|
||||||
const outPath = resolve(flag("out", join(hyperframesDir, "audio_meta.json")));
|
const outPath = resolve(flag("out", join(hyperframesDir, "audio_meta.json")));
|
||||||
@@ -156,7 +167,7 @@ if (only.has("tts") && lines.length) {
|
|||||||
}
|
}
|
||||||
return { id, path: rel, duration_s: r3(dur), words: withWordIds(wordArr) };
|
return { id, path: rel, duration_s: r3(dur), words: withWordIds(wordArr) };
|
||||||
};
|
};
|
||||||
const results = await Promise.all(lines.map(synthLine));
|
const results = await mapWithConcurrency(lines, ttsConcurrency, synthLine);
|
||||||
voices = results.filter(Boolean);
|
voices = results.filter(Boolean);
|
||||||
for (const v of voices)
|
for (const v of voices)
|
||||||
console.error(` voice ${v.id}: ${v.path} (${v.duration_s}s, ${v.words.length} words)`);
|
console.error(` voice ${v.id}: ${v.path} (${v.duration_s}s, ${v.words.length} words)`);
|
||||||
|
|||||||
@@ -0,0 +1,14 @@
|
|||||||
|
// mapWithConcurrency — run `fn` over `items` with at most `limit` in flight at
|
||||||
|
// once. Preserves input order in the result array regardless of completion order.
|
||||||
|
export async function mapWithConcurrency(items, limit, fn) {
|
||||||
|
const results = new Array(items.length);
|
||||||
|
let next = 0;
|
||||||
|
async function worker() {
|
||||||
|
while (next < items.length) {
|
||||||
|
const i = next++;
|
||||||
|
results[i] = await fn(items[i], i);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
await Promise.all(Array.from({ length: Math.min(limit, items.length) }, worker));
|
||||||
|
return results;
|
||||||
|
}
|
||||||
@@ -0,0 +1,41 @@
|
|||||||
|
import { test } from "node:test";
|
||||||
|
import assert from "node:assert/strict";
|
||||||
|
import { mapWithConcurrency } from "./concurrency.mjs";
|
||||||
|
|
||||||
|
// Regression: audio.mjs used a bare Promise.all(lines.map(synthLine)) to
|
||||||
|
// synthesize every TTS line at once, spawning one Kokoro/whisper model load
|
||||||
|
// per line concurrently. Two independent reports of this overwhelming a
|
||||||
|
// machine (OOM, and cold-start contention causing spurious failures).
|
||||||
|
// mapWithConcurrency is the extracted cap; test it in isolation since
|
||||||
|
// audio.mjs itself is a script (runs CLI/exit side effects on import).
|
||||||
|
test("processes every item and preserves input order regardless of completion order", async () => {
|
||||||
|
const order = [5, 1, 3, 2, 4];
|
||||||
|
const results = await mapWithConcurrency(order, 2, async (n) => {
|
||||||
|
await new Promise((r) => setTimeout(r, n));
|
||||||
|
return n * 10;
|
||||||
|
});
|
||||||
|
assert.deepEqual(results, [50, 10, 30, 20, 40]);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("never runs more than `limit` at once", async () => {
|
||||||
|
let inFlight = 0;
|
||||||
|
let maxInFlight = 0;
|
||||||
|
const items = Array.from({ length: 10 }, (_, i) => i);
|
||||||
|
await mapWithConcurrency(items, 3, async () => {
|
||||||
|
inFlight++;
|
||||||
|
maxInFlight = Math.max(maxInFlight, inFlight);
|
||||||
|
await new Promise((r) => setTimeout(r, 5));
|
||||||
|
inFlight--;
|
||||||
|
});
|
||||||
|
assert.equal(maxInFlight, 3);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("limit larger than the item count runs everything without hanging", async () => {
|
||||||
|
const results = await mapWithConcurrency([1, 2], 10, async (n) => n * 2);
|
||||||
|
assert.deepEqual(results, [2, 4]);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("empty input resolves to an empty array", async () => {
|
||||||
|
const results = await mapWithConcurrency([], 4, async (n) => n);
|
||||||
|
assert.deepEqual(results, []);
|
||||||
|
});
|
||||||
Reference in New Issue
Block a user