Remove a frame of latency and (maybe) fix b-frames. (#721)

Author: kixelated

js/hang/src/watch/video/source.ts

@@ -3,7 +3,7 @@ import { Effect, Signal } from "@kixelated/signals";
 import type * as Catalog from "../../catalog";
 import * as Frame from "../../frame";
 import { PRIORITY } from "../../publish/priority";
-import * as Time from "../../time";
+import type * as Time from "../../time";
 import * as Hex from "../../util/hex";
 
 export type SourceProps = {
@@ -33,10 +33,16 @@ type SyncStatus = {
 	bufferDuration?: number;
 };
 
+// Only count it as buffering if we had to sleep for 200ms or more before rendering the next frame.
+// Unfortunately, this has to be quite high because of b-frames.
+// TODO Maybe we need to detect b-frames and make this dynamic?
+const MIN_SYNC_WAIT_MS = 200 as Time.Milli;
+
+// The maximum number of concurrent b-frames that we support.
+const MAX_BFRAMES = 10;
+
 // Responsible for switching between video tracks and buffering frames.
 export class Source {
-	#MIN_SYNC_WAIT_MS = 50 as Time.Milli;
-
 	broadcast: Signal<Moq.Broadcast | undefined>;
 	enabled: Signal<boolean>; // Don't download any longer
 
@@ -59,12 +65,10 @@ export class Source {
 	// Used as a tiebreaker when there are multiple tracks (HD vs SD).
 	target = new Signal<Target | undefined>(undefined);
 
-	// Unfortunately, browsers don't let us hold on to multiple VideoFrames.
-	// ex. Firefox only allows 2 outstanding VideoFrames at a time.
-	// In order to semi-support b-frames, we buffer two frames and expose the earliest one.
+	// Expose the current frame to render as a signal
 	frame = new Signal<VideoFrame | undefined>(undefined);
-	#next?: VideoFrame;
 
+	// The target latency in milliseconds.
 	latency: Signal<Time.Milli>;
 
 	// The display size of the video in pixels, ideally sourced from the catalog.
@@ -76,11 +80,7 @@ export class Source {
 	// Used to convert PTS to wall time.
 	#reference: DOMHighResTimeStamp | undefined;
 
-	// The latency after we've accounted for the extra frame buffering and jitter buffer.
-	#jitter: Signal<Time.Milli>;
-
 	bufferStatus = new Signal<BufferStatus>({ state: "empty" });
-
 	syncStatus = new Signal<SyncStatus>({ state: "ready" });
 
 	#signals = new Effect();
@@ -94,10 +94,6 @@ export class Source {
 		this.latency = Signal.from(props?.latency ?? (100 as Time.Milli));
 		this.enabled = Signal.from(props?.enabled ?? false);
 
-		// We subtract a frame from the jitter buffer to account for the extra buffered frame.
-		// Assume 30fps by default.
-		this.#jitter = new Signal(Math.max(0, this.latency.peek() - 33) as Time.Milli);
-
 		this.#signals.effect((effect) => {
 			const c = effect.get(catalog)?.video;
 			effect.set(this.catalog, c);
@@ -108,7 +104,6 @@ export class Source {
 		this.#signals.effect(this.#runSelected.bind(this));
 		this.#signals.effect(this.#runPending.bind(this));
 		this.#signals.effect(this.#runDisplay.bind(this));
-		this.#signals.effect(this.#runJitter.bind(this));
 		this.#signals.effect(this.#runBuffer.bind(this));
 	}
 
@@ -170,9 +165,6 @@ export class Source {
 				return undefined;
 			});
 
-			this.#next?.close();
-			this.#next = undefined;
-
 			return;
 		}
 
@@ -192,62 +184,89 @@ export class Source {
 
 		// Create consumer that reorders groups/frames up to the provided latency.
 		const consumer = new Frame.Consumer(sub, {
-			latency: this.#jitter,
+			latency: this.latency,
 		});
 		effect.cleanup(() => consumer.close());
 
+		// We need a queue because VideoDecoder doesn't block on a Promise returned by output.
+		// NOTE: We will drain this queue almost immediately, so the highWaterMark is just a safety net.
+		const queue = new TransformStream<VideoFrame, VideoFrame>(
+			undefined,
+			{ highWaterMark: MAX_BFRAMES },
+			{ highWaterMark: MAX_BFRAMES },
+		);
+
+		const writer = queue.writable.getWriter();
+		effect.cleanup(() => writer.close());
+
+		const reader = queue.readable.getReader();
+		effect.cleanup(() => reader.cancel());
+
 		const decoder = new VideoDecoder({
-			output: (frame) => {
-				// Keep track of the two newest frames.
-				// this.frame is older than this.#next, if it exists.
-				const prev = this.frame.peek();
-				if (prev && prev.timestamp >= frame.timestamp) {
-					// NOTE: This can happen if you have more than 1 b-frame in a row.
-					// Sorry, blame Firefox.
+			output: async (frame: VideoFrame) => {
+				// Insert into a queue so we can perform ordered sleeps.
+				// If this were to block, I believe WritableStream is still ordered.
+				try {
+					await writer.write(frame);
+				} catch {
 					frame.close();
-					return;
 				}
+			},
+			// TODO bubble up error
+			error: (error) => {
+				console.error(error);
+				effect.close();
+			},
+		});
+		effect.cleanup(() => decoder.close());
 
-				if (!prev) {
-					// As time-to-video optimization, use the first frame we see.
-					// We know this is an i-frame so there's no need to re-order it.
-					this.frame.set(frame);
-					return;
-				}
+		effect.spawn(async () => {
+			for (;;) {
+				const { value: frame } = await reader.read();
+				if (!frame) break;
+
+				// Sleep until it's time to decode the next frame.
+				const ref = performance.now() - frame.timestamp / 1000;
 
-				// If jitter is 0, then we disable buffering frames.
-				const jitter = this.#jitter.peek();
-				if (jitter === 0) {
-					prev.close();
-					this.frame.set(frame);
-					return;
+				let sleep = 0;
+				if (!this.#reference || ref < this.#reference) {
+					this.#reference = ref;
+					// Don't sleep so we immediately render this frame.
+				} else {
+					sleep = this.#reference - ref + this.latency.peek();
 				}
 
-				if (!this.#next) {
-					// We know we're newer than the current frame, so buffer it.
-					this.#next = frame;
-					return;
+				if (sleep > MIN_SYNC_WAIT_MS) {
+					this.syncStatus.set({ state: "wait", bufferDuration: sleep });
 				}
 
-				// Close the previous frame, and check if we need to replace #next or this.frame.
-				prev.close();
+				if (sleep > 0) {
+					// NOTE: WebCodecs doesn't block on output promises (I think?), so these sleeps will occur concurrently.
+					// TODO: This cause the `syncStatus` to be racey especially
+					await new Promise((resolve) => setTimeout(resolve, sleep));
+				}
 
-				if (this.#next.timestamp < frame.timestamp) {
-					// Replace #next with the new frame.
-					this.frame.set(this.#next);
-					this.#next = frame;
+				if (sleep > MIN_SYNC_WAIT_MS) {
+					// Include how long we slept if it was above the threshold.
+					this.syncStatus.set({ state: "ready", bufferDuration: sleep });
 				} else {
-					// #next is newer than this new frame, so keep it.
-					this.frame.set(frame);
+					this.syncStatus.set({ state: "ready" });
+
+					// If the track switch was pending, complete it now.
+					if (this.#pending === effect) {
+						this.#active?.close();
+						this.#active = effect;
+						this.#pending = undefined;
+						effect.set(this.active, name);
+					}
 				}
-			},
-			// TODO bubble up error
-			error: (error) => {
-				console.error(error);
-				effect.close();
-			},
+
+				this.frame.update((prev) => {
+					prev?.close();
+					return frame;
+				});
+			}
 		});
-		effect.cleanup(() => decoder.close());
 
 		decoder.configure({
 			...config,
@@ -262,47 +281,6 @@ export class Source {
 				const next = await Promise.race([consumer.decode(), effect.cancel]);
 				if (!next) break;
 
-				// See if we can upgrade ourselves to the active track once we catch up.
-				// TODO: This is a racey when latency === 0, but I think it's fine.
-				const prev = this.frame.peek();
-				if (this.#pending === effect && (!prev || next.timestamp > prev.timestamp)) {
-					this.#active?.close();
-					this.#active = effect;
-					this.#pending = undefined;
-					effect.set(this.active, name);
-				}
-
-				// Sleep until it's time to decode the next frame.
-				const ref = performance.now() - Time.Milli.fromMicro(next.timestamp);
-
-				if (!this.#reference || ref < this.#reference) {
-					this.#reference = ref;
-				} else {
-					const sleep = this.#reference - ref + this.#jitter.peek();
-					const isWaitRequired = sleep >= this.#MIN_SYNC_WAIT_MS;
-					if (sleep > 0) {
-						// The planned jitter buffer size
-						const bufferDuration: Time.Milli = this.#jitter.peek();
-
-						if (isWaitRequired) {
-							this.syncStatus.set({ state: "wait", bufferDuration });
-						}
-
-						await new Promise((resolve) => setTimeout(resolve, sleep));
-
-						if (isWaitRequired) {
-							this.syncStatus.set({ state: "ready", bufferDuration });
-						}
-					} else {
-						this.syncStatus.set({ state: "ready" });
-					}
-				}
-
-				if (decoder.state === "closed") {
-					// Closed during the sleep
-					break;
-				}
-
 				const chunk = new EncodedVideoChunk({
 					type: next.keyframe ? "key" : "delta",
 					data: next.data,
@@ -373,48 +351,23 @@ export class Source {
 	}
 
 	#runBuffer(effect: Effect): void {
-		const currentFrame = effect.get(this.frame);
-		const nextFrame = this.#next;
+		const frame = effect.get(this.frame);
 		const enabled = effect.get(this.enabled);
 
-		const isBufferEmpty = enabled && !currentFrame && !nextFrame;
-
+		const isBufferEmpty = enabled && !frame;
 		if (isBufferEmpty) {
 			this.bufferStatus.set({ state: "empty" });
 		} else {
 			this.bufferStatus.set({ state: "filled" });
 		}
 	}
 
-	#runJitter(effect: Effect): void {
-		const config = effect.get(this.#selectedConfig);
-		if (!config) return;
-
-		// Use the framerate to compute the jitter buffer size.
-		// We always buffer a single frame, so subtract that from the jitter buffer.
-		const fps = config.framerate && config.framerate > 0 ? config.framerate : 30;
-		const delay = 1000 / fps;
-		const latency = effect.get(this.latency);
-
-		const jitter = Math.max(0, latency - delay) as Time.Milli;
-		this.#jitter.set(jitter);
-
-		// If we're not buffering any frames, then close the next frame.
-		if (jitter === 0 && this.#next) {
-			this.#next.close();
-			this.#next = undefined;
-		}
-	}
-
 	close() {
 		this.frame.update((prev) => {
 			prev?.close();
 			return undefined;
 		});
 
-		this.#next?.close();
-		this.#next = undefined;
-
 		this.#signals.close();
 	}
 }