diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index a967d5b555..10733fb88d 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -310,6 +310,7 @@ export { // ── Utilities ────────────────────────────────────────────────────────────────── export { quantizeTimeToFrame, MEDIA_VISUAL_STYLE_PROPERTIES } from "@hyperframes/core"; export { frameFileExtension } from "./services/frameCapture.js"; +export { appendSoftwareEncoderThreadsArg } from "./services/softwareEncoderThreads.js"; export { frameSequencePath } from "./services/extractedFrameIndex.js"; export type { MotionBlurOptions, MotionBlurBlendSpace } from "./services/motionBlur.js"; export type { PageMotionBlurOff } from "./types.js"; diff --git a/packages/engine/src/services/chunkEncoder.ts b/packages/engine/src/services/chunkEncoder.ts index 66d99147e0..898063ddb8 100644 --- a/packages/engine/src/services/chunkEncoder.ts +++ b/packages/engine/src/services/chunkEncoder.ts @@ -47,6 +47,7 @@ import { extractAudioMetadata } from "../utils/ffprobe.js"; import { type Fps, fpsToFfmpegArg, fpsToNumber } from "@hyperframes/core"; import type { EncoderOptions, EncodeResult, MuxResult } from "./chunkEncoder.types.js"; import { appendVp9CpuUsedArg } from "./vp9Options.js"; +import { appendSoftwareEncoderThreadsArg } from "./softwareEncoderThreads.js"; import { appendRenderProvenanceArgs } from "../utils/renderProvenance.js"; export type { EncoderOptions, EncodeResult, MuxResult } from "./chunkEncoder.types.js"; @@ -326,6 +327,7 @@ export function buildEncoderArgs( } else { const encoderName = codec === "h264" ? "libx264" : "libx265"; args.push("-c:v", encoderName, "-preset", preset); + appendSoftwareEncoderThreadsArg(args); if (bitrate) args.push("-b:v", bitrate); else args.push("-crf", String(quality)); diff --git a/packages/engine/src/services/softwareEncoderThreads.test.ts b/packages/engine/src/services/softwareEncoderThreads.test.ts new file mode 100644 index 0000000000..f82b46c5c7 --- /dev/null +++ b/packages/engine/src/services/softwareEncoderThreads.test.ts @@ -0,0 +1,98 @@ +import { spawnSync } from "node:child_process"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { getFfmpegBinary } from "../utils/ffmpegBinaries.js"; + +const host = vi.hoisted(() => ({ quota: 6, cores: 8 })); +vi.mock("node:os", async (importOriginal) => ({ + ...(await importOriginal()), + availableParallelism: () => host.quota, + cpus: () => Array.from({ length: host.cores }, () => ({}) as never), +})); + +import { buildEncoderArgs } from "./chunkEncoder.js"; +import type { EncoderOptions } from "./chunkEncoder.types.js"; +import { appendSoftwareEncoderThreadsArg } from "./softwareEncoderThreads.js"; +import { buildStreamingArgs } from "./streamingEncoder.js"; + +const FFMPEG = getFfmpegBinary(); +const HAS_TASKSET = process.platform === "linux" && spawnSync("taskset", ["-V"]).status === 0; +const HAS_FFMPEG = spawnSync(FFMPEG, ["-version"]).status === 0; +const QUIET = ["-v", "error"]; +const SOURCE = [ + "-f", + "lavfi", + "-i", + "testsrc2=size=320x240:rate=30:duration=2,noise=alls=12:allf=t", +]; + +const options = (codec: "h264" | "h265"): EncoderOptions => ({ + fps: { num: 30, den: 1 }, + width: 320, + height: 240, + codec, + preset: "medium", + quality: 23, +}); + +const threadsOf = (args: string[]) => args[args.indexOf("-threads") + 1]; + +describe("software encoder thread count", () => { + it("takes the CPU quota and floors at one", () => { + const args: string[] = []; + appendSoftwareEncoderThreadsArg(args, 6.9); + appendSoftwareEncoderThreadsArg(args, 0); + expect(args).toEqual(["-threads", "6", "-threads", "1"]); + }); + + it.each(["h264", "h265"] as const)( + "pins %s threads in both encoders to the quota, not the host", + (codec) => { + const seen = new Set(); + for (const cores of [8, 16, 64]) { + host.cores = cores; + seen.add(threadsOf(buildEncoderArgs(options(codec), SOURCE, "out.mp4"))); + seen.add( + threadsOf( + buildStreamingArgs({ ...options(codec), preset: "medium" } as never, "out.mp4"), + ), + ); + } + expect([...seen]).toEqual(["6"]); + }, + ); +}); + +describe.skipIf(!HAS_TASKSET || !HAS_FFMPEG)("encoded bytes across host sizes", () => { + const dirs: string[] = []; + afterEach(() => { + for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }); + }); + + /** Encode with the production args as a host with `cpus` cores would run them. */ + function videoMd5(codec: "h264" | "h265", cpus: string): string { + const dir = mkdtempSync(join(tmpdir(), "hf-threads-")); + dirs.push(dir); + const out = join(dir, "out.mp4"); + const args = buildEncoderArgs(options(codec), SOURCE, out); + expect(spawnSync("taskset", ["-c", cpus, FFMPEG, ...QUIET, ...args]).status).toBe(0); + const md5 = spawnSync( + FFMPEG, + ["-v", "error", "-i", out, "-map", "0:v", "-c", "copy", "-f", "md5", "-"], + { + encoding: "utf-8", + }, + ); + return md5.stdout.trim(); + } + + it.each(["h264", "h265"] as const)( + "gives %s identical bytes on a 2-core and a 7-core host", + (codec) => { + expect(videoMd5(codec, "0-1")).toBe(videoMd5(codec, "0-6")); + }, + 120_000, + ); +}); diff --git a/packages/engine/src/services/softwareEncoderThreads.ts b/packages/engine/src/services/softwareEncoderThreads.ts new file mode 100644 index 0000000000..e767a0a4a0 --- /dev/null +++ b/packages/engine/src/services/softwareEncoderThreads.ts @@ -0,0 +1,9 @@ +import { availableParallelism } from "node:os"; + +/** x264/x265 bitstreams depend on the thread count, so it comes from the CPU quota, never ffmpeg's host-core guess. */ +export function appendSoftwareEncoderThreadsArg( + args: string[], + parallelism: number = availableParallelism(), +): void { + args.push("-threads", String(Math.max(1, Math.floor(parallelism)))); +} diff --git a/packages/engine/src/services/streamingEncoder.ts b/packages/engine/src/services/streamingEncoder.ts index 0e00b4d215..1b6f39b7d8 100644 --- a/packages/engine/src/services/streamingEncoder.ts +++ b/packages/engine/src/services/streamingEncoder.ts @@ -46,6 +46,7 @@ import { import { DEFAULT_CONFIG, type EngineConfig } from "../config.js"; import { fpsToFfmpegArg, fpsToNumber, type Fps } from "@hyperframes/core"; import { appendVp9CpuUsedArg } from "./vp9Options.js"; +import { appendSoftwareEncoderThreadsArg } from "./softwareEncoderThreads.js"; import { appendRenderProvenanceArgs } from "../utils/renderProvenance.js"; import { appendLockedGopArgs, lockedGopCodecParams, resolveLockedGopSize } from "./chunkEncoder.js"; @@ -338,6 +339,7 @@ export function buildStreamingArgs( } else { const encoderName = codec === "h264" ? "libx264" : "libx265"; args.push("-c:v", encoderName, "-preset", preset); + appendSoftwareEncoderThreadsArg(args); if (bitrate) args.push("-b:v", bitrate); else args.push("-crf", String(quality)); diff --git a/packages/producer/src/services/distributed/assemble.test.ts b/packages/producer/src/services/distributed/assemble.test.ts index 65711d2946..436f60fff1 100644 --- a/packages/producer/src/services/distributed/assemble.test.ts +++ b/packages/producer/src/services/distributed/assemble.test.ts @@ -26,7 +26,7 @@ import { statSync, writeFileSync, } from "node:fs"; -import { tmpdir } from "node:os"; +import { availableParallelism, tmpdir } from "node:os"; import { join } from "node:path"; import { applyFaststart, @@ -38,7 +38,7 @@ import { } from "@hyperframes/engine"; import { padOrTrimAudioToVideoFrameCount } from "../render/audioPadTrim.js"; import type { ChunkSliceJson } from "../render/stages/freezePlan.js"; -import { assemble, cfrRateControlArgs } from "./assemble.js"; +import { assemble, cfrEncodeArgs, cfrRateControlArgs } from "./assemble.js"; import type { DistributedFormat } from "./shared.js"; let runRoot: string; @@ -663,6 +663,11 @@ describe("assemble()", () => { TIMEOUT_MS, ); + it("cfr re-encode pins x264 threads to the CPU quota", () => { + const args = cfrEncodeArgs(["-i", "chunk.mp4"], {}, "30"); + expect(args[args.indexOf("-threads") + 1]).toBe(String(availableParallelism())); + }); + it("cfr re-encode keeps the chunks' rate control", () => { expect(cfrRateControlArgs({})).toEqual(["-preset", "medium", "-crf", "18"]); expect(cfrRateControlArgs({ quality: "high" })).toEqual(["-preset", "slow", "-crf", "15"]); diff --git a/packages/producer/src/services/distributed/assemble.ts b/packages/producer/src/services/distributed/assemble.ts index bda1533755..85e05e17ea 100644 --- a/packages/producer/src/services/distributed/assemble.ts +++ b/packages/producer/src/services/distributed/assemble.ts @@ -40,6 +40,7 @@ import { muxVideoWithAudio, resolveConfig, runFfmpeg, + appendSoftwareEncoderThreadsArg, } from "@hyperframes/engine"; import { fpsToFfmpegArg } from "@hyperframes/core"; import { defaultLogger, type ProducerLogger } from "../../logger.js"; @@ -267,6 +268,27 @@ function videoInputArgs(ctx: JoinContext, videoSources: readonly string[]): stri : concatDemuxerInput(videoSources, ctx.workDir, ctx.fpsArg); } +export function cfrEncodeArgs( + inputArgs: string[], + encoder: CfrRateControlSource, + fpsArg: string, +): string[] { + const args = [ + ...inputArgs, + "-c:v", + "libx264", + ...cfrRateControlArgs(encoder), + "-pix_fmt", + "yuv420p", + "-fps_mode", + "cfr", + "-r", + fpsArg, + ]; + appendSoftwareEncoderThreadsArg(args); + return args; +} + // ── 2c. Optional exact-CFR re-encode ────────────────────────────────── // Lands the stream-level `avg_frame_rate` (PTS-derived under stream copy) // on the requested rational, for consumers that strict-check it. @@ -276,18 +298,7 @@ async function reencodeCfr( encoder: CfrRateControlSource, ): Promise { const cfrOutputPath = join(ctx.workDir, `cfr.${ctx.format}`); - const cfrArgs = [ - ...videoInputArgs(ctx, chunkPaths), - "-c:v", - "libx264", - ...cfrRateControlArgs(encoder), - "-pix_fmt", - "yuv420p", - "-fps_mode", - "cfr", - "-r", - ctx.fpsArg, - ]; + const cfrArgs = cfrEncodeArgs(videoInputArgs(ctx, chunkPaths), encoder, ctx.fpsArg); appendRenderProvenanceArgs(cfrArgs, cfrOutputPath); cfrArgs.push("-y", cfrOutputPath); await ctx.timed("cfrMs", () => diff --git a/scripts/package-size-budgets.json b/scripts/package-size-budgets.json index 60136d0215..571193f22d 100644 --- a/scripts/package-size-budgets.json +++ b/scripts/package-size-budgets.json @@ -12,7 +12,7 @@ "unpacked": 3950000 }, "@hyperframes/engine": { - "packed": 428347, + "packed": 428562, "unpacked": 1436293 }, "@hyperframes/gcp-cloud-run": {