From f723a8821bc530ad33e3d4528beb88c649cff1f2 Mon Sep 17 00:00:00 2001 From: joaner <1726541+joaner@users.noreply.github.com> Date: Mon, 13 Jul 2026 20:53:39 +0800 Subject: [PATCH 1/2] perf: improve H264 playback on low-end machines Decouple playback clock from MCAP prefetch, add adaptive H264 backpressure with IDR resync, hardware-preferring WebCodecs config, and benchmark tooling so progress and video stay responsive under decode pressure. --- docs/H264_PERFORMANCE.md | 50 ++ package.json | 1 + scripts/benchmark-h264.mjs | 415 ++++++++++++++ src/core/players/IterablePlayer.test.ts | 515 +++++++++++++++++- src/core/players/IterablePlayer.ts | 422 ++++++++------ src/features/panels/Image/ImagePanel.tsx | 15 +- .../panels/Image/core/ImageRender.worker.ts | 421 ++++++++++++-- src/features/panels/Image/core/h264.test.ts | 47 +- src/features/panels/Image/core/h264.ts | 64 ++- .../Image/core/h264Backpressure.test.ts | 56 ++ .../panels/Image/core/h264Backpressure.ts | 79 +++ .../panels/Image/core/h264Queue.test.ts | 128 +++++ src/features/panels/Image/core/h264Queue.ts | 121 ++++ .../panels/Image/core/h264SeekRepair.test.ts | 119 +++- .../panels/Image/core/h264SeekRepair.ts | 29 +- .../panels/Image/core/imageTypes.test.ts | 17 +- src/features/panels/Image/core/imageTypes.ts | 27 +- .../panels/Image/core/imageWorkerProtocol.ts | 24 +- .../Image/core/messageFrameAdapter.test.ts | 21 + .../panels/Image/core/messageFrameAdapter.ts | 13 +- tests/image-h264.spec.ts | 31 +- 21 files changed, 2348 insertions(+), 267 deletions(-) create mode 100644 docs/H264_PERFORMANCE.md create mode 100644 scripts/benchmark-h264.mjs create mode 100644 src/features/panels/Image/core/h264Backpressure.test.ts create mode 100644 src/features/panels/Image/core/h264Backpressure.ts create mode 100644 src/features/panels/Image/core/h264Queue.test.ts create mode 100644 src/features/panels/Image/core/h264Queue.ts diff --git a/docs/H264_PERFORMANCE.md b/docs/H264_PERFORMANCE.md new file mode 100644 index 0000000..202d2e5 --- /dev/null +++ b/docs/H264_PERFORMANCE.md @@ -0,0 +1,50 @@ +# H.264 playback performance + +Use a representative, unmodified MCAP from the target workload rather than only the generated +`public/examples/test_h264.mcap`. Record its size, duration, H.264 profile/level, resolution, frame +rate, keyframe interval, and image topic. Do not copy or rewrite the recording for a benchmark. + +## Automated Chromium benchmark + +Start a production preview (`npm run build && npm run preview`), then run: + +```bash +npm run benchmark:h264 -- \ + --file /absolute/path/representative.mcap \ + --base-url http://127.0.0.1:4173 \ + --duration 60 \ + --output benchmark-h264.json +``` + +The benchmark exposes the original file through a temporary read-only Range/CORS server. It records +progress updates, random seeks, Image panel H.264 metrics, decode/page/console errors, and Chromium +JS heap when available. Run `npm run benchmark:h264 -- --help` for all options. + +## Manual browser pass + +Expose the same original MCAP through a read-only HTTP server with byte Range and CORS support. In +current stable Chrome, Edge, and Firefox, open +`http://127.0.0.1:4173/?url=`, play for 60 seconds, and seek to three +non-keyframe positions. Confirm that progress remains responsive, frames resume after each seek, and +no decode failure appears. Firefox H.264/WebCodecs availability depends on the OS codec stack; record +an unsupported-codec result separately instead of comparing it as a performance regression. + +Suggested acceptance targets: + +- The E2E smoke test observes at least two progress advances and at least 1 percentage point of + movement in about one second. +- No decode, page, or console errors; Image metrics are non-negative and pressure is `normal`, + `degraded`, or `recovery`. +- The H.264 pending queue is hard-bounded at 120 frames and a 1,000 ms media-time span. Soft + pressure keeps a sole complete GOP intact; if either hard bound is exceeded without a newer IDR + suffix that fits, the worker keeps the current picture, drops the complete backlog, and waits for + the next real IDR rather than decoding a truncated delta chain. Playback must recover after every + seek. +- For a 60-second run, progress updates on at least 50% of 200 ms samples and the longest unexplained + stall is below 1 second. +- After warm-up, JS heap does not grow continuously; investigate growth above 25% or 256 MiB. Treat + these as regression gates against a saved baseline, not universal limits. + +Record browser/version, OS, CPU, GPU, RAM, display resolution, power mode, hardware acceleration +setting, cold/warm run, file metadata, and the JSON output. Compare results only on equivalent +hardware and browser settings. diff --git a/package.json b/package.json index f55c1eb..a6886fd 100644 --- a/package.json +++ b/package.json @@ -65,6 +65,7 @@ "test": "vitest run", "preview": "npm run build && vite preview", "gen:e2e:fixtures": "node scripts/gen-e2e-fixtures.mjs", + "benchmark:h264": "node scripts/benchmark-h264.mjs", "pretest:e2e": "npm run gen:e2e:fixtures", "test:e2e": "playwright test" }, diff --git a/scripts/benchmark-h264.mjs b/scripts/benchmark-h264.mjs new file mode 100644 index 0000000..c6d2c7e --- /dev/null +++ b/scripts/benchmark-h264.mjs @@ -0,0 +1,415 @@ +#!/usr/bin/env node +import { createReadStream } from 'node:fs'; +import { stat, writeFile } from 'node:fs/promises'; +import http from 'node:http'; +import path from 'node:path'; +import process from 'node:process'; +import { chromium } from '@playwright/test'; + +const DEFAULT_BASE_URL = 'http://127.0.0.1:4173'; +const DEFAULT_DURATION_SECONDS = 15; +const SAMPLE_INTERVAL_MS = 200; + +function usage() { + return `Usage: + npm run benchmark:h264 -- --file /absolute/path/recording.mcap [options] + +Options: + --file Absolute path to a local MCAP file (required) + --base-url Running rosview URL (default: ${DEFAULT_BASE_URL}) + --duration Benchmark duration in seconds (default: ${DEFAULT_DURATION_SECONDS}) + --output Also write the JSON result to this path + --help Show this help +`; +} + +function parseArgs(argv) { + const options = { + baseUrl: DEFAULT_BASE_URL, + durationSeconds: DEFAULT_DURATION_SECONDS, + file: undefined, + output: undefined, + help: false, + }; + const valueOptions = new Map([ + ['--file', 'file'], + ['--base-url', 'baseUrl'], + ['--duration', 'durationSeconds'], + ['--output', 'output'], + ]); + + for (let index = 2; index < argv.length; index += 1) { + const argument = argv[index]; + if (argument === '--help' || argument === '-h') { + options.help = true; + continue; + } + const separator = argument.indexOf('='); + const name = separator >= 0 ? argument.slice(0, separator) : argument; + const key = valueOptions.get(name); + if (!key) { + throw new Error(`Unknown option: ${argument}`); + } + const value = separator >= 0 ? argument.slice(separator + 1) : argv[++index]; + if (!value || value.startsWith('--')) { + throw new Error(`${name} requires a value`); + } + options[key] = key === 'durationSeconds' ? Number(value) : value; + } + return options; +} + +async function validateOptions(options) { + if (!options.file) { + throw new Error('--file is required'); + } + if (!path.isAbsolute(options.file)) { + throw new Error('--file must be an absolute path'); + } + const fileStat = await stat(options.file).catch(() => undefined); + if (!fileStat?.isFile()) { + throw new Error(`--file is not a readable regular file: ${options.file}`); + } + if (path.extname(options.file).toLowerCase() !== '.mcap') { + throw new Error('--file must point to an .mcap file'); + } + let baseUrl; + try { + baseUrl = new URL(options.baseUrl); + } catch { + throw new Error(`--base-url is not a valid URL: ${options.baseUrl}`); + } + if (!['http:', 'https:'].includes(baseUrl.protocol)) { + throw new Error('--base-url must use http or https'); + } + if (!Number.isFinite(options.durationSeconds) || options.durationSeconds < 2) { + throw new Error('--duration must be a number of at least 2 seconds'); + } + if (options.output) { + options.output = path.resolve(options.output); + } + return { ...options, baseUrl: baseUrl.toString(), fileSize: fileStat.size }; +} + +function parseRange(rangeHeader, size) { + if (!rangeHeader) { + return undefined; + } + const match = /^bytes=(\d*)-(\d*)$/.exec(rangeHeader); + if (!match || (!match[1] && !match[2])) { + return null; + } + let start; + let end; + if (!match[1]) { + const suffixLength = Number(match[2]); + if (!Number.isInteger(suffixLength) || suffixLength <= 0) { + return null; + } + start = Math.max(0, size - suffixLength); + end = size - 1; + } else { + start = Number(match[1]); + end = match[2] ? Number(match[2]) : size - 1; + } + if ( + !Number.isSafeInteger(start) || + !Number.isSafeInteger(end) || + start < 0 || + start >= size || + end < start + ) { + return null; + } + return { start, end: Math.min(end, size - 1) }; +} + +async function startFileServer(file, size) { + const server = http.createServer((request, response) => { + response.setHeader('Access-Control-Allow-Origin', '*'); + response.setHeader('Access-Control-Allow-Methods', 'GET, HEAD, OPTIONS'); + response.setHeader('Access-Control-Allow-Headers', 'Range'); + response.setHeader('Access-Control-Expose-Headers', 'Accept-Ranges, Content-Length, Content-Range'); + response.setHeader('Accept-Ranges', 'bytes'); + response.setHeader('Cache-Control', 'no-store'); + + if (request.method === 'OPTIONS') { + response.writeHead(204); + response.end(); + return; + } + if (request.url !== '/recording.mcap' || !['GET', 'HEAD'].includes(request.method ?? '')) { + response.writeHead(404); + response.end(); + return; + } + + const range = parseRange(request.headers.range, size); + if (range === null) { + response.setHeader('Content-Range', `bytes */${size}`); + response.writeHead(416); + response.end(); + return; + } + const start = range?.start ?? 0; + const end = range?.end ?? size - 1; + const contentLength = end - start + 1; + response.setHeader('Content-Type', 'application/octet-stream'); + response.setHeader('Content-Length', contentLength); + if (range) { + response.setHeader('Content-Range', `bytes ${start}-${end}/${size}`); + response.writeHead(206); + } else { + response.writeHead(200); + } + if (request.method === 'HEAD') { + response.end(); + return; + } + const stream = createReadStream(file, { start, end }); + stream.on('error', (error) => response.destroy(error)); + stream.pipe(response); + }); + + await new Promise((resolve, reject) => { + server.once('error', reject); + server.listen(0, '127.0.0.1', resolve); + }); + const address = server.address(); + if (!address || typeof address === 'string') { + throw new Error('Could not determine benchmark file server address'); + } + return { + server, + url: `http://127.0.0.1:${address.port}/recording.mcap`, + }; +} + +async function closeServer(server) { + if (!server) { + return; + } + await new Promise((resolve) => server.close(resolve)); +} + +async function readHeap(page) { + return page.evaluate(() => { + const memory = performance.memory; + return memory && Number.isFinite(memory.usedJSHeapSize) ? memory.usedJSHeapSize : null; + }); +} + +async function readProgress(fill) { + return fill.evaluate((element) => { + const value = Number.parseFloat(element.style.width); + return Number.isFinite(value) ? value : null; + }); +} + +async function readImagePanel(page) { + const panel = page.getByTestId('image-panel').first(); + if (!(await panel.isVisible().catch(() => false))) { + return null; + } + return panel.evaluate((element) => { + const numberAttribute = (name) => { + const value = element.getAttribute(name); + if (value === null) return null; + const number = Number(value); + return Number.isFinite(number) ? number : null; + }; + return { + pressure: element.getAttribute('data-h264-pressure'), + queueFrames: numberAttribute('data-h264-queue-frames'), + droppedFrames: numberAttribute('data-h264-dropped-frames'), + }; + }); +} + +function summarizeProgress(samples) { + const widths = samples.map(({ width }) => width).filter((value) => value !== null); + let updateCount = 0; + let forwardUpdateCount = 0; + let longestStallSamples = 0; + let currentStallSamples = 0; + for (let index = 1; index < widths.length; index += 1) { + const delta = widths[index] - widths[index - 1]; + if (Math.abs(delta) >= 0.05) { + updateCount += 1; + currentStallSamples = 0; + } else { + currentStallSamples += 1; + longestStallSamples = Math.max(longestStallSamples, currentStallSamples); + } + if (delta >= 0.05) { + forwardUpdateCount += 1; + } + } + return { + sampleCount: widths.length, + updateCount, + forwardUpdateCount, + updateRatio: widths.length > 1 ? updateCount / (widths.length - 1) : 0, + minPercent: widths.length > 0 ? Math.min(...widths) : null, + maxPercent: widths.length > 0 ? Math.max(...widths) : null, + longestStallMs: longestStallSamples * SAMPLE_INTERVAL_MS, + }; +} + +async function runBenchmark(options, fixtureUrl) { + const browser = await chromium.launch(); + try { + const page = await browser.newPage(); + const pageErrors = []; + const consoleErrors = []; + page.on('pageerror', (error) => pageErrors.push(error.message)); + page.on('console', (message) => { + if (message.type() === 'error') consoleErrors.push(message.text()); + }); + + const benchmarkUrl = new URL(options.baseUrl); + benchmarkUrl.searchParams.set('url', fixtureUrl); + const startedAt = Date.now(); + await page.goto(benchmarkUrl.toString(), { waitUntil: 'domcontentloaded', timeout: 60_000 }); + await page + .locator('#rosview-root[data-player-presence="ready"]') + .waitFor({ state: 'visible', timeout: 90_000 }); + const play = page.getByRole('button', { name: 'Play playback' }); + await play.waitFor({ state: 'visible', timeout: 30_000 }); + await play.click(); + + const fill = page.getByTestId('playback-progress-fill'); + const track = page.getByTestId('playback-track'); + await fill.waitFor({ state: 'visible' }); + const samples = []; + const imageSamples = []; + const heapSamples = []; + const seeks = []; + const durationMs = options.durationSeconds * 1_000; + const seekTimes = [0.35, 0.6, 0.82].map((ratio) => durationMs * ratio); + let nextSeek = 0; + const samplingStartedAt = Date.now(); + + while (Date.now() - samplingStartedAt < durationMs) { + const elapsedMs = Date.now() - samplingStartedAt; + if (nextSeek < seekTimes.length && elapsedMs >= seekTimes[nextSeek]) { + const targetRatio = 0.1 + Math.random() * 0.8; + const box = await track.boundingBox(); + if (box && box.width > 10 && box.height > 0) { + await track.click({ + position: { + x: Math.max(1, Math.min(box.width - 1, box.width * targetRatio)), + y: box.height / 2, + }, + }); + seeks.push({ elapsedMs, targetRatio }); + } + nextSeek += 1; + } + + const resume = page.getByRole('button', { name: 'Play playback' }); + if (await resume.isVisible().catch(() => false)) { + await resume.click(); + } + samples.push({ elapsedMs, width: await readProgress(fill) }); + const image = await readImagePanel(page); + if (image) imageSamples.push({ elapsedMs, ...image }); + const heapBytes = await readHeap(page); + if (heapBytes !== null) heapSamples.push({ elapsedMs, bytes: heapBytes }); + await page.waitForTimeout(SAMPLE_INTERVAL_MS); + } + + const decodeErrorTexts = await page + .getByText(/decode failed|could not be decoded/i) + .allTextContents() + .catch(() => []); + const statusTexts = await page.getByTestId('image-panel-status').allTextContents().catch(() => []); + const heapValues = heapSamples.map(({ bytes }) => bytes); + const latestImage = imageSamples.at(-1) ?? null; + const metricsReasonable = + latestImage === null || + (['normal', 'degraded', 'recovery'].includes(latestImage.pressure) && + Number.isInteger(latestImage.queueFrames) && + latestImage.queueFrames >= 0 && + Number.isInteger(latestImage.droppedFrames) && + latestImage.droppedFrames >= 0); + + return { + schemaVersion: 1, + generatedAt: new Date().toISOString(), + input: { + file: options.file, + fileBytes: options.fileSize, + baseUrl: options.baseUrl, + durationSeconds: options.durationSeconds, + }, + environment: { + platform: process.platform, + arch: process.arch, + node: process.version, + browserVersion: browser.version(), + }, + timing: { + readyAndPlayMs: samplingStartedAt - startedAt, + sampledMs: Date.now() - samplingStartedAt, + }, + progress: { + ...summarizeProgress(samples), + samples, + }, + seeks, + image: { + appeared: imageSamples.length > 0, + statusTexts, + metricsReasonable, + latestMetrics: latestImage, + samples: imageSamples, + decodeErrors: decodeErrorTexts, + }, + heap: + heapValues.length > 0 + ? { + available: true, + initialBytes: heapValues[0], + finalBytes: heapValues.at(-1), + maxBytes: Math.max(...heapValues), + deltaBytes: heapValues.at(-1) - heapValues[0], + samples: heapSamples, + } + : { available: false }, + errors: { + page: pageErrors, + console: consoleErrors, + }, + }; + } finally { + await browser.close(); + } +} + +async function main() { + let fileServer; + try { + const parsed = parseArgs(process.argv); + if (parsed.help) { + process.stdout.write(usage()); + return; + } + const options = await validateOptions(parsed); + const startedServer = await startFileServer(options.file, options.fileSize); + fileServer = startedServer.server; + const result = await runBenchmark(options, startedServer.url); + const json = `${JSON.stringify(result, null, 2)}\n`; + if (options.output) { + await writeFile(options.output, json, 'utf8'); + } + process.stdout.write(json); + } finally { + await closeServer(fileServer); + } +} + +main().catch((error) => { + process.stderr.write(`benchmark:h264: ${error instanceof Error ? error.message : String(error)}\n`); + process.exitCode = 1; +}); diff --git a/src/core/players/IterablePlayer.test.ts b/src/core/players/IterablePlayer.test.ts index 4ae2bbb..2490e6e 100644 --- a/src/core/players/IterablePlayer.test.ts +++ b/src/core/players/IterablePlayer.test.ts @@ -59,6 +59,16 @@ function makeImageMessageAt(sec: number): MessageEvent { }; } +function makeImageMessageAtMs(ms: number): MessageEvent { + const sec = Math.floor(ms / 1000); + const nsec = (ms % 1000) * 1_000_000; + return { + ...makeImageMessage(), + receiveTime: { sec, nsec }, + publishTime: { sec, nsec }, + }; +} + function makeSource(messages: MessageEvent[]): WorkerSerializedSource { return { initialize: vi.fn(async () => makeInitialization()), @@ -201,11 +211,8 @@ describe('IterablePlayer high-frequency lane', () => { return 1; }; globalThis.cancelAnimationFrame = vi.fn() as typeof cancelAnimationFrame; - const first = makeImageMessage(); - const second = { - ...makeImageMessage(), - receiveTime: { sec: 2, nsec: 0 }, - }; + const first = makeImageMessageAtMs(10); + const second = makeImageMessageAtMs(20); const source = makeSource([first]); const cursor = { nextBatch: vi.fn(async () => [first, second]), @@ -236,7 +243,7 @@ describe('IterablePlayer high-frequency lane', () => { ); expect(onMessageBatch).toHaveBeenCalledWith([ expect.objectContaining({ topic: TOPIC }), - expect.objectContaining({ receiveTime: { sec: 2, nsec: 0 } }), + expect.objectContaining({ receiveTime: { sec: 0, nsec: 20_000_000 } }), ]); } finally { player.close(); @@ -437,6 +444,498 @@ describe('IterablePlayer playback clock', () => { } }); + it('keeps time moving during a short cursor stall, then buffers without accumulating stale video', async () => { + let now = 0; + let nextRafId = 1; + const oldPerformanceNow = performance.now; + const oldRequestAnimationFrame = globalThis.requestAnimationFrame; + const oldCancelAnimationFrame = globalThis.cancelAnimationFrame; + const rafCallbacks = new Map(); + Object.defineProperty(performance, 'now', { + configurable: true, + value: () => now, + }); + globalThis.requestAnimationFrame = vi.fn((cb: FrameRequestCallback) => { + const id = nextRafId++; + rafCallbacks.set(id, cb); + return id; + }); + globalThis.cancelAnimationFrame = vi.fn((id: number) => { + rafCallbacks.delete(id); + }); + + const delayedBatch = deferred(); + const source = makeSource([]); + const cursor = { + nextBatch: vi.fn(() => delayedBatch.promise), + end: vi.fn(async () => undefined), + }; + vi.mocked(source.getMessageCursor).mockResolvedValue(cursor as never); + const player = new IterablePlayer(source); + const seenTimes: number[] = []; + let latestState: PlayerState | undefined; + + const runNextRaf = () => { + const id = Math.min(...rafCallbacks.keys()); + const callback = rafCallbacks.get(id); + rafCallbacks.delete(id); + callback?.(now); + }; + + try { + player.setListener((state) => { + latestState = state; + }); + await player.initialize({}); + player.registerSubscriptions('panel', [{ topic: TOPIC, subscriberId: 'panel' }]); + await flushAsyncWork(); + player.subscribeCurrentTime((time) => { + seenTimes.push(time.sec + time.nsec / 1e9); + }); + player.play(); + + now = 100; + runNextRaf(); + await flushAsyncWork(); + expect(cursor.nextBatch).toHaveBeenCalledTimes(1); + expect(seenTimes.at(-1)).toBeCloseTo(0.1, 3); + + now = 200; + runNextRaf(); + await Promise.resolve(); + expect(seenTimes.at(-1)).toBeCloseTo(0.2, 3); + expect(latestState?.progress.buffering).toBe(false); + + now = 600; + runNextRaf(); + await Promise.resolve(); + expect(seenTimes.at(-1)).toBeCloseTo(0.2, 3); + expect(latestState?.progress.buffering).toBe(true); + + delayedBatch.resolve([makeImageMessageAtMs(150), makeImageMessageAtMs(50)]); + await flushAsyncWork(); + + expect(messageBus.getSubscriberMessages('panel').map((message) => message.receiveTime)).toEqual([ + { sec: 0, nsec: 50_000_000 }, + { sec: 0, nsec: 150_000_000 }, + ]); + expect(latestState?.progress.buffering).toBe(false); + } finally { + player.close(); + Object.defineProperty(performance, 'now', { + configurable: true, + value: oldPerformanceNow, + }); + globalThis.requestAnimationFrame = oldRequestAnimationFrame; + globalThis.cancelAnimationFrame = oldCancelAnimationFrame; + } + }); + + it('keeps a slow-starting cursor after an empty batch and delivers the next message', async () => { + let now = 0; + let nextRafId = 1; + const oldPerformanceNow = performance.now; + const oldRequestAnimationFrame = globalThis.requestAnimationFrame; + const oldCancelAnimationFrame = globalThis.cancelAnimationFrame; + const rafCallbacks = new Map(); + Object.defineProperty(performance, 'now', { + configurable: true, + value: () => now, + }); + globalThis.requestAnimationFrame = vi.fn((cb: FrameRequestCallback) => { + const id = nextRafId++; + rafCallbacks.set(id, cb); + return id; + }); + globalThis.cancelAnimationFrame = vi.fn((id: number) => { + rafCallbacks.delete(id); + }); + + const source = makeSource([]); + const message = makeImageMessageAtMs(150); + const cursor = { + nextBatch: vi.fn().mockResolvedValueOnce([]).mockResolvedValueOnce([message]), + end: vi.fn(async () => undefined), + }; + vi.mocked(source.getMessageCursor).mockResolvedValue(cursor as never); + const player = new IterablePlayer(source); + const seenTimes: number[] = []; + + const runNextRaf = () => { + const id = Math.min(...rafCallbacks.keys()); + const callback = rafCallbacks.get(id); + rafCallbacks.delete(id); + callback?.(now); + }; + + try { + await player.initialize({}); + player.registerSubscriptions('panel', [{ topic: TOPIC, subscriberId: 'panel' }]); + await flushAsyncWork(); + player.subscribeCurrentTime((time) => { + seenTimes.push(time.sec + time.nsec / 1e9); + }); + player.play(); + + now = 100; + runNextRaf(); + await flushAsyncWork(); + now = 200; + runNextRaf(); + await flushAsyncWork(); + + expect(seenTimes.at(-1)).toBeCloseTo(0.2, 3); + expect(cursor.nextBatch).toHaveBeenCalledTimes(2); + expect(cursor.end).not.toHaveBeenCalled(); + expect(source.getMessageCursor).toHaveBeenCalledTimes(1); + expect(messageBus.getSubscriberMessages('panel')).toEqual([message]); + } finally { + player.close(); + Object.defineProperty(performance, 'now', { + configurable: true, + value: oldPerformanceNow, + }); + globalThis.requestAnimationFrame = oldRequestAnimationFrame; + globalThis.cancelAnimationFrame = oldCancelAnimationFrame; + } + }); + + it('keeps one cursor across sparse topic windows and recovers from buffering', async () => { + let now = 0; + let nextRafId = 1; + const oldPerformanceNow = performance.now; + const oldRequestAnimationFrame = globalThis.requestAnimationFrame; + const oldCancelAnimationFrame = globalThis.cancelAnimationFrame; + const rafCallbacks = new Map(); + Object.defineProperty(performance, 'now', { + configurable: true, + value: () => now, + }); + globalThis.requestAnimationFrame = vi.fn((cb: FrameRequestCallback) => { + const id = nextRafId++; + rafCallbacks.set(id, cb); + return id; + }); + globalThis.cancelAnimationFrame = vi.fn((id: number) => { + rafCallbacks.delete(id); + }); + + const sparseMessage = makeImageMessageAtMs(900); + const sparseBatch = deferred(); + const source = makeSource([]); + const cursor = { + nextBatch: vi.fn().mockResolvedValueOnce([]).mockResolvedValueOnce([]).mockReturnValueOnce(sparseBatch.promise), + end: vi.fn(async () => undefined), + }; + vi.mocked(source.getMessageCursor).mockResolvedValue(cursor as never); + const player = new IterablePlayer(source); + let latestState: PlayerState | undefined; + + const runNextRaf = () => { + const id = Math.min(...rafCallbacks.keys()); + const callback = rafCallbacks.get(id); + rafCallbacks.delete(id); + callback?.(now); + }; + + try { + player.setListener((state) => { + latestState = state; + }); + await player.initialize({}); + player.registerSubscriptions('panel', [{ topic: TOPIC, subscriberId: 'panel' }]); + await flushAsyncWork(); + player.play(); + + now = 100; + runNextRaf(); + await flushAsyncWork(); + now = 300; + runNextRaf(); + await flushAsyncWork(); + now = 600; + runNextRaf(); + await Promise.resolve(); + + expect(cursor.nextBatch).toHaveBeenCalledTimes(3); + expect(cursor.end).not.toHaveBeenCalled(); + expect(source.getMessageCursor).toHaveBeenCalledTimes(1); + expect(latestState?.progress.buffering).toBe(true); + + sparseBatch.resolve([sparseMessage]); + await flushAsyncWork(); + expect(latestState?.progress.buffering).toBe(false); + + now = 1200; + runNextRaf(); + await flushAsyncWork(); + + expect(messageBus.getSubscriberMessages('panel')).toEqual([sparseMessage]); + } finally { + player.close(); + Object.defineProperty(performance, 'now', { + configurable: true, + value: oldPerformanceNow, + }); + globalThis.requestAnimationFrame = oldRequestAnimationFrame; + globalThis.cancelAnimationFrame = oldCancelAnimationFrame; + } + }); + + it('invalidates buffered messages when subscriptions change on the same topic', async () => { + let now = 0; + let nextRafId = 1; + const oldPerformanceNow = performance.now; + const oldRequestAnimationFrame = globalThis.requestAnimationFrame; + const oldCancelAnimationFrame = globalThis.cancelAnimationFrame; + const rafCallbacks = new Map(); + Object.defineProperty(performance, 'now', { + configurable: true, + value: () => now, + }); + globalThis.requestAnimationFrame = vi.fn((cb: FrameRequestCallback) => { + const id = nextRafId++; + rafCallbacks.set(id, cb); + return id; + }); + globalThis.cancelAnimationFrame = vi.fn((id: number) => { + rafCallbacks.delete(id); + }); + + const message = makeImageMessageAtMs(300); + const source = makeSource([]); + const firstCursor = { + nextBatch: vi.fn(async () => [message]), + end: vi.fn(async () => undefined), + }; + const secondCursor = { + nextBatch: vi.fn(async () => [message]), + end: vi.fn(async () => undefined), + }; + vi.mocked(source.getMessageCursor) + .mockResolvedValueOnce(firstCursor as never) + .mockResolvedValueOnce(secondCursor as never); + const player = new IterablePlayer(source); + + const runNextRaf = () => { + const id = Math.min(...rafCallbacks.keys()); + const callback = rafCallbacks.get(id); + rafCallbacks.delete(id); + callback?.(now); + }; + + try { + await player.initialize({}); + player.registerSubscriptions('panel-a', [{ topic: TOPIC, subscriberId: 'panel-a' }]); + await flushAsyncWork(); + player.play(); + + now = 100; + runNextRaf(); + await flushAsyncWork(); + expect(messageBus.getSubscriberMessages('panel-a')).toBeNull(); + + player.registerSubscriptions('panel-b', [{ topic: TOPIC, subscriberId: 'panel-b' }]); + await flushAsyncWork(); + expect(firstCursor.end).toHaveBeenCalledTimes(1); + + now = 300; + runNextRaf(); + await flushAsyncWork(); + + expect(source.getMessageCursor).toHaveBeenCalledTimes(2); + expect(messageBus.getSubscriberMessages('panel-a')).toEqual([message]); + expect(messageBus.getSubscriberMessages('panel-b')).toEqual([message]); + } finally { + player.close(); + Object.defineProperty(performance, 'now', { + configurable: true, + value: oldPerformanceNow, + }); + globalThis.requestAnimationFrame = oldRequestAnimationFrame; + globalThis.cancelAnimationFrame = oldCancelAnimationFrame; + } + }); + + it('reaches loop and once boundaries after a sustained empty tail', async () => { + let now = 0; + let nextRafId = 1; + const oldPerformanceNow = performance.now; + const oldRequestAnimationFrame = globalThis.requestAnimationFrame; + const oldCancelAnimationFrame = globalThis.cancelAnimationFrame; + const rafCallbacks = new Map(); + Object.defineProperty(performance, 'now', { + configurable: true, + value: () => now, + }); + globalThis.requestAnimationFrame = vi.fn((cb: FrameRequestCallback) => { + const id = nextRafId++; + rafCallbacks.set(id, cb); + return id; + }); + globalThis.cancelAnimationFrame = vi.fn((id: number) => { + rafCallbacks.delete(id); + }); + + const source = makeSource([]); + vi.mocked(source.initialize).mockResolvedValueOnce({ + ...makeInitialization(), + end: { sec: 1, nsec: 0 }, + }); + const cursor = { + nextBatch: vi.fn(async () => []), + end: vi.fn(async () => undefined), + }; + vi.mocked(source.getMessageCursor).mockResolvedValue(cursor as never); + const player = new IterablePlayer(source); + let latestState: PlayerState | undefined; + + const runNextRaf = () => { + const id = Math.min(...rafCallbacks.keys()); + const callback = rafCallbacks.get(id); + rafCallbacks.delete(id); + callback?.(now); + }; + + try { + player.setListener((state) => { + latestState = state; + }); + await player.initialize({}); + player.registerSubscriptions('panel', [{ topic: TOPIC, subscriberId: 'panel' }]); + await flushAsyncWork(); + player.play(); + + now = 100; + runNextRaf(); + await flushAsyncWork(); + now = 600; + runNextRaf(); + await flushAsyncWork(); + + expect(latestState?.progress.buffering).toBe(true); + expect(player.getCurrentTime()).toEqual({ sec: 0, nsec: 600_000_000 }); + + now = 1000; + runNextRaf(); + await flushAsyncWork(); + + expect(player.getCurrentTime()).toEqual({ sec: 0, nsec: 0 }); + expect(latestState?.activeData?.isPlaying).toBe(true); + expect(latestState?.progress.buffering).toBe(false); + expect(cursor.end).toHaveBeenCalledTimes(1); + + player.setLooping(false); + now = 2000; + runNextRaf(); + await flushAsyncWork(); + + expect(player.getCurrentTime()).toEqual({ sec: 1, nsec: 0 }); + expect(latestState?.activeData?.isPlaying).toBe(false); + expect(latestState?.progress.buffering).toBe(false); + expect(cursor.end).toHaveBeenCalledTimes(1); + } finally { + player.close(); + Object.defineProperty(performance, 'now', { + configurable: true, + value: oldPerformanceNow, + }); + globalThis.requestAnimationFrame = oldRequestAnimationFrame; + globalThis.cancelAnimationFrame = oldCancelAnimationFrame; + } + }); + + it('waits for the old playback cursor to close before opening one for an H264 consumer', async () => { + let now = 0; + let nextRafId = 1; + const oldPerformanceNow = performance.now; + const oldRequestAnimationFrame = globalThis.requestAnimationFrame; + const oldCancelAnimationFrame = globalThis.cancelAnimationFrame; + const rafCallbacks = new Map(); + Object.defineProperty(performance, 'now', { + configurable: true, + value: () => now, + }); + globalThis.requestAnimationFrame = vi.fn((cb: FrameRequestCallback) => { + const id = nextRafId++; + rafCallbacks.set(id, cb); + return id; + }); + globalThis.cancelAnimationFrame = vi.fn((id: number) => { + rafCallbacks.delete(id); + }); + + const closeDeferred = deferred(); + const events: string[] = []; + const source = makeSource([]); + const firstCursor = { + nextBatch: vi.fn(async () => []), + end: vi.fn(async () => { + events.push('close-start'); + await closeDeferred.promise; + events.push('close-end'); + }), + }; + const secondCursor = { + nextBatch: vi.fn(async () => []), + end: vi.fn(async () => undefined), + }; + vi.mocked(source.getMessageCursor).mockImplementation(async () => { + const call = vi.mocked(source.getMessageCursor).mock.calls.length; + events.push(`open-${call}`); + return (call === 1 ? firstCursor : secondCursor) as never; + }); + const player = new IterablePlayer(source); + + const runNextRaf = () => { + const id = Math.min(...rafCallbacks.keys()); + const callback = rafCallbacks.get(id); + rafCallbacks.delete(id); + callback?.(now); + }; + + try { + await player.initialize({}); + player.registerSubscriptions('panel', [{ topic: TOPIC, subscriberId: 'panel' }]); + await flushAsyncWork(); + player.play(); + + now = 100; + runNextRaf(); + await flushAsyncWork(); + expect(source.getMessageCursor).toHaveBeenCalledTimes(1); + + player.registerHighFrequencyConsumer('h264-panel', { + topic: TOPIC, + lane: 'video', + onLatestMessage: vi.fn(), + }); + await flushAsyncWork(); + expect(firstCursor.end).toHaveBeenCalledTimes(1); + + now = 200; + runNextRaf(); + await flushAsyncWork(); + expect(source.getMessageCursor).toHaveBeenCalledTimes(1); + expect(events).toEqual(['open-1', 'close-start']); + + closeDeferred.resolve(); + await flushAsyncWork(); + + expect(source.getMessageCursor).toHaveBeenCalledTimes(2); + expect(events).toEqual(['open-1', 'close-start', 'close-end', 'open-2']); + } finally { + closeDeferred.resolve(); + player.close(); + Object.defineProperty(performance, 'now', { + configurable: true, + value: oldPerformanceNow, + }); + globalThis.requestAnimationFrame = oldRequestAnimationFrame; + globalThis.cancelAnimationFrame = oldCancelAnimationFrame; + } + }); + it('does not catch up wall time elapsed while the page is hidden', async () => { let now = 0; const oldPerformanceNow = performance.now; @@ -585,7 +1084,7 @@ describe('IterablePlayer playback clock', () => { now = 100; rafCallbacks.get(1)?.(now); - await Promise.resolve(); + await flushAsyncWork(); expect(cursor.nextBatch).toHaveBeenCalled(); player.seek({ sec: 5, nsec: 0 }); @@ -596,7 +1095,7 @@ describe('IterablePlayer playback clock', () => { await flushAsyncWork(); expect(seenTimes.at(-1)).toBe(5); - expect(seenTimes).not.toContain(7.1); + expect(messageBus.getLastMessage(TOPIC)).toBeNull(); } finally { player.close(); Object.defineProperty(performance, 'now', { diff --git a/src/core/players/IterablePlayer.ts b/src/core/players/IterablePlayer.ts index 1703837..7446d2a 100644 --- a/src/core/players/IterablePlayer.ts +++ b/src/core/players/IterablePlayer.ts @@ -21,8 +21,6 @@ const DEFAULT_SAMPLING_FPS = 30; const MAX_SAMPLING_FPS = 45; const LOAD_PROGRESS_POLL_INTERVAL_MS = 1000; const TRANSPORT_DIAGNOSTICS_POLL_INTERVAL_MS = 2000; -const EMPTY_BATCH_BACKFILL_COOLDOWN_MS = 1000; -const EMPTY_BATCH_BACKFILL_TRIGGER = 4; const BACKFILL_STALE_THRESHOLD_NS = 1_000_000_000n; const BACKFILL_STALE_TOPIC_PERIOD_MULTIPLIER = 3; const STALE_TOPIC_REFRESH_COOLDOWN_MS = 500; @@ -30,6 +28,14 @@ const SLOW_DISTRIBUTION_MS = 16; const RANGE_READ_MAX_MESSAGES = 80_000; const RANGE_READ_BATCH_MESSAGES = 2048; const RANGE_READ_BATCH_WALL_MS = 16; +const PLAYBACK_PREFETCH_AHEAD_MS = 250; +const PLAYBACK_PREFETCH_MAX_MESSAGES = 512; +const PLAYBACK_BUFFERING_DELAY_MS = 500; + +function compareMessagesByReceiveTime(a: MessageEvent, b: MessageEvent): number { + const diff = toNano(a.receiveTime) - toNano(b.receiveTime); + return diff < 0n ? -1 : diff > 0n ? 1 : 0; +} function isSameRanges(nextValue?: Range[], prevValue?: Range[]): boolean { if (nextValue === prevValue) { @@ -152,8 +158,13 @@ export class IterablePlayer implements Player { private _currentTime: Time = { sec: 0, nsec: 0 }; private _initialization?: Initialization; private _cursor?: IMessageCursor; + private _playbackCursorCreationPromise?: Promise>; + private _playbackCursorCloseGate: Promise = Promise.resolve(); private _clock = new PlaybackClock(this._currentTime); - private _isFetching: boolean = false; + private _prefetchedMessages: MessageEvent[] = []; + private _prefetchPromise?: Promise; + private _prefetchStartedAtMs: number | undefined; + private _prefetchRequestId = 0; private _timeSubscribers = new Set<(time: Time) => void>(); private _rafId: number | undefined; private _lastPipelineEmitMs = 0; @@ -163,16 +174,15 @@ export class IterablePlayer implements Player { private _lastTickWallMs = 0; private _pageSuspended = false; private _emptyBatchStreak = 0; + private _emptyBatchStartedAtMs: number | undefined; private _cursorRebuildCount = 0; private _fallbackBackfillCount = 0; - private _lastFallbackBackfillMs = 0; private _lastStaleRefreshMs = 0; private _staleRefreshInFlight = false; private _isBuffering = false; private _topicLastMessageNs = new Map(); private _highFrequencyConsumerSignature = ""; private _playbackEpoch = 0; - private _fetchEpoch: number | undefined; private _debugEnabled = typeof window !== "undefined" && new URLSearchParams(window.location.search).get("debugPlayback") === "1"; @@ -247,9 +257,13 @@ export class IterablePlayer implements Player { this._subscriptions = merged; this._subscriberIdsByTopic = this._buildSubscriberIndex(merged); await this._handleTopicSetChange(prevTopics); - if (subscriptionsChanged && this._cursor) { - await this._cursor.end(); - this._cursor = undefined; + if (subscriptionsChanged) { + const epoch = this._advancePlaybackEpoch(); + await this._closePlaybackCursor(); + if (!this._isPlaybackEpochCurrent(epoch)) { + return; + } + this._scheduleNextTick(); } // When subscriptions change while paused and player is ready, backfill the @@ -319,13 +333,9 @@ export class IterablePlayer implements Player { return; } const epoch = this._advancePlaybackEpoch(); - const cursor = this._cursor; - this._cursor = undefined; - if (cursor) { - await cursor.end(); - if (!this._isPlaybackEpochCurrent(epoch)) { - return; - } + await this._closePlaybackCursor(); + if (!this._isPlaybackEpochCurrent(epoch)) { + return; } this._scheduleNextTick(); } @@ -336,11 +346,9 @@ export class IterablePlayer implements Player { ): Promise { const before = new Set(previousTopics); await this._handleTopicSetChange(before); - if (this._cursor && previousSignature !== this._highFrequencyConsumerSignature) { + if (previousSignature !== this._highFrequencyConsumerSignature) { const epoch = this._advancePlaybackEpoch(); - const cursor = this._cursor; - this._cursor = undefined; - await cursor.end(); + await this._closePlaybackCursor(); if (!this._isPlaybackEpochCurrent(epoch)) { return; } @@ -484,6 +492,7 @@ export class IterablePlayer implements Player { this._pageSuspended = false; this._cancelRaf(); this._stopLoadProgressPolling(); + void this._closePlaybackCursor(); this._emitState(); } @@ -516,13 +525,9 @@ export class IterablePlayer implements Player { this._emitState(); try { - const cursor = this._cursor; - this._cursor = undefined; - if (cursor) { - await cursor.end(); - if (!this._isPlaybackEpochCurrent(epoch)) { - return; - } + await this._closePlaybackCursor(); + if (!this._isPlaybackEpochCurrent(epoch)) { + return; } if (!this._isPlaybackEpochCurrent(epoch)) { @@ -575,12 +580,8 @@ export class IterablePlayer implements Player { }); if (!this._isPlaybackEpochCurrent(epoch)) return; if (!msg) return; - const cursor = this._cursor; - this._cursor = undefined; - if (cursor) { - await cursor.end(); - if (!this._isPlaybackEpochCurrent(epoch)) return; - } + await this._closePlaybackCursor(); + if (!this._isPlaybackEpochCurrent(epoch)) return; this._currentTime = this._clampToRange(msg.receiveTime); this._clock.seek(this._currentTime, performance.now()); this._topicLastMessageNs.clear(); @@ -639,14 +640,13 @@ export class IterablePlayer implements Player { this._cancelRaf(); this._detachPageLifecycleListeners(); this._stopLoadProgressPolling(); + void this._closePlaybackCursor(); this._source.terminate(); this._state.presence = "closed"; this._state.progress = {}; this._state.activeData = undefined; this._initialization = undefined; this._clock = new PlaybackClock(); - this._cursor = undefined; - this._fetchEpoch = undefined; this._topicLastMessageNs.clear(); this._highFrequencyConsumersById.clear(); this._highFrequencyConsumersByTopic.clear(); @@ -662,6 +662,13 @@ export class IterablePlayer implements Player { private _advancePlaybackEpoch(): number { this._playbackEpoch += 1; + this._prefetchRequestId += 1; + this._prefetchPromise = undefined; + this._prefetchStartedAtMs = undefined; + this._prefetchedMessages = []; + this._emptyBatchStreak = 0; + this._emptyBatchStartedAtMs = undefined; + this._setBuffering(false); return this._playbackEpoch; } @@ -669,6 +676,37 @@ export class IterablePlayer implements Player { return epoch === this._playbackEpoch && this._state.presence !== "closed"; } + private _enqueuePlaybackCursorClose(cursor: IMessageCursor): Promise { + if (cursor === this._cursor) { + this._cursor = undefined; + } + this._playbackCursorCloseGate = this._playbackCursorCloseGate.then(async () => { + try { + await cursor.end(); + } catch (err) { + console.warn("IterablePlayer: playback cursor close failed", err); + } + }); + return this._playbackCursorCloseGate; + } + + private async _closePlaybackCursor(): Promise { + const pendingCreation = this._playbackCursorCreationPromise; + if (pendingCreation) { + try { + await pendingCreation; + } catch { + // Cursor creation errors are reported by the owning prefetch request. + } + } + const cursor = this._cursor; + if (!cursor) { + await this._playbackCursorCloseGate; + return; + } + await this._enqueuePlaybackCursorClose(cursor); + } + private _scheduleNextTick(): void { this._cancelRaf(); if (this._isPlaying && !this._pageSuspended) { @@ -931,8 +969,19 @@ export class IterablePlayer implements Player { private async _tickAsync(): Promise { if (!this._isPlaying || this._pageSuspended) return; const now = performance.now(); - if (this._isFetching) { + const slowRead = + this._prefetchStartedAtMs != undefined && + now - this._prefetchStartedAtMs >= PLAYBACK_BUFFERING_DELAY_MS; + const sustainedEmpty = + this._emptyBatchStartedAtMs != undefined && + now - this._emptyBatchStartedAtMs >= PLAYBACK_BUFFERING_DELAY_MS; + if (sustainedEmpty && this._prefetchedMessages.length === 0) { + this._setBuffering(true); + } + if (slowRead && this._prefetchedMessages.length === 0) { + this._setBuffering(true); this._clock.seek(this._currentTime, now); + this._ensurePlaybackPrefetch(this._playbackEpoch, this._currentTime, 1); this._scheduleNextTick(); return; } @@ -953,82 +1002,24 @@ export class IterablePlayer implements Player { return; } const batchDurationMs = Math.max(1, Number(nextNs - currentNs) / 1e6); - - if (!this._cursor) { - const topics = this._currentTopics(); - if (topics.length > 0) { - const cursorStartTime = this._currentTime; - const cursor = await this._source.getMessageCursor({ - startTime: cursorStartTime, - topics, - latestOnlyTopics: this._latestOnlyHighFrequencyTopics(), - }); - if (!this._isPlaybackEpochCurrent(epoch) || toNano(this._currentTime) !== toNano(cursorStartTime)) { - await cursor.end(); - return; - } - this._cursor = cursor; - } - } - - if (this._cursor) { - this._isFetching = true; - this._fetchEpoch = epoch; - try { - const cursor = this._cursor; - const messages = await cursor.nextBatch(batchDurationMs, { endTime: nextTime }); - if (!this._isPlaybackEpochCurrent(epoch) || cursor !== this._cursor) { - return; - } - if (messages.length > 0) { - this._emptyBatchStreak = 0; - this._distributeMessages(messages); - if (this._debugEnabled) { - console.debug("[Playback] nextBatch " + JSON.stringify({ - batchDurationMs, - count: messages.length, - currentTime: this._currentTime, - })); - } - } else { - await this._handleEmptyBatch(now, epoch); - if (!this._isPlaybackEpochCurrent(epoch)) { - return; - } - } - } catch (err) { - console.error("Failed to fetch messages", err); - } finally { - if (this._fetchEpoch === epoch) { - this._isFetching = false; - this._fetchEpoch = undefined; - } - } - } - - if (!this._isPlaybackEpochCurrent(epoch)) { - return; - } this._currentTime = nextTime; this._clock.seek(this._currentTime, performance.now()); + this._drainPrefetchedMessages(this._currentTime); this._scheduleStaleTopicsRefresh(now, epoch); if (this._initialization && toNano(this._currentTime) >= toNano(this._initialization.end)) { if (this._isLooping) { + const loopEpoch = this._advancePlaybackEpoch(); this._currentTime = this._initialization.start; this._clock.seek(this._currentTime, performance.now()); - const cursor = this._cursor; - this._cursor = undefined; - if (cursor) { - await cursor.end(); - if (!this._isPlaybackEpochCurrent(epoch)) { - return; - } + await this._closePlaybackCursor(); + if (!this._isPlaybackEpochCurrent(loopEpoch)) { + return; } const topics = this._currentTopics(); if (topics.length > 0) { const messages = await this._source.getBackfillMessages({ time: this._currentTime, topics }); - if (!this._isPlaybackEpochCurrent(epoch)) { + if (!this._isPlaybackEpochCurrent(loopEpoch)) { return; } this._distributeMessages(messages, this._currentTime); @@ -1045,12 +1036,160 @@ export class IterablePlayer implements Player { return; } + this._ensurePlaybackPrefetch(epoch, nextTime, batchDurationMs); this._notifyTimeSubscribers(this._currentTime); this._maybeEmitPipelineState(); this._scheduleNextTick(); } + private _ensurePlaybackPrefetch(epoch: number, nextTime: Time, batchDurationMs: number): void { + if ( + this._prefetchPromise || + this._prefetchedMessages.length >= PLAYBACK_PREFETCH_MAX_MESSAGES || + this._currentTopics().length === 0 + ) { + return; + } + const requestId = ++this._prefetchRequestId; + const lookaheadMs = PLAYBACK_PREFETCH_AHEAD_MS * Math.max(1, this._emptyBatchStreak + 1); + const endTime = this._clampToRange(addMs(nextTime, lookaheadMs)); + const durationMs = batchDurationMs + lookaheadMs; + this._prefetchStartedAtMs = performance.now(); + const promise = this._prefetchPlaybackBatch(epoch, requestId, endTime, durationMs); + this._prefetchPromise = promise; + void promise.finally(() => { + if (this._prefetchRequestId !== requestId) { + return; + } + this._prefetchPromise = undefined; + this._prefetchStartedAtMs = undefined; + this._scheduleNextTick(); + }); + } + + private async _prefetchPlaybackBatch( + epoch: number, + requestId: number, + endTime: Time, + durationMs: number, + ): Promise { + try { + if (!this._cursor) { + const pendingCreation = this._playbackCursorCreationPromise; + if (pendingCreation) { + try { + await pendingCreation; + } catch { + // The request that created it reports the error. + } + } + await this._playbackCursorCloseGate; + if (!this._isPlaybackEpochCurrent(epoch) || this._prefetchRequestId !== requestId) { + return; + } + if (!this._cursor) { + const cursorStartTime = this._currentTime; + const creationPromise = this._source.getMessageCursor({ + startTime: cursorStartTime, + topics: this._currentTopics(), + latestOnlyTopics: this._latestOnlyHighFrequencyTopics(), + }); + this._playbackCursorCreationPromise = creationPromise; + let cursor: IMessageCursor; + try { + cursor = await creationPromise; + } finally { + if (this._playbackCursorCreationPromise === creationPromise) { + this._playbackCursorCreationPromise = undefined; + } + } + if ( + !this._isPlaybackEpochCurrent(epoch) || + this._prefetchRequestId !== requestId || + toNano(this._currentTime) < toNano(cursorStartTime) + ) { + await this._enqueuePlaybackCursorClose(cursor); + return; + } + this._cursor = cursor; + } + } + + const cursor = this._cursor; + const capacity = Math.max(1, PLAYBACK_PREFETCH_MAX_MESSAGES - this._prefetchedMessages.length); + const messages = await cursor.nextBatch(durationMs, { + endTime, + maxMessages: capacity, + }); + if ( + !this._isPlaybackEpochCurrent(epoch) || + this._prefetchRequestId !== requestId || + cursor !== this._cursor + ) { + return; + } + if (messages.length === 0) { + this._recordEmptyBatch(performance.now()); + return; + } + + this._emptyBatchStreak = 0; + this._emptyBatchStartedAtMs = undefined; + this._prefetchedMessages.push(...messages); + this._prefetchedMessages.sort(compareMessagesByReceiveTime); + this._drainPrefetchedMessages(this._currentTime); + if (this._isBuffering) { + this._setBuffering(false); + this._clock.seek(this._currentTime, performance.now()); + } + if (this._debugEnabled) { + console.debug("[Playback] nextBatch " + JSON.stringify({ + durationMs, + count: messages.length, + currentTime: this._currentTime, + bufferedMessages: this._prefetchedMessages.length, + })); + } + } catch (err) { + if (this._isPlaybackEpochCurrent(epoch) && this._prefetchRequestId === requestId) { + console.error("Failed to fetch messages", err); + } + } + } + + private _drainPrefetchedMessages(time: Time): void { + if (this._prefetchedMessages.length === 0) { + return; + } + const timeNs = toNano(time); + let count = 0; + while ( + count < this._prefetchedMessages.length && + toNano(this._prefetchedMessages[count].receiveTime) <= timeNs + ) { + count += 1; + } + if (count === 0) { + return; + } + this._distributeMessages(this._prefetchedMessages.splice(0, count)); + } + + private _setBuffering(buffering: boolean): void { + if (this._isBuffering === buffering) { + return; + } + this._isBuffering = buffering; + this._state.progress = { + ...this._state.progress, + buffering, + }; + if (this._state.presence === "ready") { + this._emitState(); + } + } + private _clampToRange(time: Time): Time { if (!this._initialization) return time; const t = toNano(time); @@ -1152,94 +1291,15 @@ export class IterablePlayer implements Player { } } - private async _handleEmptyBatch(nowMs: number, epoch: number): Promise { + private _recordEmptyBatch(nowMs: number): void { this._emptyBatchStreak += 1; + this._emptyBatchStartedAtMs ??= nowMs; this._state.progress = { ...this._state.progress, emptyBatchStreak: this._emptyBatchStreak, cursorRebuildCount: this._cursorRebuildCount, fallbackBackfillCount: this._fallbackBackfillCount, }; - - if (this._emptyBatchStreak === 1) { - await this._rebuildCursorFromCurrentTime(epoch); - if (!this._isPlaybackEpochCurrent(epoch)) { - return; - } - if (this._debugEnabled) { - console.debug("[Playback] empty batch -> rebuild cursor " + JSON.stringify({ - streak: this._emptyBatchStreak, - currentTime: this._currentTime, - })); - } - return; - } - if (this._emptyBatchStreak < EMPTY_BATCH_BACKFILL_TRIGGER) { - return; - } - if (nowMs - this._lastFallbackBackfillMs < EMPTY_BATCH_BACKFILL_COOLDOWN_MS) { - return; - } - this._lastFallbackBackfillMs = nowMs; - this._fallbackBackfillCount += 1; - const referenceTime = this._currentTime; - const topics = this._currentTopics(); - if (topics.length === 0) { - return; - } - try { - const messages = await this._source.getBackfillMessages({ - time: referenceTime, - topics, - }); - if (!this._isPlaybackEpochCurrent(epoch)) { - return; - } - // For latched topics (URDF, static TF), the backfill returns the same - // message we already distributed – skip those so downstream panels don't - // needlessly rebuild (URDF rebuild fetches + parses meshes, which was - // the main cause of unbounded heap/CPU growth during long playback). - const freshMessages = this._filterAlreadyDeliveredMessages(messages); - if (freshMessages.length === 0) { - return; - } - this._distributeMessages(freshMessages, referenceTime); - if (this._debugEnabled) { - console.debug("[Playback] empty batch -> fallback backfill " + JSON.stringify({ - streak: this._emptyBatchStreak, - topics: topics.length, - count: messages.length, - currentTime: this._currentTime, - })); - } - } catch (err) { - console.warn("IterablePlayer: fallback backfill failed", err); - } - } - - private async _rebuildCursorFromCurrentTime(epoch: number): Promise { - const previousCursor = this._cursor; - this._cursor = undefined; - if (previousCursor) { - await previousCursor.end(); - if (!this._isPlaybackEpochCurrent(epoch)) { - return; - } - } - const topics = this._currentTopics(); - if (topics.length === 0) return; - this._cursorRebuildCount += 1; - const cursorStartTime = this._currentTime; - const cursor = await this._source.getMessageCursor({ - startTime: cursorStartTime, - topics, - latestOnlyTopics: this._latestOnlyHighFrequencyTopics(), - }); - if (!this._isPlaybackEpochCurrent(epoch) || toNano(this._currentTime) !== toNano(cursorStartTime)) { - await cursor.end(); - return; - } - this._cursor = cursor; } private _scheduleStaleTopicsRefresh(nowMs: number, epoch: number): void { @@ -1274,7 +1334,7 @@ export class IterablePlayer implements Player { if (!this._isPlaybackEpochCurrent(epoch)) { return; } - // Same reasoning as in _handleEmptyBatch: latched topics would otherwise + // Latched topics would otherwise // get re-delivered on every refresh tick (~5 Hz), causing panels like // the 3D/URDF renderer to rebuild from scratch and leak GPU buffers. const freshMessages = this._filterAlreadyDeliveredMessages(messages); diff --git a/src/features/panels/Image/ImagePanel.tsx b/src/features/panels/Image/ImagePanel.tsx index bd3811f..1dd1e61 100644 --- a/src/features/panels/Image/ImagePanel.tsx +++ b/src/features/panels/Image/ImagePanel.tsx @@ -6,6 +6,7 @@ import { scheduleFrame } from '@/shared/utils/rafScheduler'; import { toNano } from '@/shared/utils/time'; import type { RawImageDecodeOptions } from './core/imageColorMode'; import type { + ImageRenderMetrics, ImageRenderOptions, ImageRenderWorkerEvent, ImageRenderWorkerRequest, @@ -74,6 +75,7 @@ export const ImagePanel: React.FC = (props) => { const lastUiStatusRef = useRef({ phase: 'idle' }); const h264ModeRef = useRef(false); const [status, setStatus] = useState({ phase: 'idle' }); + const [metrics, setMetrics] = useState(null); const mainConsumerId = `${panelId}:image-main`; const h264ConsumerId = `${panelId}:image-main-h264`; @@ -124,6 +126,10 @@ export const ImagePanel: React.FC = (props) => { worker.onmessage = (event) => { const data = event.data as ImageRenderWorkerEvent; + if (data.type === 'metrics') { + setMetrics(data.metrics); + return; + } if (data.type !== 'status') { return; } @@ -181,6 +187,7 @@ export const ImagePanel: React.FC = (props) => { transferredCanvasRef.current = null; lastUiStatusRef.current = { phase: 'idle' }; setStatus({ phase: 'idle' }); + setMetrics(null); workerDisposeTimerRef.current = null; }, 0); }; @@ -198,6 +205,7 @@ export const ImagePanel: React.FC = (props) => { return; } h264ModeRef.current = false; + setMetrics(null); worker.postMessage({ type: 'reset' } satisfies ImageRenderWorkerRequest); player.registerHighFrequencyConsumer(mainConsumerId, { topic, @@ -304,6 +312,9 @@ export const ImagePanel: React.FC = (props) => { className="flex flex-col h-full overflow-hidden relative" style={{ background: backgroundColor }} data-testid="image-panel" + data-h264-pressure={metrics?.pressureMode} + data-h264-queue-frames={metrics?.queueFrames} + data-h264-dropped-frames={metrics?.droppedFrames} > }[] = []; #lastDeltaTimeKey = 0n; + #lastTimestampUs = -1; + #configuredCodec: string | null = null; + #streamCodec: string | null = null; #mutex = Promise.resolve(undefined); #resolveFrame: ((frame: VideoFrame) => void) | null = null; #rejectFrame: ((err: Error) => void) | null = null; #outputTimeout: ReturnType | null = null; public dispose(): void { - this.#clearOutputWait(); + this.reset(); + this.#lastKeyChunks = []; + this.#lastDeltaTimeKey = 0n; + this.#lastTimestampUs = -1; + this.#mutex = Promise.resolve(); + } + + public reset(): void { + this.#cancelOutputWait(new Error('H.264 decode was reset')); if (this.#decoder && this.#decoder.state !== 'closed') { this.#decoder.close(); } this.#decoder = null; + this.#configuredCodec = null; + this.#streamCodec = null; this.#lastKeyChunks = []; this.#lastDeltaTimeKey = 0n; - this.#mutex = Promise.resolve(); } - public decodeFrame(data: Uint8Array, sortTimeKey: bigint): Promise { + public get codec(): string | undefined { + return this.#configuredCodec ?? undefined; + } + + public decodeFrame(data: Uint8Array, sortTimeKey: bigint): Promise { if (typeof VideoDecoder === 'undefined') { return Promise.reject(new Error('WebCodecs VideoDecoder is not supported')); } - const run = async (): Promise => { - await this.#ensureDecoder(); + const run = async (): Promise => { + await this.#ensureDecoder(data); const decoder = this.#decoder!; const type = getH264ChunkType(data); + const nalTypes = scanH264NalTypes(data); - if (type === 'key') { - this.#lastKeyChunks.push({ timeKey: sortTimeKey, data: cloneBytes(data) }); + if (containsH264IdrNal(data)) { + // Worker messages already own their ArrayBuffer; retaining the view is safe. + this.#lastKeyChunks.push({ timeKey: sortTimeKey, data }); if (this.#lastKeyChunks.length > MAX_KEY_CHUNKS) { this.#lastKeyChunks.shift(); } } + if (!nalTypes.some((nalType) => nalType === 1 || nalType === 5)) { + decoder.decode( + new EncodedVideoChunk({ + type, + timestamp: this.#monotonicTimestampUs(sortTimeKey), + data, + }), + ); + return null; + } + let lastFrame: VideoFrame | null = null; if (type === 'delta' && sortTimeKey < this.#lastDeltaTimeKey) { for (const { timeKey, data: keyData } of this.#lastKeyChunks) { if (timeKey < sortTimeKey) { try { - const recovered = await this.#decodeOneChunk(decoder, keyData, 'key'); + const recovered = await this.#decodeOneChunk(decoder, keyData, 'key', timeKey); lastFrame?.close(); lastFrame = recovered; } catch { @@ -110,7 +160,7 @@ class WorkerH264Decoder { } } this.#lastDeltaTimeKey = sortTimeKey; - const frame = await this.#decodeOneChunk(decoder, data, type); + const frame = await this.#decodeOneChunk(decoder, data, type, sortTimeKey); lastFrame?.close(); return frame; }; @@ -123,14 +173,46 @@ class WorkerH264Decoder { return promise; } - async #ensureDecoder(): Promise { + async #ensureDecoder(data: Uint8Array): Promise { + const parsedCodec = parseH264SpsCodec(data); + if ( + parsedCodec && + parsedCodec !== this.#streamCodec && + this.#decoder && + this.#decoder.state !== 'closed' + ) { + this.reset(); + } if (this.#decoder && this.#decoder.state !== 'closed') { return; } - const support = await VideoDecoder.isConfigSupported({ codec: H264_CODEC }); - if (!support.supported) { - throw new Error(`H.264 codec ${H264_CODEC} is not supported`); + + let supportedConfig: VideoDecoderConfig | null = null; + for (const codec of getH264CodecCandidates(data)) { + const candidates: VideoDecoderConfig[] = [ + { codec, hardwareAcceleration: 'prefer-hardware', optimizeForLatency: true }, + { codec, hardwareAcceleration: 'no-preference', optimizeForLatency: true }, + ]; + for (const candidate of candidates) { + try { + const support = await VideoDecoder.isConfigSupported(candidate); + if (support.supported) { + supportedConfig = support.config ?? candidate; + break; + } + } catch { + // Some implementations throw for an unsupported acceleration mode. + } + } + if (supportedConfig) { + break; + } + } + if (!supportedConfig) { + const parsed = parseH264SpsCodec(data); + throw new Error(`H.264 codec ${parsed ?? 'fallback candidates'} is not supported`); } + this.#decoder = new VideoDecoder({ output: (frame) => { this.#finishOk(frame); @@ -139,20 +221,30 @@ class WorkerH264Decoder { this.#finishErr(new Error(String(error))); }, }); - this.#decoder.configure({ codec: H264_CODEC }); + try { + this.#decoder.configure(supportedConfig); + this.#configuredCodec = supportedConfig.codec; + this.#streamCodec = parsedCodec; + } catch (error) { + this.#decoder.close(); + this.#decoder = null; + this.#configuredCodec = null; + throw error; + } } async #decodeOneChunk( decoder: VideoDecoder, data: Uint8Array, type: 'key' | 'delta', + timeKey: bigint, ): Promise { const wait = this.#beginWait(); try { decoder.decode( new EncodedVideoChunk({ type, - timestamp: 0, + timestamp: this.#monotonicTimestampUs(timeKey), data, }), ); @@ -186,6 +278,18 @@ class WorkerH264Decoder { this.#rejectFrame = null; } + #cancelOutputWait(error: Error): void { + const rejectFrame = this.#rejectFrame; + this.#clearOutputWait(); + rejectFrame?.(error); + } + + #monotonicTimestampUs(timeKey: bigint): number { + const timestamp = monotonicH264TimestampUs(timeKey, this.#lastTimestampUs); + this.#lastTimestampUs = timestamp; + return timestamp; + } + #finishOk(frame: VideoFrame): void { if (this.#outputTimeout != null) { clearTimeout(this.#outputTimeout); @@ -268,6 +372,18 @@ class ImageRenderWorkerRuntime { #imageDecoderMimeSupported = new Map(); /** The last decoded frame retained for instant-redraw on option changes. */ #cachedFrame: CachedFrame | null = null; + #h264Pressure: H264PressureState = initialH264PressureState(); + #h264DecodeMs = 0; + #h264WaitingForIdr = false; + #h264ConfigBeforeIdr: ImageWorkerFrameEnvelope[] = []; + #h264RecentConfig: ImageWorkerFrameEnvelope[] = []; + #h264NeedsResync = false; + #lastH264RenderAt = -Infinity; + #lastH264BitmapAt = -Infinity; + #droppedH264Frames = 0; + #renderedH264Frames = 0; + #lastMetricsAt = -Infinity; + #epoch = 0; public constructor() { if (!this.#bufferCtx) { @@ -319,9 +435,12 @@ class ImageRenderWorkerRuntime { return; case 'reset': + this.#epoch += 1; this.#pendingFrame = null; this.#pendingH264Frames = []; this.#haltUntilReset = false; + this.#resetH264RuntimeState(); + this.#decoder.reset(); this.#disposeAuxiliaryDecodeState(); this.#disposeCachedBitmap(); this.#cachedFrame = null; @@ -330,6 +449,7 @@ class ImageRenderWorkerRuntime { return; case 'dispose': + this.#epoch += 1; this.#pendingFrame = null; this.#pendingH264Frames = []; this.#haltUntilReset = false; @@ -347,23 +467,84 @@ class ImageRenderWorkerRuntime { this.#pendingFrame = frame; return; } + this.#h264RecentConfig = updateH264ConfigPackets(this.#h264RecentConfig, frame); + if (this.#h264WaitingForIdr && !containsH264IdrNal(frame.data)) { + if (isH264ConfigOnly(frame.data)) { + this.#h264ConfigBeforeIdr = updateH264ConfigPackets( + this.#h264ConfigBeforeIdr, + frame, + ); + return; + } + this.#droppedH264Frames += 1; + this.#emitMetricsIfDue(); + return; + } + if (containsH264IdrNal(frame.data)) { + this.#h264WaitingForIdr = false; + if (this.#h264ConfigBeforeIdr.length > 0) { + this.#pendingH264Frames.push(...this.#h264ConfigBeforeIdr); + this.#h264ConfigBeforeIdr = []; + } + } this.#pendingH264Frames.push(frame); - this.#trimPendingH264Frames(); + this.#updateH264Pressure(); + this.#trimPendingH264FramesIfNeeded(); + this.#emitMetricsIfDue(); } - #trimPendingH264Frames(): void { - if (this.#pendingH264Frames.length <= MAX_H264_PENDING_FRAMES) { + #trimPendingH264FramesIfNeeded(): void { + const queueSpanMs = h264QueueSpanMs(this.#pendingH264Frames); + const hardLimitExceeded = isH264HardLimitExceeded( + this.#pendingH264Frames.length, + queueSpanMs, + ); + const pressureTrim = + this.#h264Pressure.mode === 'degraded' && + (this.#pendingH264Frames.length > 36 || queueSpanMs > 250); + if (!hardLimitExceeded && !pressureTrim) { return; } - const latestKeyIndex = findLatestH264KeyFrameIndex(this.#pendingH264Frames); - if (latestKeyIndex > 0) { - this.#pendingH264Frames = this.#pendingH264Frames.slice(latestKeyIndex); + + const selection = selectLatestCompleteH264Gop( + this.#pendingH264Frames, + this.#h264RecentConfig, + ); + if (selection.resync) { + this.#pendingH264Frames = selection.frames; + this.#h264NeedsResync = true; + this.#droppedH264Frames += selection.droppedFrames; } - if (this.#pendingH264Frames.length > MAX_H264_PENDING_FRAMES) { - this.#pendingH264Frames = this.#pendingH264Frames.slice(-MAX_H264_PENDING_FRAMES); + + if ( + isH264HardLimitExceeded( + this.#pendingH264Frames.length, + h264QueueSpanMs(this.#pendingH264Frames), + ) + ) { + this.#waitForNextH264Idr(); + return; + } + + if (!selection.resync) { + // No newer complete GOP exists. Preserve every dependent delta in the + // current GOP while pressure is soft. The hard-limit branch above drops + // the complete backlog rather than decoding a truncated dependency chain. + return; } } + #waitForNextH264Idr(): void { + const plan = applyH264HardLimit(this.#pendingH264Frames, true); + this.#droppedH264Frames += plan.droppedFrames; + this.#pendingH264Frames = plan.frames; + this.#h264WaitingForIdr = true; + this.#h264ConfigBeforeIdr = [...this.#h264RecentConfig]; + this.#h264NeedsResync = true; + this.#updateH264Pressure(); + this.#emitMetricsIfDue(true); + } + #takeNextFrame(): ImageWorkerFrameEnvelope | null { const h264Frame = this.#pendingH264Frames.shift(); if (h264Frame) { @@ -379,10 +560,18 @@ class ImageRenderWorkerRuntime { return; } this.#isProcessing = true; + const epoch = this.#epoch; try { let frame: ImageWorkerFrameEnvelope | null; while ((frame = this.#takeNextFrame())) { - await this.#decodeAndRender(frame); + if (epoch !== this.#epoch) { + break; + } + if (isH264Frame(frame) && this.#h264NeedsResync) { + this.#decoder.reset(); + this.#h264NeedsResync = false; + } + await this.#decodeAndRender(frame, epoch); if (this.#haltUntilReset) { this.#pendingFrame = null; this.#pendingH264Frames = []; @@ -391,10 +580,13 @@ class ImageRenderWorkerRuntime { } } finally { this.#isProcessing = false; + if (this.#pendingFrame || this.#pendingH264Frames.length > 0) { + void this.#drainLatestFrame(); + } } } - async #decodeAndRender(frame: ImageWorkerFrameEnvelope): Promise { + async #decodeAndRender(frame: ImageWorkerFrameEnvelope, epoch: number): Promise { this.#emitStatus({ phase: 'decoding', receiveTime: frame.receiveTime }); try { if (frame.kind === 'compressed') { @@ -422,11 +614,40 @@ class ImageRenderWorkerRuntime { const sortKey = timeToKey(frame.receiveTime); if (kind === 'h264') { + const decodeStartedAt = performance.now(); const videoFrame = await this.#decoder.decodeFrame(bytes, sortKey); + if (!videoFrame) { + this.#h264DecodeMs = updateDecodeDurationEwma( + this.#h264DecodeMs, + performance.now() - decodeStartedAt, + ); + this.#emitMetricsIfDue(); + return; + } try { + if (epoch !== this.#epoch) { + return; + } + this.#h264DecodeMs = updateDecodeDurationEwma( + this.#h264DecodeMs, + performance.now() - decodeStartedAt, + ); + this.#updateH264Pressure(); + const now = performance.now(); + const shouldRender = + this.#h264Pressure.mode !== 'degraded' || + now - this.#lastH264RenderAt >= H264_DEGRADED_RENDER_INTERVAL_MS; + if (!shouldRender) { + this.#droppedH264Frames += 1; + this.#emitMetricsIfDue(); + return; + } + const width = videoFrame.displayWidth || videoFrame.codedWidth; const height = videoFrame.displayHeight || videoFrame.codedHeight; this.#drawCanvasImageSource(videoFrame, width, height); + this.#lastH264RenderAt = now; + this.#renderedH264Frames += 1; this.#emitStatus({ phase: 'ready', width, @@ -434,14 +655,22 @@ class ImageRenderWorkerRuntime { encoding: frame.format, receiveTime: frame.receiveTime, }); - this.#disposeCachedBitmap(); - this.#cachedFrame = null; - try { - const bitmap = await createImageBitmap(videoFrame); - this.#storeBitmap(bitmap, width, height, frame.format, frame.receiveTime); - } catch { - // The frame is already on screen; cache creation is only for cheap redraws. + // Bitmap caching is only for resize/option redraws. Sampling it at + // low frequency avoids a GPU copy for every decoded video frame. + if (this.#h264Pressure.mode === 'normal' && now - this.#lastH264BitmapAt >= 500) { + try { + const bitmap = await createImageBitmap(videoFrame); + if (epoch === this.#epoch) { + this.#storeBitmap(bitmap, width, height, frame.format, frame.receiveTime); + this.#lastH264BitmapAt = now; + } else { + bitmap.close(); + } + } catch { + // The frame is already on screen; cache creation is optional. + } } + this.#emitMetricsIfDue(); } finally { videoFrame.close(); } @@ -499,6 +728,42 @@ class ImageRenderWorkerRuntime { data: bytes, }); } catch (error) { + if (epoch !== this.#epoch) { + return; + } + if (isH264Frame(frame)) { + // Decoder failures usually invalidate the current GOP. Preserve the + // current canvas and resume only when a real IDR arrives. + this.#decoder.reset(); + const hasPendingIdr = this.#pendingH264Frames.some( + (pending) => isH264Frame(pending) && containsH264IdrNal(pending.data), + ); + if (hasPendingIdr) { + const recovery = selectLatestCompleteH264Gop( + this.#pendingH264Frames, + this.#h264RecentConfig, + true, + ); + this.#pendingH264Frames = recovery.frames; + this.#h264WaitingForIdr = false; + this.#h264NeedsResync = true; + this.#droppedH264Frames += recovery.droppedFrames + 1; + } else { + this.#droppedH264Frames += this.#pendingH264Frames.length + 1; + this.#pendingH264Frames = []; + this.#h264WaitingForIdr = true; + this.#h264ConfigBeforeIdr = [...this.#h264RecentConfig]; + this.#h264NeedsResync = false; + } + if (this.#renderedH264Frames === 0 && !this.#cachedFrame) { + this.#emitStatus({ + phase: 'error', + message: error instanceof Error ? error.message : String(error), + }); + } + this.#emitMetricsIfDue(true); + return; + } this.#haltUntilReset = true; this.#emitStatus({ phase: 'error', @@ -507,6 +772,52 @@ class ImageRenderWorkerRuntime { } } + #updateH264Pressure(): void { + const previousMode = this.#h264Pressure.mode; + this.#h264Pressure = updateH264Pressure(this.#h264Pressure, { + queueFrames: this.#pendingH264Frames.length, + queueSpanMs: h264QueueSpanMs(this.#pendingH264Frames), + decodeMs: this.#h264DecodeMs, + }); + if (previousMode !== this.#h264Pressure.mode) { + this.#applyViewport(); + this.#redrawCachedFrame(); + this.#emitMetricsIfDue(true); + } + } + + #resetH264RuntimeState(): void { + this.#h264Pressure = initialH264PressureState(); + this.#h264DecodeMs = 0; + this.#h264WaitingForIdr = false; + this.#h264ConfigBeforeIdr = []; + this.#h264RecentConfig = []; + this.#h264NeedsResync = false; + this.#lastH264RenderAt = -Infinity; + this.#lastH264BitmapAt = -Infinity; + this.#droppedH264Frames = 0; + this.#renderedH264Frames = 0; + this.#lastMetricsAt = -Infinity; + } + + #emitMetricsIfDue(force = false): void { + const now = performance.now(); + if (!force && now - this.#lastMetricsAt < METRICS_INTERVAL_MS) { + return; + } + this.#lastMetricsAt = now; + const metrics: ImageRenderMetrics = { + pressureMode: this.#h264Pressure.mode, + queueFrames: this.#pendingH264Frames.length, + queueSpanMs: h264QueueSpanMs(this.#pendingH264Frames), + decodeMs: this.#h264DecodeMs, + droppedFrames: this.#droppedH264Frames, + renderedFrames: this.#renderedH264Frames, + codec: this.#decoder.codec, + }; + workerScope.postMessage({ type: 'metrics', metrics } satisfies ImageRenderWorkerEvent); + } + #renderRawFrame(frame: { receiveTime: Time; encoding: string; @@ -711,12 +1022,14 @@ class ImageRenderWorkerRuntime { const drawHeight = Math.max(1, sourceHeight * scale); ctx.save(); - ctx.setTransform(this.#viewport.devicePixelRatio, 0, 0, this.#viewport.devicePixelRatio, 0, 0); + const renderDpr = this.#renderDevicePixelRatio(); + ctx.setTransform(renderDpr, 0, 0, renderDpr, 0, 0); ctx.clearRect(0, 0, viewportWidth, viewportHeight); ctx.fillStyle = this.#renderOptions.backgroundColor; ctx.fillRect(0, 0, viewportWidth, viewportHeight); ctx.imageSmoothingEnabled = this.#renderOptions.smoothing; - ctx.imageSmoothingQuality = this.#renderOptions.smoothing ? 'high' : 'low'; + ctx.imageSmoothingQuality = + this.#renderOptions.smoothing && this.#h264Pressure.mode === 'normal' ? 'high' : 'low'; ctx.translate(viewportWidth / 2, viewportHeight / 2); ctx.rotate((rotDeg * Math.PI) / 180); ctx.scale(this.#renderOptions.flipHorizontal ? -1 : 1, this.#renderOptions.flipVertical ? -1 : 1); @@ -728,8 +1041,9 @@ class ImageRenderWorkerRuntime { if (!this.#canvas) { return; } - const pixelWidth = Math.max(1, Math.round(Math.max(0, this.#viewport.cssWidth) * Math.max(1, this.#viewport.devicePixelRatio))); - const pixelHeight = Math.max(1, Math.round(Math.max(0, this.#viewport.cssHeight) * Math.max(1, this.#viewport.devicePixelRatio))); + const renderDpr = this.#renderDevicePixelRatio(); + const pixelWidth = Math.max(1, Math.round(Math.max(0, this.#viewport.cssWidth) * renderDpr)); + const pixelHeight = Math.max(1, Math.round(Math.max(0, this.#viewport.cssHeight) * renderDpr)); if (this.#canvas.width !== pixelWidth) { this.#canvas.width = pixelWidth; } @@ -738,6 +1052,11 @@ class ImageRenderWorkerRuntime { } } + #renderDevicePixelRatio(): number { + const dpr = Math.max(1, this.#viewport.devicePixelRatio); + return this.#h264Pressure.mode === 'normal' ? dpr : Math.min(dpr, 1); + } + #clearCanvas(): void { if (!this.#ctx || !this.#canvas) { return; @@ -802,17 +1121,21 @@ function isH264Frame(frame: ImageWorkerFrameEnvelope): boolean { return frame.kind === 'compressed' && getCompressedKind(frame.format) === 'h264'; } -function findLatestH264KeyFrameIndex(frames: ImageWorkerFrameEnvelope[]): number { - for (let i = frames.length - 1; i >= 0; i -= 1) { - const frame = frames[i]; - if (!frame || !isH264Frame(frame)) { - continue; - } - if (getH264ChunkType(frame.data) === 'key') { - return i; - } +function h264QueueSpanMs(frames: ImageWorkerFrameEnvelope[]): number { + if (frames.length < 2) { + return 0; + } + const first = frames.find( + (frame) => !isH264Frame(frame) || !isH264ConfigOnly(frame.data), + ); + const last = frames.findLast( + (frame) => !isH264Frame(frame) || !isH264ConfigOnly(frame.data), + ); + if (!first || !last) { + return 0; } - return -1; + const spanNs = timeToKey(last.receiveTime) - timeToKey(first.receiveTime); + return Math.max(0, Number(spanNs) / 1_000_000); } function timeToKey(time: Time): bigint { diff --git a/src/features/panels/Image/core/h264.test.ts b/src/features/panels/Image/core/h264.test.ts index 2de2ab6..caf7e9e 100644 --- a/src/features/panels/Image/core/h264.test.ts +++ b/src/features/panels/Image/core/h264.test.ts @@ -1,5 +1,12 @@ import { describe, expect, it } from 'vitest'; -import { getH264ChunkType, scanH264NalTypes } from './h264'; +import { + containsH264IdrNal, + getH264ChunkType, + getH264CodecCandidates, + monotonicH264TimestampUs, + parseH264SpsCodec, + scanH264NalTypes, +} from './h264'; describe('H.264 NAL parsing', () => { it('detects IDR frames in Annex-B chunks', () => { @@ -32,4 +39,42 @@ describe('H.264 NAL parsing', () => { expect(scanH264NalTypes(chunk)).toEqual([7, 8, 5]); expect(getH264ChunkType(chunk)).toBe('key'); }); + + it('derives profile, compatibility flags, and level from an Annex-B SPS', () => { + const chunk = new Uint8Array([ + 0, 0, 0, 1, 0x67, 0x64, 0x00, 0x29, 0xac, + 0, 0, 1, 0x68, 0xee, 0x3c, 0x80, + ]); + + expect(parseH264SpsCodec(chunk)).toBe('avc1.640029'); + expect(getH264CodecCandidates(chunk)).toEqual([ + 'avc1.640029', + 'avc1.42E01E', + 'avc1.4D4020', + 'avc1.640028', + ]); + }); + + it('keeps compatibility-normalized and fixed fallbacks for constrained SPS profiles', () => { + const chunk = new Uint8Array([0, 0, 1, 0x67, 0x42, 0xe0, 0x1f]); + expect(getH264CodecCandidates(chunk).slice(0, 2)).toEqual([ + 'avc1.42E01F', + 'avc1.42001F', + ]); + }); + + it('distinguishes SPS/PPS configuration from a true IDR resync point', () => { + expect(containsH264IdrNal(new Uint8Array([0, 0, 1, 0x67, 0x42, 0, 0x1e]))).toBe(false); + expect(containsH264IdrNal(new Uint8Array([0, 0, 1, 0x65, 1]))).toBe(true); + }); + + it('uses source microseconds while forcing repeats and rewinds to stay monotonic', () => { + const first = monotonicH264TimestampUs(1_234_567_890n, -1); + const repeated = monotonicH264TimestampUs(1_234_567_890n, first); + const rewound = monotonicH264TimestampUs(1_000_000_000n, repeated); + + expect(first).toBe(1_234_567); + expect(repeated).toBe(1_234_568); + expect(rewound).toBe(1_234_569); + }); }); diff --git a/src/features/panels/Image/core/h264.ts b/src/features/panels/Image/core/h264.ts index 89feba6..0838784 100644 --- a/src/features/panels/Image/core/h264.ts +++ b/src/features/panels/Image/core/h264.ts @@ -2,6 +2,11 @@ export function getH264ChunkType(data: Uint8Array): 'key' | 'delta' { return containsH264KeyNal(data) ? 'key' : 'delta'; } +/** IDR slices are the only NAL units that are safe random-access points. */ +export function containsH264IdrNal(data: Uint8Array): boolean { + return scanH264NalTypes(data).includes(5); +} + export function containsH264KeyNal(data: Uint8Array): boolean { for (const nalType of scanH264NalTypes(data)) { // IDR slices are random access points. Treat SPS/PPS as key chunks too so @@ -13,8 +18,53 @@ export function containsH264KeyNal(data: Uint8Array): boolean { return false; } +/** + * Derive an RFC 6381 AVC codec string from the first Annex-B SPS. + * The three bytes after the SPS NAL header are profile_idc, + * constraint_set flags/profile compatibility, and level_idc. + */ +export function parseH264SpsCodec(data: Uint8Array): string | null { + for (const nalOffset of scanH264NalOffsets(data)) { + if ((data[nalOffset] & 0x1f) !== 7 || nalOffset + 3 >= data.byteLength) { + continue; + } + const profile = data[nalOffset + 1]; + const compatibility = data[nalOffset + 2]; + const level = data[nalOffset + 3]; + return `avc1.${hexByte(profile)}${hexByte(compatibility)}${hexByte(level)}`; + } + return null; +} + +/** + * Decoder candidates ordered from stream-specific to broadly compatible. + * Chromium accepts Annex-B chunks when no AVCDecoderConfigurationRecord is + * supplied, so a codec string is sufficient for in-band SPS/PPS streams. + */ +export function getH264CodecCandidates(data: Uint8Array): string[] { + const parsed = parseH264SpsCodec(data); + const candidates = [ + parsed, + parsed ? `avc1.${parsed.slice(5, 7)}00${parsed.slice(-2)}` : null, + 'avc1.42E01E', + 'avc1.4D4020', + 'avc1.640028', + ]; + return [...new Set(candidates.filter((candidate): candidate is string => candidate != null))]; +} + +export function monotonicH264TimestampUs(timeNs: bigint, previousUs: number): number { + const sourceUs = Number(timeNs / 1_000n); + return Math.max(sourceUs, previousUs + 1); +} + export function scanH264NalTypes(data: Uint8Array): number[] { - const types: number[] = []; + const offsets = scanH264NalOffsets(data); + return offsets.map((offset) => data[offset] & 0x1f); +} + +function scanH264NalOffsets(data: Uint8Array): number[] { + const offsets: number[] = []; let offset = 0; while (offset < data.byteLength - 3) { const start = findAnnexBStartCode(data, offset); @@ -24,14 +74,14 @@ export function scanH264NalTypes(data: Uint8Array): number[] { const prefixLength = data[start + 2] === 1 ? 3 : 4; const nalOffset = start + prefixLength; if (nalOffset < data.byteLength) { - types.push(data[nalOffset] & 0x1f); + offsets.push(nalOffset); } offset = nalOffset + 1; } - if (types.length > 0) { - return types; + if (offsets.length > 0) { + return offsets; } - return data.byteLength > 0 ? [data[0] & 0x1f] : []; + return data.byteLength > 0 ? [0] : []; } function findAnnexBStartCode(data: Uint8Array, offset: number): number { @@ -45,3 +95,7 @@ function findAnnexBStartCode(data: Uint8Array, offset: number): number { } return -1; } + +function hexByte(value: number): string { + return value.toString(16).padStart(2, '0').toUpperCase(); +} diff --git a/src/features/panels/Image/core/h264Backpressure.test.ts b/src/features/panels/Image/core/h264Backpressure.test.ts new file mode 100644 index 0000000..400ca43 --- /dev/null +++ b/src/features/panels/Image/core/h264Backpressure.test.ts @@ -0,0 +1,56 @@ +import { describe, expect, it } from 'vitest'; +import { + H264_MAX_PENDING_FRAMES, + H264_MAX_PENDING_SPAN_MS, + initialH264PressureState, + isH264HardLimitExceeded, + updateDecodeDurationEwma, + updateH264Pressure, +} from './h264Backpressure'; + +describe('H.264 adaptive backpressure', () => { + it('treats frame count and queue span as strict hard bounds', () => { + expect(isH264HardLimitExceeded(H264_MAX_PENDING_FRAMES, H264_MAX_PENDING_SPAN_MS)).toBe(false); + expect(isH264HardLimitExceeded(H264_MAX_PENDING_FRAMES + 1, 0)).toBe(true); + expect(isH264HardLimitExceeded(1, H264_MAX_PENDING_SPAN_MS + 1)).toBe(true); + }); + + it('enters degraded mode from queue time span even below the frame bound', () => { + const next = updateH264Pressure(initialH264PressureState(), { + queueFrames: 20, + queueSpanMs: 400, + decodeMs: 10, + }); + expect(next.mode).toBe('degraded'); + }); + + it('uses hysteresis before returning to normal', () => { + let state = updateH264Pressure(initialH264PressureState(), { + queueFrames: 80, + queueSpanMs: 500, + decodeMs: 60, + }); + state = updateH264Pressure(state, { queueFrames: 2, queueSpanMs: 20, decodeMs: 10 }); + expect(state.mode).toBe('recovery'); + + for (let i = 0; i < 10; i++) { + state = updateH264Pressure(state, { queueFrames: 2, queueSpanMs: 20, decodeMs: 10 }); + } + expect(state.mode).toBe('recovery'); + state = updateH264Pressure(state, { queueFrames: 2, queueSpanMs: 20, decodeMs: 10 }); + expect(state.mode).toBe('normal'); + }); + + it('relapses quickly when recovery pressure rises again', () => { + let state = { mode: 'degraded' as const, healthySamples: 0 }; + state = updateH264Pressure(state, { queueFrames: 0, queueSpanMs: 0, decodeMs: 5 }); + expect(state.mode).toBe('recovery'); + state = updateH264Pressure(state, { queueFrames: 45, queueSpanMs: 300, decodeMs: 20 }); + expect(state.mode).toBe('degraded'); + }); + + it('smooths decode duration samples', () => { + expect(updateDecodeDurationEwma(20, 40)).toBe(24); + expect(updateDecodeDurationEwma(0, 15)).toBe(15); + }); +}); diff --git a/src/features/panels/Image/core/h264Backpressure.ts b/src/features/panels/Image/core/h264Backpressure.ts new file mode 100644 index 0000000..b00000b --- /dev/null +++ b/src/features/panels/Image/core/h264Backpressure.ts @@ -0,0 +1,79 @@ +export type H264PressureMode = 'normal' | 'degraded' | 'recovery'; + +export interface H264PressureState { + mode: H264PressureMode; + healthySamples: number; +} + +export interface H264PressureObservation { + queueFrames: number; + queueSpanMs: number; + decodeMs: number; +} + +/** + * Hard pending-queue bounds. If the newest complete GOP still exceeds either + * bound, the worker drops that backlog and waits for the next real IDR. + */ +export const H264_MAX_PENDING_FRAMES = 120; +export const H264_MAX_PENDING_SPAN_MS = 1_000; +export const H264_DEGRADED_RENDER_INTERVAL_MS = 80; + +const ENTER_DEGRADED = { frames: 72, spanMs: 350, decodeMs: 55 }; +const ENTER_RECOVERY = { frames: 18, spanMs: 120, decodeMs: 32 }; +const RELAPSE = { frames: 40, spanMs: 250, decodeMs: 45 }; +const RECOVERY_SAMPLES = 12; + +export function initialH264PressureState(): H264PressureState { + return { mode: 'normal', healthySamples: 0 }; +} + +export function isH264HardLimitExceeded(queueFrames: number, queueSpanMs: number): boolean { + return queueFrames > H264_MAX_PENDING_FRAMES || queueSpanMs > H264_MAX_PENDING_SPAN_MS; +} + +/** + * Hysteretic pressure controller. Queue age is the primary signal while the + * decode EWMA catches expensive streams before the bounded queue overflows. + */ +export function updateH264Pressure( + state: H264PressureState, + observation: H264PressureObservation, +): H264PressureState { + const overloaded = + observation.queueFrames >= ENTER_DEGRADED.frames || + observation.queueSpanMs >= ENTER_DEGRADED.spanMs || + observation.decodeMs >= ENTER_DEGRADED.decodeMs; + const healthy = + observation.queueFrames <= ENTER_RECOVERY.frames && + observation.queueSpanMs <= ENTER_RECOVERY.spanMs && + observation.decodeMs <= ENTER_RECOVERY.decodeMs; + const relapsed = + observation.queueFrames >= RELAPSE.frames || + observation.queueSpanMs >= RELAPSE.spanMs || + observation.decodeMs >= RELAPSE.decodeMs; + + if (state.mode === 'normal') { + return overloaded ? { mode: 'degraded', healthySamples: 0 } : state; + } + if (state.mode === 'degraded') { + return healthy ? { mode: 'recovery', healthySamples: 1 } : state; + } + if (relapsed) { + return { mode: 'degraded', healthySamples: 0 }; + } + if (!healthy) { + return { mode: 'recovery', healthySamples: 0 }; + } + const healthySamples = state.healthySamples + 1; + return healthySamples >= RECOVERY_SAMPLES + ? { mode: 'normal', healthySamples: 0 } + : { mode: 'recovery', healthySamples }; +} + +export function updateDecodeDurationEwma(previousMs: number, sampleMs: number): number { + if (!Number.isFinite(sampleMs) || sampleMs < 0) { + return previousMs; + } + return previousMs === 0 ? sampleMs : previousMs * 0.8 + sampleMs * 0.2; +} diff --git a/src/features/panels/Image/core/h264Queue.test.ts b/src/features/panels/Image/core/h264Queue.test.ts new file mode 100644 index 0000000..1b4b9e8 --- /dev/null +++ b/src/features/panels/Image/core/h264Queue.test.ts @@ -0,0 +1,128 @@ +import { describe, expect, it } from 'vitest'; +import { + applyH264HardLimit, + selectLatestCompleteH264Gop, + updateH264ConfigPackets, +} from './h264Queue'; + +const sps = new Uint8Array([0, 0, 1, 0x67, 0x42, 0, 0x1e]); +const pps = new Uint8Array([0, 0, 1, 0x68, 0xce, 0x3c]); +const idr = new Uint8Array([0, 0, 1, 0x65, 1]); +const delta = new Uint8Array([0, 0, 1, 0x41, 2]); + +type Frame = { id: string; data: Uint8Array }; + +function frame(id: string, data: Uint8Array): Frame { + return { id, data }; +} + +describe('H.264 GOP queue selection', () => { + it('retains split SPS and PPS packets separately and in order while waiting for IDR', () => { + let config: Frame[] = []; + config = updateH264ConfigPackets(config, frame('sps', sps)); + config = updateH264ConfigPackets(config, frame('delta', delta)); + config = updateH264ConfigPackets(config, frame('pps', pps)); + + expect(config.map(({ id }) => id)).toEqual(['sps', 'pps']); + }); + + it('tracks the latest split SPS/PPS generation for future GOP resets', () => { + let config = [frame('old-sps', sps), frame('old-pps', pps)]; + config = updateH264ConfigPackets(config, frame('new-sps', sps)); + config = updateH264ConfigPackets(config, frame('new-pps', pps)); + + expect(config.map(({ id }) => id)).toEqual(['new-sps', 'new-pps']); + }); + + it('jumps over whole old GOPs to the latest IDR and keeps every new-GOP delta', () => { + const frames = [ + frame('sps', sps), + frame('pps', pps), + frame('idr-1', idr), + frame('delta-1a', delta), + frame('delta-1b', delta), + frame('idr-2', idr), + frame('delta-2a', delta), + frame('delta-2b', delta), + ]; + + const selected = selectLatestCompleteH264Gop(frames); + + expect(selected.resync).toBe(true); + expect(selected.droppedFrames).toBe(3); + expect(selected.frames.map(({ id }) => id)).toEqual([ + 'sps', + 'pps', + 'idr-2', + 'delta-2a', + 'delta-2b', + ]); + expect(selected.frames.length).toBeLessThan(frames.length); + }); + + it('does not truncate the sole complete GOP during soft pressure', () => { + const frames = [ + frame('idr', idr), + ...Array.from({ length: 80 }, (_, index) => frame(`delta-${index}`, delta)), + ]; + + const selected = selectLatestCompleteH264Gop(frames); + + expect(selected.resync).toBe(false); + expect(selected.droppedFrames).toBe(0); + expect(selected.frames.map(({ id }) => id)).toEqual(frames.map(({ id }) => id)); + }); + + it('drops an entire hard-overflow backlog and waits instead of returning truncated deltas', () => { + const frames = [ + frame('idr', idr), + ...Array.from({ length: 200 }, (_, index) => frame(`delta-${index}`, delta)), + ]; + + const plan = applyH264HardLimit(frames, true); + + expect(plan.waitForIdr).toBe(true); + expect(plan.droppedFrames).toBe(frames.length); + expect(plan.frames).toEqual([]); + }); + + it('can safely reset at the sole IDR by prepending previously consumed config', () => { + const frames = [frame('idr', idr), frame('delta', delta)]; + const config = [frame('sps', sps), frame('pps', pps)]; + + const selected = selectLatestCompleteH264Gop(frames, config, true); + + expect(selected.resync).toBe(true); + expect(selected.droppedFrames).toBe(0); + expect(selected.frames.map(({ id }) => id)).toEqual(['sps', 'pps', 'idr', 'delta']); + }); + + it('prepends previously consumed split config when jumping to a queued IDR', () => { + const recentConfig = [frame('sps', sps), frame('pps', pps)]; + const frames = [ + frame('old-delta-1', delta), + frame('old-delta-2', delta), + frame('new-idr', idr), + frame('new-delta', delta), + ]; + + const selected = selectLatestCompleteH264Gop(frames, recentConfig); + + expect(selected.resync).toBe(true); + expect(selected.droppedFrames).toBe(2); + expect(selected.frames.map(({ id }) => id)).toEqual([ + 'sps', + 'pps', + 'new-idr', + 'new-delta', + ]); + }); + + it('does not drop delta-only backlog from a GOP whose decoder state is still valid', () => { + const frames = [frame('delta-1', delta), frame('delta-2', delta)]; + const selected = selectLatestCompleteH264Gop(frames); + + expect(selected.resync).toBe(false); + expect(selected.frames).toEqual(frames); + }); +}); diff --git a/src/features/panels/Image/core/h264Queue.ts b/src/features/panels/Image/core/h264Queue.ts new file mode 100644 index 0000000..fb9c608 --- /dev/null +++ b/src/features/panels/Image/core/h264Queue.ts @@ -0,0 +1,121 @@ +import { containsH264IdrNal, scanH264NalTypes } from './h264'; + +export interface H264QueueEntry { + data: Uint8Array; +} + +export interface H264GopSelection { + frames: T[]; + droppedFrames: number; + resync: boolean; +} + +export interface H264HardLimitPlan { + frames: T[]; + droppedFrames: number; + waitForIdr: boolean; +} + +/** Hard overflow never returns a truncated dependency chain for decoding. */ +export function applyH264HardLimit( + frames: readonly T[], + hardLimitExceeded: boolean, +): H264HardLimitPlan { + return hardLimitExceeded + ? { frames: [], droppedFrames: frames.length, waitForIdr: true } + : { frames: [...frames], droppedFrames: 0, waitForIdr: false }; +} + +export function isH264ConfigOnly(data: Uint8Array): boolean { + const nalTypes = scanH264NalTypes(data); + const hasConfig = nalTypes.some((type) => type === 7 || type === 8); + const hasVcl = nalTypes.some((type) => type === 1 || type === 5); + return hasConfig && !hasVcl; +} + +/** Track the latest ordered SPS/PPS generation without collapsing split packets. */ +export function updateH264ConfigPackets( + packets: readonly T[], + frame: T, +): T[] { + if (!isH264ConfigOnly(frame.data)) { + return [...packets]; + } + return scanH264NalTypes(frame.data).includes(7) ? [frame] : [...packets, frame]; +} + +/** + * Select the newest complete random-access suffix. + * + * Delta frames are never removed from within the selected GOP. With no newer + * IDR (including when the only IDR is index 0), the queue is returned intact. + */ +export function selectLatestCompleteH264Gop( + frames: readonly T[], + fallbackConfig: readonly T[] = [], + forceResync = false, +): H264GopSelection { + const latestIdrIndex = findLatestH264IdrIndex(frames); + if (latestIdrIndex < 0) { + return { frames: [...frames], droppedFrames: 0, resync: false }; + } + if (latestIdrIndex === 0) { + if (!forceResync) { + return { frames: [...frames], droppedFrames: 0, resync: false }; + } + const idrContainsSps = scanH264NalTypes(frames[0].data).includes(7); + return { + frames: idrContainsSps ? [...frames] : [...fallbackConfig, ...frames], + droppedFrames: 0, + resync: true, + }; + } + + const idrNalTypes = scanH264NalTypes(frames[latestIdrIndex].data); + const inQueueConfig = idrNalTypes.includes(7) + ? [] + : findLatestCompleteConfig(frames, latestIdrIndex); + const configFrames = + idrNalTypes.includes(7) || inQueueConfig.length > 0 + ? inQueueConfig + : [...fallbackConfig]; + const selected = [...configFrames, ...frames.slice(latestIdrIndex)]; + const droppedFrames = latestIdrIndex - inQueueConfig.length; + if (droppedFrames === 0) { + return { frames: [...frames], droppedFrames: 0, resync: false }; + } + return { + frames: selected, + droppedFrames, + resync: true, + }; +} + +function findLatestH264IdrIndex(frames: readonly T[]): number { + for (let index = frames.length - 1; index >= 0; index -= 1) { + if (containsH264IdrNal(frames[index].data)) { + return index; + } + } + return -1; +} + +function findLatestCompleteConfig( + frames: readonly T[], + endIndex: number, +): T[] { + let latestSpsIndex = -1; + for (let index = endIndex - 1; index >= 0; index -= 1) { + const frame = frames[index]; + if (isH264ConfigOnly(frame.data) && scanH264NalTypes(frame.data).includes(7)) { + latestSpsIndex = index; + break; + } + } + if (latestSpsIndex < 0) { + return []; + } + return frames + .slice(latestSpsIndex, endIndex) + .filter((frame) => isH264ConfigOnly(frame.data)); +} diff --git a/src/features/panels/Image/core/h264SeekRepair.test.ts b/src/features/panels/Image/core/h264SeekRepair.test.ts index 3e45365..a631a7a 100644 --- a/src/features/panels/Image/core/h264SeekRepair.test.ts +++ b/src/features/panels/Image/core/h264SeekRepair.test.ts @@ -1,9 +1,18 @@ import { describe, expect, it } from 'vitest'; +import type { Player } from '@/core/types/player'; import type { MessageEvent as RosMessageEvent } from '@/core/types/ros'; -import { findLatestH264KeyFrameIndex, selectH264SeekRepairFrames } from './h264SeekRepair'; +import { + H264_SEEK_MAX_FRAMES, + findLatestH264KeyFrameIndex, + repairH264Seek, + selectH264SeekRepairFrames, +} from './h264SeekRepair'; const keyChunk = new Uint8Array([0, 0, 0, 1, 0x67, 1, 2, 0, 0, 1, 0x65, 3, 4]); const deltaChunk = new Uint8Array([0, 0, 1, 0x41, 9, 9]); +const spsChunk = new Uint8Array([0, 0, 1, 0x67, 0x42, 0, 0x1e]); +const ppsChunk = new Uint8Array([0, 0, 1, 0x68, 0xce, 0x3c]); +const idrChunk = new Uint8Array([0, 0, 1, 0x65, 3, 4]); function makeEvent(sec: number, data: Uint8Array, format = 'h264'): RosMessageEvent { const receiveTime = { sec, nsec: 0 }; @@ -22,6 +31,17 @@ describe('h264SeekRepair', () => { expect(findLatestH264KeyFrameIndex(messages)).toBe(0); }); + it('does not treat standalone SPS/PPS packets as random-access points', () => { + const messages = [ + makeEvent(1, idrChunk), + makeEvent(2, deltaChunk), + makeEvent(3, spsChunk), + makeEvent(4, ppsChunk), + ]; + expect(findLatestH264KeyFrameIndex(messages)).toBe(0); + expect(findLatestH264KeyFrameIndex([makeEvent(1, spsChunk), makeEvent(2, ppsChunk)])).toBe(-1); + }); + it('selectH264SeekRepairFrames returns frames from keyframe through target time', () => { const messages = [ makeEvent(1, keyChunk), @@ -51,4 +71,101 @@ describe('h264SeekRepair', () => { const messages = [makeEvent(1, deltaChunk), makeEvent(2, deltaChunk)]; expect(selectH264SeekRepairFrames(messages, { sec: 2, nsec: 0 })).toEqual([]); }); + + it('prepends ordered split SPS/PPS packets to the selected IDR GOP', () => { + const messages = [ + makeEvent(1, spsChunk), + makeEvent(2, ppsChunk), + makeEvent(3, idrChunk), + makeEvent(4, deltaChunk), + ]; + + const repair = selectH264SeekRepairFrames(messages, { sec: 4, nsec: 0 }); + + expect(repair.map((event) => event.receiveTime.sec)).toEqual([1, 2, 3, 4]); + }); + + it('skips the older GOP only at the latest real IDR boundary', () => { + const messages = [ + makeEvent(1, spsChunk), + makeEvent(2, ppsChunk), + makeEvent(3, idrChunk), + makeEvent(4, deltaChunk), + makeEvent(5, spsChunk), + makeEvent(6, ppsChunk), + makeEvent(7, idrChunk), + makeEvent(8, deltaChunk), + ]; + + const repair = selectH264SeekRepairFrames(messages, { sec: 8, nsec: 0 }); + + expect(repair.map((event) => event.receiveTime.sec)).toEqual([5, 6, 7, 8]); + }); + + it('caps a long GOP at a safe IDR-prefixed decodable prefix', () => { + const messages = [ + makeEvent(1, spsChunk), + makeEvent(2, ppsChunk), + makeEvent(3, idrChunk), + ...Array.from({ length: 1_000 }, (_, index) => makeEvent(index + 4, deltaChunk)), + ]; + + const repair = selectH264SeekRepairFrames(messages, { sec: 2_000, nsec: 0 }); + + expect(repair).toHaveLength(H264_SEEK_MAX_FRAMES); + expect(repair.slice(0, 3).map((event) => event.receiveTime.sec)).toEqual([1, 2, 3]); + expect(findLatestH264KeyFrameIndex(repair)).toBe(2); + expect(repair.at(-1)?.receiveTime.sec).toBe(H264_SEEK_MAX_FRAMES); + }); + + it('posts at most 180 frame messages for a long-GOP seek repair', async () => { + const messages = [ + makeEvent(1, keyChunk), + ...Array.from({ length: 1_000 }, (_, index) => makeEvent(index + 2, deltaChunk)), + ]; + let framePosts = 0; + const worker = { + postMessage(request: { type?: string }) { + if (request.type === 'frame') { + framePosts += 1; + } + }, + } as unknown as Worker; + const player = { + getMessagesInTimeRange: async () => messages, + } as unknown as Player; + + await expect( + repairH264Seek(player, worker, '/camera/video', { sec: 2_000, nsec: 0 }), + ).resolves.toBe(true); + expect(framePosts).toBe(H264_SEEK_MAX_FRAMES); + }); + + it('keeps range-query payloads borrowed while posting seek-repair frames', async () => { + const keyPayload = keyChunk.slice(); + const deltaPayload = deltaChunk.slice(); + const messages = [makeEvent(1, keyPayload), makeEvent(2, deltaPayload)]; + const postedFrames: unknown[] = []; + const worker = { + postMessage(request: unknown, transfer: Transferable[] = []) { + const cloned = structuredClone(request, { transfer }); + if ((cloned as { type?: string }).type === 'frame') { + postedFrames.push(cloned); + } + }, + } as unknown as Worker; + const player = { + getMessagesInTimeRange: async () => messages, + } as unknown as Player; + + await expect(repairH264Seek(player, worker, '/camera/video', { sec: 2, nsec: 0 })).resolves.toBe( + true, + ); + + expect(postedFrames).toHaveLength(2); + expect(Array.from(keyPayload)).toEqual(Array.from(keyChunk)); + expect(Array.from(deltaPayload)).toEqual(Array.from(deltaChunk)); + expect(keyPayload.byteLength).toBeGreaterThan(0); + expect(deltaPayload.byteLength).toBeGreaterThan(0); + }); }); diff --git a/src/features/panels/Image/core/h264SeekRepair.ts b/src/features/panels/Image/core/h264SeekRepair.ts index e6ef1b9..c825ed5 100644 --- a/src/features/panels/Image/core/h264SeekRepair.ts +++ b/src/features/panels/Image/core/h264SeekRepair.ts @@ -1,14 +1,15 @@ import type { Player } from '@/core/types/player'; import type { MessageEvent as RosMessageEvent, Time } from '@/core/types/ros'; import { addMs, toNano } from '@/shared/utils/time'; -import { getH264ChunkType } from './h264'; +import { containsH264IdrNal } from './h264'; +import { selectLatestCompleteH264Gop } from './h264Queue'; import type { ImageRenderWorkerRequest } from './imageWorkerProtocol'; import { getH264MessagePayload, isH264MessageEvent, toWorkerFrame } from './messageFrameAdapter'; /** Progressive lookback windows when searching for a keyframe before a seek target. */ export const H264_SEEK_WINDOWS_MS = [2000, 5000, 10_000, 30_000] as const; -/** Cap frames fed during seek repair so high-FPS streams stay responsive. */ +/** Maximum frame messages posted for one seek repair. */ export const H264_SEEK_MAX_FRAMES = 180; export function findLatestH264KeyFrameIndex(messages: RosMessageEvent[]): number { @@ -18,7 +19,7 @@ export function findLatestH264KeyFrameIndex(messages: RosMessageEvent[]): number continue; } const payload = getH264MessagePayload(event); - if (payload && getH264ChunkType(payload) === 'key') { + if (payload && containsH264IdrNal(payload)) { return i; } } @@ -43,12 +44,28 @@ export function selectH264SeekRepairFrames( return 0; }); - const keyIndex = findLatestH264KeyFrameIndex(h264Messages); - if (keyIndex < 0) { + const candidates = h264Messages.flatMap((event) => { + const data = getH264MessagePayload(event); + return data ? [{ event, data }] : []; + }); + if (findLatestH264KeyFrameIndex(h264Messages) < 0) { return []; } - return h264Messages.slice(keyIndex).slice(-H264_SEEK_MAX_FRAMES); + return limitH264SeekRepairFrames(selectLatestCompleteH264Gop(candidates).frames) + .map(({ event }) => event); +} + +export function limitH264SeekRepairFrames( + frames: readonly T[], +): T[] { + const idrIndex = frames.findIndex(({ data }) => containsH264IdrNal(data)); + if (idrIndex < 0 || idrIndex >= H264_SEEK_MAX_FRAMES) { + return []; + } + // Keep the decodable prefix from config/real IDR forward. Taking the tail + // would discard dependencies and turn a delta into an invalid access point. + return frames.slice(0, H264_SEEK_MAX_FRAMES); } export async function repairH264Seek( diff --git a/src/features/panels/Image/core/imageTypes.test.ts b/src/features/panels/Image/core/imageTypes.test.ts index 5da4ce5..6ecd027 100644 --- a/src/features/panels/Image/core/imageTypes.test.ts +++ b/src/features/panels/Image/core/imageTypes.test.ts @@ -200,7 +200,7 @@ describe('prepareImageWorkerBytes', () => { const input = new Uint8Array(shared); input.set([1, 2, 3, 4]); - const prepared = prepareImageWorkerBytes(input); + const prepared = prepareImageWorkerBytes(input, { transferOwnership: true }); expect(prepared).not.toBeNull(); expect(prepared!.data).not.toBe(input); @@ -224,11 +224,22 @@ describe('prepareImageWorkerBytes', () => { expect(Array.from(prepared!.data)).toEqual([1, 2, 3]); }); - it('copies sliced ArrayBuffer-backed views into a compact transfer buffer', () => { + it('transfers a full ArrayBuffer-backed view when ownership is explicitly handed off', () => { + const input = new Uint8Array([4, 5, 6]); + + const prepared = prepareImageWorkerBytes(input, { transferOwnership: true }); + + expect(prepared).not.toBeNull(); + expect(prepared!.data).toBe(input); + expect(prepared!.data.buffer).toBe(input.buffer); + expect(prepared!.transfer).toEqual([input.buffer]); + }); + + it('copies sliced ArrayBuffer-backed views into a compact transfer buffer even with ownership', () => { const input = new Uint8Array(new ArrayBuffer(8), 2, 3); input.set([7, 8, 9]); - const prepared = prepareImageWorkerBytes(input); + const prepared = prepareImageWorkerBytes(input, { transferOwnership: true }); expect(prepared).not.toBeNull(); expect(prepared!.data).not.toBe(input); diff --git a/src/features/panels/Image/core/imageTypes.ts b/src/features/panels/Image/core/imageTypes.ts index 3b29880..9076368 100644 --- a/src/features/panels/Image/core/imageTypes.ts +++ b/src/features/panels/Image/core/imageTypes.ts @@ -51,13 +51,32 @@ export function snapshotBytes(data: unknown): Uint8Array | null { return copy; } -export function prepareImageWorkerBytes(data: unknown): { data: Uint8Array; transfer: Transferable[] } | null { +export interface PrepareImageWorkerBytesOptions { + /** + * The caller gives the worker exclusive ownership of this payload. Only a + * full-span ArrayBuffer view can be transferred without first copying. + */ + transferOwnership?: boolean; +} + +export function prepareImageWorkerBytes( + data: unknown, + options: PrepareImageWorkerBytesOptions = {}, +): { data: Uint8Array; transfer: Transferable[] } | null { if (!(data instanceof Uint8Array)) { return null; } - // Always hand the render worker an owned ArrayBuffer. SAB-backed views may - // point into the playback ring, whose slot can be reused before the worker's - // postMessage is processed. + if ( + options.transferOwnership === true && + data.buffer instanceof ArrayBuffer && + data.byteOffset === 0 && + data.byteLength === data.buffer.byteLength + ) { + return { data, transfer: [data.buffer] }; + } + // Borrowed payloads must not be detached. SAB-backed views and sliced views + // may also alias storage that remains in use, so compact-copy them even when + // ownership transfer was requested. const copy = new Uint8Array(data.byteLength); copy.set(data); return { data: copy, transfer: [copy.buffer] }; diff --git a/src/features/panels/Image/core/imageWorkerProtocol.ts b/src/features/panels/Image/core/imageWorkerProtocol.ts index 62da502..84fcb16 100644 --- a/src/features/panels/Image/core/imageWorkerProtocol.ts +++ b/src/features/panels/Image/core/imageWorkerProtocol.ts @@ -1,6 +1,7 @@ import type { Time } from '@/core/types/ros'; import type { RawImageDecodeOptions } from './imageColorMode'; import type { ImageSurfaceStatus } from './imageTypes'; +import type { H264PressureMode } from './h264Backpressure'; export interface ImageRenderOptions { /** CSS color string (e.g. `#ff0000`) used to fill letterbox/pillarbox and idle canvas. */ @@ -64,7 +65,22 @@ export type ImageRenderWorkerRequest = type: 'dispose'; }; -export type ImageRenderWorkerEvent = { - type: 'status'; - status: ImageSurfaceStatus; -}; +export interface ImageRenderMetrics { + pressureMode: H264PressureMode; + queueFrames: number; + queueSpanMs: number; + decodeMs: number; + droppedFrames: number; + renderedFrames: number; + codec?: string; +} + +export type ImageRenderWorkerEvent = + | { + type: 'status'; + status: ImageSurfaceStatus; + } + | { + type: 'metrics'; + metrics: ImageRenderMetrics; + }; diff --git a/src/features/panels/Image/core/messageFrameAdapter.test.ts b/src/features/panels/Image/core/messageFrameAdapter.test.ts index efdb863..b9c4fce 100644 --- a/src/features/panels/Image/core/messageFrameAdapter.test.ts +++ b/src/features/panels/Image/core/messageFrameAdapter.test.ts @@ -56,4 +56,25 @@ describe('messageFrameAdapter', () => { expect(isH264MessageEvent(makeCompressedVideoEvent(payload))).toBe(true); expect(isH264MessageEvent(makeCompressedVideoEvent(payload, 'vp9'))).toBe(false); }); + + it('keeps borrowed message payloads intact by default', () => { + const payload = new Uint8Array([1, 2, 3]); + + const prepared = toWorkerFrame(makeCompressedImageEvent(payload)); + + expect(prepared).not.toBeNull(); + expect(prepared!.frame.data).not.toBe(payload); + expect(prepared!.frame.data.buffer).not.toBe(payload.buffer); + expect(prepared!.transfer).toEqual([prepared!.frame.data.buffer]); + }); + + it('hands off full-span payloads only when ownership is explicit', () => { + const payload = new Uint8Array([1, 2, 3]); + + const prepared = toWorkerFrame(makeCompressedImageEvent(payload), { transferOwnership: true }); + + expect(prepared).not.toBeNull(); + expect(prepared!.frame.data).toBe(payload); + expect(prepared!.transfer).toEqual([payload.buffer]); + }); }); diff --git a/src/features/panels/Image/core/messageFrameAdapter.ts b/src/features/panels/Image/core/messageFrameAdapter.ts index ca8dcfa..e0e0cdc 100644 --- a/src/features/panels/Image/core/messageFrameAdapter.ts +++ b/src/features/panels/Image/core/messageFrameAdapter.ts @@ -13,10 +13,17 @@ export type PreparedImageWorkerFrame = { transfer: Transferable[]; }; -export function toWorkerFrame(messageEvent: RosMessageEvent): PreparedImageWorkerFrame | null { +export interface ToWorkerFrameOptions { + transferOwnership?: boolean; +} + +export function toWorkerFrame( + messageEvent: RosMessageEvent, + options: ToWorkerFrameOptions = {}, +): PreparedImageWorkerFrame | null { const message = messageEvent.message; if (isCompressedFrameMessage(message)) { - const payload = prepareImageWorkerBytes(message.data); + const payload = prepareImageWorkerBytes(message.data, options); if (!payload) { return null; } @@ -31,7 +38,7 @@ export function toWorkerFrame(messageEvent: RosMessageEvent): PreparedImageWorke }; } if (isRawImageMessage(message)) { - const payload = prepareImageWorkerBytes(message.data); + const payload = prepareImageWorkerBytes(message.data, options); if (!payload) { return null; } diff --git a/tests/image-h264.spec.ts b/tests/image-h264.spec.ts index 36a0258..79276ee 100644 --- a/tests/image-h264.spec.ts +++ b/tests/image-h264.spec.ts @@ -16,14 +16,43 @@ test('H.264 CompressedImage decodes without error', async ({ page }) => { await play.click(); } + const progressFill = page.getByTestId('playback-progress-fill'); + const progressWidths: number[] = []; + for (let sample = 0; sample < 6; sample += 1) { + const width = await progressFill.evaluate((element) => + Number.parseFloat((element as HTMLElement).style.width), + ); + expect(Number.isFinite(width)).toBe(true); + progressWidths.push(width); + await page.waitForTimeout(200); + } + const meaningfulAdvances = progressWidths + .slice(1) + .filter((width, index) => width - progressWidths[index] >= 0.1); + expect(meaningfulAdvances.length).toBeGreaterThanOrEqual(2); + expect(Math.max(...progressWidths) - Math.min(...progressWidths)).toBeGreaterThanOrEqual(1); + await expect(page.locator('canvas')).not.toHaveCount(0, { timeout: 90_000 }); const hasDecodeFailure = await page.getByText(/decode failed|could not be decoded/i).count(); expect(hasDecodeFailure).toBe(0); const imageStatus = page.getByTestId('image-panel-status'); - if (await page.getByTestId('image-panel-canvas').isVisible().catch(() => false)) { + const imagePanel = page.getByTestId('image-panel'); + if (await imagePanel.isVisible().catch(() => false)) { await expect(imageStatus).toBeVisible({ timeout: 90_000 }); await expect(imageStatus).toHaveText(/\d+x\d+/); + + await expect(imagePanel).toHaveAttribute('data-h264-pressure', /^(normal|degraded|recovery)$/, { + timeout: 90_000, + }); + const metrics = await imagePanel.evaluate((element) => ({ + queueFrames: Number(element.getAttribute('data-h264-queue-frames')), + droppedFrames: Number(element.getAttribute('data-h264-dropped-frames')), + })); + expect(Number.isInteger(metrics.queueFrames)).toBe(true); + expect(metrics.queueFrames).toBeGreaterThanOrEqual(0); + expect(Number.isInteger(metrics.droppedFrames)).toBe(true); + expect(metrics.droppedFrames).toBeGreaterThanOrEqual(0); } }); From ea279ee73281f3319787632774ff1512a26a19a4 Mon Sep 17 00:00:00 2001 From: joaner <1726541+joaner@users.noreply.github.com> Date: Mon, 13 Jul 2026 20:53:39 +0800 Subject: [PATCH 2/2] chore: release v1.7.3 --- package-lock.json | 4 ++-- package.json | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/package-lock.json b/package-lock.json index eb5c970..1770d3a 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@ioai/rosview", - "version": "1.7.2", + "version": "1.7.3", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@ioai/rosview", - "version": "1.7.2", + "version": "1.7.3", "license": "MIT", "devDependencies": { "@eslint/js": "^9.39.4", diff --git a/package.json b/package.json index a6886fd..99cbf27 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@ioai/rosview", - "version": "1.7.2", + "version": "1.7.3", "description": "High-performance robotics data visualization for MCAP, ROS bag, ROS2 db3, HDF5 and BVH — embeddable React component and standalone SPA", "keywords": [ "ros",