fix(goLive): deterministic token-bucket video pacing at demuxer (replace unreliable -re)
Root cause (3rd iteration): ffmpeg '-re' on the demuxer does NOT reliably throttle a multi-stage live pipe (merge ffmpeg -> encoder x264 -> NUT -> demuxer). In production the demuxer still emitted ~240fps while the WebRTC sender consumed 30fps, building a 100k+ frame backlog (observed: frames=197490 vs sent #24600, ~8.4 min in). The sender always emitted the OLDEST buffered frame -> video frozen ~10 min behind live, while audio (tiny, jitter-buffer recovered) stayed smooth. Local file/pipe tests showed -re working (30fps) but the live YouTube/WebM pipeline did not — -re is not trustworthy here. Fix: enforce 1x video output with a token-bucket limiter in the demuxer (Node side), independent of ffmpeg. Capacity = 1s of frames, refill 1 token per 1000/fps ms. Surplus non-key frames are DROPPED (never buffered) so the sender always emits the newest frame; keyframes are forced through even over budget so the decoder keeps a fresh IDR. The limiter does NOT stall the ffmpeg process (unlike the earlier proc.stdout pause), so audio on fd3 keeps flowing. Verified: tsc --noEmit clean.
This commit is contained in:
@@ -377,18 +377,16 @@ export async function demux(
|
|||||||
// NALs and flush one frame per slice, prepending the parameter sets that
|
// NALs and flush one frame per slice, prepending the parameter sets that
|
||||||
// precede it, and timestamp it as ONE frame at the video frame rate.
|
// precede it, and timestamp it as ONE frame at the video frame rate.
|
||||||
|
|
||||||
// PACING: ffmpeg is spawned with `-re` (see spawn args) so it reads the
|
// PACING: a token-bucket rate limiter enforces 1x video output at the
|
||||||
// piped NUT input at 1x wall-clock. That back-pressures the whole upstream
|
// encoder's frame rate, independent of how fast ffmpeg produces frames.
|
||||||
// chain (encoder x264 → merge ffmpeg → yt-dlp download) through their OS
|
// ffmpeg `-re` was tried here (see spawn args) but does NOT reliably
|
||||||
// pipes, pinning production at real-time — the old design (no throttle)
|
// throttle a multi-stage live pipe (encoder x264 -> NUT -> demuxer): in
|
||||||
// had the encoder burst ~10x faster than the 30fps WebRTC sender, so the
|
// production the demuxer still emitted ~240fps while the sender consumed
|
||||||
// demuxer buffered an unbounded backlog and the sender always emitted the
|
// 30fps, building a 100k+ frame backlog → video frozen on the oldest
|
||||||
// OLDEST frames → frozen/laggy video while audio (tiny, jitter-buffer
|
// buffered frame while audio (tiny, jitter-buffer recoverable) stayed
|
||||||
// recoverable) stayed smooth. That is the "video stuck, voice normal"
|
// smooth. A Node-level limiter is deterministic and never stalls the
|
||||||
// symptom. The first fix paused proc.stdout when vPipe was full, but that
|
// ffmpeg process, so audio on fd3 keeps flowing. Surplus video frames are
|
||||||
// stalled fd3 audio too (same process) → audio patah-patah, and the
|
// DROPPED (not buffered) so the sender always emits the newest frame.
|
||||||
// accumulated backlog never drained → permanent ~8s lag. `-re` + bounded
|
|
||||||
// drop is the correct throttle.
|
|
||||||
let videoBuf = Buffer.alloc(0);
|
let videoBuf = Buffer.alloc(0);
|
||||||
let frameCount = 0;
|
let frameCount = 0;
|
||||||
let pendingNals: Buffer[] = [];
|
let pendingNals: Buffer[] = [];
|
||||||
@@ -399,6 +397,25 @@ export async function demux(
|
|||||||
// setting); it drives both RTP timestamp advance and pacing.
|
// setting); it drives both RTP timestamp advance and pacing.
|
||||||
const videoFps =
|
const videoFps =
|
||||||
opts.frameRate ?? (vInfo.framerate_num / vInfo.framerate_den || 30);
|
opts.frameRate ?? (vInfo.framerate_num / vInfo.framerate_den || 30);
|
||||||
|
// token-bucket: capacity = 1s of frames; refill one token per 1000/fps ms.
|
||||||
|
const tokenIntervalMs = 1000 / videoFps;
|
||||||
|
let tokens = videoFps; // start with 1s of credit
|
||||||
|
let lastToken = Date.now();
|
||||||
|
let droppedFrames = 0;
|
||||||
|
|
||||||
|
const tryConsume = (): boolean => {
|
||||||
|
const now = Date.now();
|
||||||
|
const elapsed = now - lastToken;
|
||||||
|
if (elapsed >= tokenIntervalMs) {
|
||||||
|
tokens = Math.min(videoFps, tokens + Math.floor(elapsed / tokenIntervalMs));
|
||||||
|
lastToken = now - (elapsed % tokenIntervalMs);
|
||||||
|
}
|
||||||
|
if (tokens >= 1) {
|
||||||
|
tokens -= 1;
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
return false;
|
||||||
|
};
|
||||||
|
|
||||||
const flushAccessUnit = () => {
|
const flushAccessUnit = () => {
|
||||||
if (pendingNals.length === 0) return;
|
if (pendingNals.length === 0) return;
|
||||||
@@ -412,16 +429,17 @@ export async function demux(
|
|||||||
pendingNals = [];
|
pendingNals = [];
|
||||||
pendingHasSlice = false;
|
pendingHasSlice = false;
|
||||||
pendingIsKey = false;
|
pendingIsKey = false;
|
||||||
// Bounded backlog: if the sender ever stalls (slow network, CPU spike),
|
// Drop surplus frames: if no token is available, discard this AU so the
|
||||||
// drop the oldest queued frame instead of letting vPipe grow. The
|
// sender never emits a stale (old) frame. Keyframes are forced through
|
||||||
// WebRTC sender paces 30fps from the newest frames; ffmpeg's `-re` (see
|
// even when over budget so the decoder always has a fresh IDR to recover.
|
||||||
// spawn args) keeps production at 1x so this only triggers on stalls —
|
if (!isKey && !tryConsume()) {
|
||||||
// never during normal playback. Without it, a transient sender stall
|
droppedFrames++;
|
||||||
// becomes a permanent multi-second lag (sender emits oldest frames
|
if (droppedFrames === 1 || droppedFrames % 300 === 0) {
|
||||||
// forever). NEVER pause proc.stdout here: fd3 audio shares the same
|
console.log(
|
||||||
// process, so pausing video output also stalls audio (patah-patah).
|
`[goLive:Demuxer] drop=${droppedFrames} (pacing ${videoFps}fps; encoder burst)`,
|
||||||
if (vPipe.readableLength >= 30) {
|
);
|
||||||
vPipe.read(); // drop the oldest queued frame
|
}
|
||||||
|
return;
|
||||||
}
|
}
|
||||||
vPipe.write({
|
vPipe.write({
|
||||||
data: au,
|
data: au,
|
||||||
|
|||||||
Reference in New Issue
Block a user