From 85b090a8b92c0f17081ac8dade0282cb3937b2e9 Mon Sep 17 00:00:00 2001 From: ravi Date: Thu, 4 Jun 2026 17:05:58 +0530 Subject: [PATCH] playback issues --- sam-tool-tauri/src-tauri/src/decoder.rs | 14 + sam-tool-tauri/src-tauri/src/lib.rs | 12 +- sam-tool-tauri/src-tauri/src/state.rs | 14 +- sam-tool-tauri/src-tauri/src/stream.rs | 208 +++++++++++++++ sam-tool-tauri/src/ipc.ts | 21 +- sam-tool-tauri/src/modes/ExtractMode.tsx | 249 ++++++++++-------- .../src/services/frameDecoderPool.ts | 118 +++++++++ 7 files changed, 524 insertions(+), 112 deletions(-) create mode 100644 sam-tool-tauri/src-tauri/src/stream.rs create mode 100644 sam-tool-tauri/src/services/frameDecoderPool.ts diff --git a/sam-tool-tauri/src-tauri/src/decoder.rs b/sam-tool-tauri/src-tauri/src/decoder.rs index 125f03f..92da6a9 100644 --- a/sam-tool-tauri/src-tauri/src/decoder.rs +++ b/sam-tool-tauri/src-tauri/src/decoder.rs @@ -309,6 +309,20 @@ impl Drop for DecoderSession { } } +/// Pack a frame for the JS wire: 8-byte little-endian i64 frame index +/// followed by the raw JPEG. The backend is authoritative on which frame +/// actually decoded — JS reads the prefix to bind paints to the real frame +/// number rather than the optimistic request, which matters when codec +/// keyframe-snap or seek-ahead returns something other than the requested +/// index. (Today our ffmpeg path is exact-seek so they always match, but +/// the prefix makes the system robust to that ever changing.) +pub fn pack_frame(idx: u64, jpeg: &[u8]) -> Vec { + let mut out = Vec::with_capacity(8 + jpeg.len()); + out.extend_from_slice(&(idx as i64).to_le_bytes()); + out.extend_from_slice(jpeg); + out +} + /// Pick a VAAPI render node. `SAM_TOOL_VAAPI_DEVICE` overrides; otherwise we /// probe `/dev/dri/renderD128..199` for the first that exists. Multi-GPU /// systems can end up with the integrated card on D128 and a discrete card diff --git a/sam-tool-tauri/src-tauri/src/lib.rs b/sam-tool-tauri/src-tauri/src/lib.rs index 649e44d..855957d 100644 --- a/sam-tool-tauri/src-tauri/src/lib.rs +++ b/sam-tool-tauri/src-tauri/src/lib.rs @@ -6,6 +6,7 @@ mod extract; mod sam; mod settings; mod state; +mod stream; mod user; mod video; @@ -62,12 +63,15 @@ fn current_video(state: State<'_, AppState>) -> Result, String #[tauri::command] fn decode_frame(state: State<'_, AppState>, frame_idx: u64) -> Result { - let bytes = { + let jpeg = { let mut guard = state.decoder.lock().map_err(|e| format!("lock: {e}"))?; let decoder = guard.as_mut().ok_or("no video loaded")?; decoder.decode_at(frame_idx)? }; - Ok(Response::new(bytes)) + // Wire format: 8-byte LE frame number + JPEG bytes. JS unpacks the + // prefix in the worker so paints bind to the backend-authoritative + // index, not the optimistic request. + Ok(Response::new(decoder::pack_frame(frame_idx, &jpeg))) } #[tauri::command] @@ -309,6 +313,10 @@ pub fn run() { sam_health_check, sam_list_models, sam_segment_bbox, + stream::start_stream, + stream::set_stream_state, + stream::stop_stream, + stream::stream_current_frame, ]) .run(tauri::generate_context!()) .expect("error while running tauri application"); diff --git a/sam-tool-tauri/src-tauri/src/state.rs b/sam-tool-tauri/src-tauri/src/state.rs index 01c57dc..b5db1b8 100644 --- a/sam-tool-tauri/src-tauri/src/state.rs +++ b/sam-tool-tauri/src-tauri/src/state.rs @@ -1,14 +1,22 @@ use crate::decoder::DecoderSession; -use std::sync::Mutex; +use crate::stream::StreamControl; +use std::sync::{Arc, Mutex}; pub struct AppState { - pub decoder: Mutex>, + /// Wrapped in Arc so the stream producer thread can hold its own + /// handle and reach back into the decoder without going through Tauri. + pub decoder: Arc>>, + /// Atomic control struct read by the stream producer each tick. JS + /// mutates via tauri commands; the producer polls it for play / pause / + /// speed / target / stride. + pub stream_control: Arc, } impl AppState { pub fn new() -> Self { Self { - decoder: Mutex::new(None), + decoder: Arc::new(Mutex::new(None)), + stream_control: Arc::new(StreamControl::new()), } } } diff --git a/sam-tool-tauri/src-tauri/src/stream.rs b/sam-tool-tauri/src-tauri/src/stream.rs new file mode 100644 index 0000000..47d5617 --- /dev/null +++ b/sam-tool-tauri/src-tauri/src/stream.rs @@ -0,0 +1,208 @@ +//! Forward playback transport. A Rust producer thread decodes via the +//! existing DecoderSession and pushes packed frames to the JS side via a +//! Tauri Channel — eliminates the per-frame IPC round-trip that pure +//! `invoke()` playback paid for. Backward / step / scrub / pause stay on +//! the invoke-pull path (see lib.rs::decode_frame). + +use crate::decoder::{pack_frame, DecoderSession}; +use crate::state::AppState; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::{Duration, Instant}; +use tauri::ipc::{Channel, InvokeResponseBody}; +use tauri::State; + +pub struct StreamControl { + /// Bumped on each `start_stream`. Producer threads check this and exit + /// when it changes — guarantees only one producer at a time. + pub session_id: AtomicU64, + /// True while the producer should keep emitting. JS clears via + /// `set_stream_state` / `stop_stream` to pause. + pub playing: AtomicBool, + /// Speed multiplier × 100 (atomic u64; floats aren't atomic). + pub speed_x100: AtomicU64, + /// Next frame to emit. JS may overwrite this for in-stream seeks. + pub current_frame: AtomicU64, + /// Advance step (1 = every frame, N = every Nth — frame_skip). + pub stride: AtomicU64, + /// Stop when current_frame >= total_frames. + pub total_frames: AtomicU64, +} + +impl StreamControl { + pub fn new() -> Self { + Self { + session_id: AtomicU64::new(0), + playing: AtomicBool::new(false), + speed_x100: AtomicU64::new(100), + current_frame: AtomicU64::new(0), + stride: AtomicU64::new(1), + total_frames: AtomicU64::new(0), + } + } +} + +#[tauri::command] +pub fn start_stream( + state: State<'_, AppState>, + channel: Channel, + from_idx: u64, + speed: f64, + stride: u64, +) -> Result { + let (total, fps) = { + let guard = state.decoder.lock().map_err(|e| format!("lock: {e}"))?; + let info = guard.as_ref().ok_or("no video loaded")?.info(); + (info.total_frames, info.fps) + }; + if from_idx >= total { + return Err("from_idx beyond total_frames".into()); + } + + let control = state.stream_control.clone(); + // fetch_add returns the previous value; +1 = our new id. Any existing + // producer with the prior id will exit on its next iteration. + let sid = control.session_id.fetch_add(1, Ordering::SeqCst) + 1; + control + .speed_x100 + .store((speed.max(0.01) * 100.0) as u64, Ordering::SeqCst); + control.current_frame.store(from_idx, Ordering::SeqCst); + control.stride.store(stride.max(1), Ordering::SeqCst); + control.total_frames.store(total, Ordering::SeqCst); + // playing must be the LAST flag we set — once it's true the producer + // can start consuming the other fields. + control.playing.store(true, Ordering::SeqCst); + + let decoder = state.decoder.clone(); + thread::spawn(move || { + producer_loop(decoder, control, channel, sid, fps); + }); + + Ok(sid) +} + +#[tauri::command] +pub fn set_stream_state( + state: State<'_, AppState>, + playing: Option, + speed: Option, + target_frame: Option, + stride: Option, +) -> Result<(), String> { + let c = &state.stream_control; + if let Some(s) = speed { + c.speed_x100 + .store((s.max(0.01) * 100.0) as u64, Ordering::SeqCst); + } + if let Some(t) = target_frame { + c.current_frame.store(t, Ordering::SeqCst); + } + if let Some(s) = stride { + c.stride.store(s.max(1), Ordering::SeqCst); + } + if let Some(p) = playing { + c.playing.store(p, Ordering::SeqCst); + } + Ok(()) +} + +#[tauri::command] +pub fn stop_stream(state: State<'_, AppState>) -> Result<(), String> { + let c = &state.stream_control; + c.playing.store(false, Ordering::SeqCst); + // Bump session_id too so a producer that misses the playing flag still + // exits on its session check. + c.session_id.fetch_add(1, Ordering::SeqCst); + Ok(()) +} + +#[tauri::command] +pub fn stream_current_frame(state: State<'_, AppState>) -> Result { + Ok(state.stream_control.current_frame.load(Ordering::Relaxed)) +} + +fn producer_loop( + decoder: Arc>>, + control: Arc, + channel: Channel, + my_sid: u64, + fps: f64, +) { + let mut next_tick = Instant::now(); + loop { + if control.session_id.load(Ordering::Relaxed) != my_sid { + return; + } + if !control.playing.load(Ordering::Relaxed) { + return; + } + + let speed = (control.speed_x100.load(Ordering::Relaxed) as f64 / 100.0).max(0.01); + let stride = control.stride.load(Ordering::Relaxed).max(1); + let total = control.total_frames.load(Ordering::Relaxed); + let frame = control.current_frame.load(Ordering::Relaxed); + + if total == 0 || frame >= total { + control.playing.store(false, Ordering::SeqCst); + return; + } + + // Decode this frame via the existing DecoderSession (preserves + // hwaccel fallback + stderr capture + frame cache). + let jpeg = { + let mut guard = match decoder.lock() { + Ok(g) => g, + Err(_) => return, + }; + match guard.as_mut() { + Some(d) => d.decode_at(frame), + None => Err("no video loaded".into()), + } + }; + + let send_ok = match jpeg { + Ok(bytes) => { + let packed = pack_frame(frame, &bytes); + channel.send(InvokeResponseBody::Raw(packed)).is_ok() + } + Err(_) => { + // Surface decode errors as an empty packed frame so JS knows + // to stop. With just an 8-byte prefix and no JPEG it's + // unambiguous on the receive side. + let packed = pack_frame(frame, &[]); + let _ = channel.send(InvokeResponseBody::Raw(packed)); + false + } + }; + if !send_ok { + // Channel closed (frontend tore down) — stop cleanly. + control.playing.store(false, Ordering::SeqCst); + return; + } + + // Advance the cursor. JS may have already moved it (an in-stream + // seek) — in that case the next iteration picks up the new value. + let next_frame = frame.saturating_add(stride); + // Only advance if no one else moved it (compare_exchange would be + // strictly correct, but a plain store is fine: a race here just + // means we emit the seeked-to frame on the next iteration). + let _ = control.current_frame.compare_exchange( + frame, + next_frame, + Ordering::SeqCst, + Ordering::Relaxed, + ); + + // Pace: stride frames per (fps × speed) per second. + let frame_interval = Duration::from_secs_f64(stride as f64 / (fps * speed)); + next_tick += frame_interval; + let now = Instant::now(); + if next_tick > now { + thread::sleep(next_tick - now); + } else { + // Behind real-time — reset baseline instead of accumulating debt. + next_tick = now; + } + } +} diff --git a/sam-tool-tauri/src/ipc.ts b/sam-tool-tauri/src/ipc.ts index fa03c27..b21c760 100644 --- a/sam-tool-tauri/src/ipc.ts +++ b/sam-tool-tauri/src/ipc.ts @@ -1,4 +1,4 @@ -import { invoke } from "@tauri-apps/api/core"; +import { invoke, Channel } from "@tauri-apps/api/core"; import type { CocoCategory, CocoShape, @@ -70,6 +70,25 @@ export const api = { readFolderSource: (folder: string) => invoke("read_folder_source", { folder }), + // Forward playback uses a Rust producer thread that pushes packed-frame + // ArrayBuffers via this channel. Pass `new Channel()` from the + // caller and attach `.onmessage` before invoking — Rust starts sending + // immediately. + startStream: ( + channel: Channel, + fromIdx: number, + speed: number, + stride: number + ) => invoke("start_stream", { channel, fromIdx, speed, stride }), + setStreamState: (state: { + playing?: boolean; + speed?: number; + targetFrame?: number; + stride?: number; + }) => invoke("set_stream_state", state), + stopStream: () => invoke("stop_stream"), + streamCurrentFrame: () => invoke("stream_current_frame"), + getSettings: () => invoke("get_settings"), saveSettings: (settings: Settings) => invoke("save_settings", { settings }), diff --git a/sam-tool-tauri/src/modes/ExtractMode.tsx b/sam-tool-tauri/src/modes/ExtractMode.tsx index ad7c88e..ea11846 100644 --- a/sam-tool-tauri/src/modes/ExtractMode.tsx +++ b/sam-tool-tauri/src/modes/ExtractMode.tsx @@ -1,10 +1,12 @@ import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { open as openDialog } from "@tauri-apps/plugin-dialog"; import { listen } from "@tauri-apps/api/event"; +import { Channel } from "@tauri-apps/api/core"; import { api } from "../ipc"; import type { CocoShape, ImportedCoco, Settings, VideoInfo } from "../types"; import { FrameCanvas } from "../components/extract/FrameCanvas"; import { drawOverlay } from "../components/extract/overlay"; +import { sharedFrameDecoderPool } from "../services/frameDecoderPool"; const SPEEDS = [0.25, 0.5, 1, 1.5, 2, 3, 4]; const FRAME_BITMAP_CACHE_LIMIT = 24; @@ -186,34 +188,44 @@ export function ExtractMode({ username, settings }: Props) { ); // Returns true if we drew, false if we were preempted by a newer request, - // and throws if bitmap creation failed — pump uses this to decide whether - // to mark the frame rendered or surface the error to the playback loop. + // and throws if worker decode failed. The packed buffer carries an 8-byte + // LE frame-number prefix; the worker is authoritative on which frame + // actually decoded, so we use that for caching + drawing rather than the + // optimistic `atFrameIdx` (`atFrameIdx` only stays useful as a fallback + // for the preempt check). const renderFrame = useCallback( async (bytes: ArrayBuffer, atFrameIdx: number): Promise => { - // Diagnostic timing log: set `window.SAM_FRAME_DEBUG = true` in devtools - // on a slow machine to see per-frame bitmap and draw timings — confirms - // where the bottleneck lives (decode vs createImageBitmap vs drawImage). const debug = (window as unknown as { SAM_FRAME_DEBUG?: boolean }) .SAM_FRAME_DEBUG === true; const t0 = debug ? performance.now() : 0; - const blob = new Blob([bytes], { type: "image/jpeg" }); + let frameNumber: number; let bitmap: ImageBitmap; try { - bitmap = await createImageBitmap(blob); + const result = await sharedFrameDecoderPool.decode(bytes); + frameNumber = result.frameNumber; + bitmap = result.bitmap; } catch (e) { throw new Error( - `createImageBitmap failed: ${e instanceof Error ? e.message : String(e)}` + `worker decode failed: ${e instanceof Error ? e.message : String(e)}` ); } const t1 = debug ? performance.now() : 0; - rememberBitmap(atFrameIdx, bitmap); - if (requestedIdx.current !== atFrameIdx) return false; - drawBitmap(bitmap, atFrameIdx); + rememberBitmap(frameNumber, bitmap); + // Preempt check: if a newer request landed during decode, drop this + // paint. Match against either the prefix frame number (authoritative) + // or the original request — keyframe snap can make them disagree. + if ( + requestedIdx.current !== frameNumber && + requestedIdx.current !== atFrameIdx + ) { + return false; + } + drawBitmap(bitmap, frameNumber); if (debug) { const t2 = performance.now(); // eslint-disable-next-line no-console console.debug( - `[SAM] f${atFrameIdx} bytes=${bytes.byteLength} bitmap=${(t1 - t0).toFixed(0)}ms draw=${(t2 - t1).toFixed(0)}ms` + `[SAM] f${frameNumber} bytes=${bytes.byteLength} decode=${(t1 - t0).toFixed(0)}ms draw=${(t2 - t1).toFixed(0)}ms` ); } return true; @@ -329,109 +341,126 @@ export function ExtractMode({ username, settings }: Props) { return () => ro.disconnect(); }, [video, drawBitmap, getCachedBitmap, pump]); - // --- playback loop --------------------------------------------------- - // Render-paced: advance ONLY after the previous frame actually drew. On a - // fast box this hits target fps; on a slow box it plays smoothly at the - // achievable decode/render rate instead of dropping 9-of-10 frames at the - // canvas while the wall-clock UI counter races ahead. + // --- playback (Channel-push transport) ----------------------------------- + // Forward playback runs through a Rust producer thread that pushes packed + // frames over a Tauri Channel — eliminates the per-frame IPC round-trip + // that was the main stutter source. Backward / step / scrub / pause all + // stay on the invoke-pull path (pump() above). Three guards live here: + // * `paused` — set true on review-mode stop so any in-flight messages + // queued behind us don't paint; + // * `lastPainted` — direction-aware out-of-order drop in case the worker + // pool returns N+1 before N; + // * cleanup stops the stream and re-fetches the last painted frame via + // invoke so the canvas exactly matches `frameIdx` on pause. useEffect(() => { if (!playing || !video) return; let cancelled = false; - let nextDueAt = performance.now(); + let paused = false; + let lastPainted = frameIdxRef.current; + const skip = Math.max(1, frameSkipRef.current | 0); + const startFrom = Math.min( + video.total_frames - 1, + frameIdxRef.current + skip + ); + if (startFrom <= frameIdxRef.current) { + setPlaying(false); + return; + } - const waitForFrame = (target: number) => - new Promise((resolve) => { - const check = () => { - if ( - cancelled || - renderedIdx.current === target || - decodeErrorRef.current !== null - ) { - resolve(); - return; - } - requestAnimationFrame(check); - }; - check(); - }); - - const loop = async () => { - while (!cancelled && playing) { - const skip = Math.max(1, frameSkipRef.current | 0); - const current = frameIdxRef.current; - let target = Math.min(video.total_frames - 1, current + skip); - if (target <= current) { - setPlaying(false); - return; - } - // Review-mode scan: with skip > 1 a naive +skip jump can leap past - // annotations between current+1 and current+skip. Walk that window - // and clamp `target` down to the first annotated frame. - if (reviewMode) { - for (let i = current + 1; i <= target; i++) { - if (annotatedFrames.current.has(i)) { - target = i; - break; - } - } - } - requestedIdx.current = target; - // Fast path: await pump directly. When pump owns the queue it - // resolves the instant target is drawn (no 16ms rAF latency tax). - // If another caller (ResizeObserver / overlay-change effect) was - // already running pump, our call short-circuits — fall back to - // rAF polling in that case so we still wait correctly. - await pump(); - if ( - !cancelled && - renderedIdx.current !== target && - decodeErrorRef.current === null - ) { - await waitForFrame(target); - } + const channel = new Channel(); + channel.onmessage = async (buffer: ArrayBuffer) => { + if (cancelled || paused) return; + // Empty-frame sentinel from the producer = decode failed. + if (buffer.byteLength <= 8) { + setPlaybackError("decode failed during playback"); + setPlaying(false); + return; + } + let frameNumber: number; + let bitmap: ImageBitmap; + try { + const result = await sharedFrameDecoderPool.decode(buffer); + frameNumber = result.frameNumber; + bitmap = result.bitmap; + } catch (e) { if (cancelled) return; - // Decode failed mid-play: stop and surface so the UI doesn't claim to - // be playing while the canvas is frozen on the last good frame. - if (decodeErrorRef.current) { - setPlaybackError(decodeErrorRef.current); - setPlaying(false); - return; - } - const advanced = Math.max(1, target - current); - // Real-time pacing regardless of skip: skip controls density, - // `speed` controls speed. Without the advanced-frame factor here, - // skip=3 would play at 3x wall-clock, which is what the speed slider - // is for. Compute after review-mode clamping so annotated stops do not - // wait for the full skip window. - const interval = (advanced * 1000) / (video.fps * speed); - // Keep the fast playback loop's cursor in sync immediately. Waiting - // for React's frameIdx effect can make a slow decode path replay the - // same target once and add an extra frame interval per rendered frame. - frameIdxRef.current = target; - setFrameIdx(target); - // Pause AFTER rendering so the user actually sees the annotated frame. - if (reviewMode && annotatedFrames.current.has(target)) { - setPlaying(false); - return; - } + setPlaybackError(e instanceof Error ? e.message : String(e)); + setPlaying(false); + return; + } + if (cancelled || paused) { + bitmap.close(); + return; + } + // Direction-aware out-of-order drop: forward play is monotonic, so + // any frame number < the last painted came from a parallel worker + // and is stale. Cheap defence against worker-pool races. + if (frameNumber < lastPainted) { + bitmap.close(); + return; + } + lastPainted = frameNumber; + rememberBitmap(frameNumber, bitmap); + drawBitmap(bitmap, frameNumber); + renderedIdx.current = frameNumber; + requestedIdx.current = frameNumber; + frameIdxRef.current = frameNumber; + setFrameIdx(frameNumber); + decodeErrorRef.current = null; - const now = performance.now(); - const due = nextDueAt + interval; - if (due > now) { - await new Promise((r) => setTimeout(r, due - now)); - nextDueAt = due; - } else { - // Behind real-time — don't accumulate debt and then "catch up" by - // skipping frames; just reset the clock to now. - nextDueAt = now; - } + // Review-mode stops AT the annotated frame, so the canvas already + // shows it by the time we ask the producer to halt. + if (reviewMode && annotatedFrames.current.has(frameNumber)) { + paused = true; + try { await api.stopStream(); } catch { /* ignore */ } + setPlaying(false); } }; - loop(); + + api + .startStream(channel, startFrom, speed, skip) + .catch((e) => { + if (cancelled) return; + setPlaybackError(`start_stream failed: ${e}`); + setPlaying(false); + }); + return () => { cancelled = true; + // Fire-and-forget: stop the producer, then re-fetch the last painted + // frame via the authoritative invoke path so the canvas matches + // `frameIdx` exactly (the producer may have been mid-flight). + const lastFrame = frameIdxRef.current; + api + .stopStream() + .catch(() => { /* ignore */ }) + .then(() => api.decodeFrame(lastFrame)) + .then((buf) => sharedFrameDecoderPool.decode(buf)) + .then(({ frameNumber, bitmap }) => { + // Newer request may have started by the time this resolves — + // only paint if it still matches. + if (requestedIdx.current !== frameNumber) { + bitmap.close(); + return; + } + rememberBitmap(frameNumber, bitmap); + drawBitmap(bitmap, frameNumber); + renderedIdx.current = frameNumber; + }) + .catch(() => { /* invoke path will recover on next seek */ }); }; - }, [playing, video, speed, reviewMode, pump]); + }, [playing, video, speed, reviewMode, drawBitmap, rememberBitmap]); + + // Live-update the producer when speed or frame-skip changes mid-play, so + // the user doesn't have to pause + resume to see the new setting. + useEffect(() => { + if (!playing) return; + api.setStreamState({ speed }).catch(() => { /* ignore */ }); + }, [playing, speed]); + useEffect(() => { + if (!playing) return; + api.setStreamState({ stride: Math.max(1, settings?.frame_skip ?? 1) }).catch(() => { /* ignore */ }); + }, [playing, settings?.frame_skip]); // --- navigation helpers ---------------------------------------------- const seekToFrame = useCallback((idx: number) => { @@ -443,11 +472,19 @@ export function ExtractMode({ username, settings }: Props) { if (cached) { drawBitmap(cached, clamped); renderedIdx.current = clamped; - } else { + } else if (!playing) { + // Pump only handles seeks while paused. During Channel-push play the + // producer is the source of frames; we re-target it instead. pump(); } + if (playing) { + // Hand the new target to the running producer so it resumes from the + // scrub destination instead of where it was. The producer's monotonic + // session_id keeps the existing channel valid. + api.setStreamState({ targetFrame: clamped }).catch(() => { /* ignore */ }); + } setFrameIdx((prev) => (prev === clamped ? prev : clamped)); - }, [drawBitmap, getCachedBitmap, pump, video]); + }, [drawBitmap, getCachedBitmap, pump, video, playing]); const stepFrames = useCallback((delta: number) => { if (!video) return; diff --git a/sam-tool-tauri/src/services/frameDecoderPool.ts b/sam-tool-tauri/src/services/frameDecoderPool.ts new file mode 100644 index 0000000..68a3cd2 --- /dev/null +++ b/sam-tool-tauri/src/services/frameDecoderPool.ts @@ -0,0 +1,118 @@ +// Worker pool that takes a packed-frame ArrayBuffer (8-byte LE i64 frame +// number + JPEG bytes), decodes the JPEG to an ImageBitmap off the main +// thread, and returns both via transferable postMessage so the main thread +// only does the final drawImage. Inlined as a Blob URL so we don't need a +// separate build entry for the worker file. + +const WORKER_SOURCE = ` +self.onmessage = async (e) => { + const { id, buffer } = e.data; + let frameNumber = -1; + try { + const headerView = new BigInt64Array(buffer, 0, 1); + frameNumber = Number(headerView[0]); + const jpegView = new Uint8Array(buffer, 8); + // Copy out of the transferred buffer so Blob ownership is independent. + const jpegCopy = jpegView.slice(); + const blob = new Blob([jpegCopy], { type: "image/jpeg" }); + const bitmap = await createImageBitmap(blob); + self.postMessage({ id, frameNumber, bitmap }, [bitmap]); + } catch (err) { + const msg = err && err.message ? err.message : String(err); + self.postMessage({ id, frameNumber, error: msg }); + } +}; +`; + +const workerBlob = new Blob([WORKER_SOURCE], { + type: "application/javascript", +}); +const workerUrl = URL.createObjectURL(workerBlob); + +export interface DecodeResult { + frameNumber: number; + bitmap: ImageBitmap; +} + +interface Pending { + resolve: (r: DecodeResult) => void; + reject: (e: Error) => void; +} + +interface WorkerMessage { + id: number; + frameNumber: number; + bitmap?: ImageBitmap; + error?: string; +} + +export class FrameDecoderPool { + private workers: Worker[] = []; + private next = 0; + private pending = new Map(); + private nextId = 1; + + constructor(size: number = 2) { + const count = Math.max(1, Math.min(size, 8)); + for (let i = 0; i < count; i++) { + const worker = new Worker(workerUrl); + worker.onmessage = (e: MessageEvent) => { + const { id, frameNumber, bitmap, error } = e.data; + const slot = this.pending.get(id); + if (!slot) { + // The caller discarded this request — close the bitmap so it + // doesn't leak (an ImageBitmap holds GPU memory). + bitmap?.close?.(); + return; + } + this.pending.delete(id); + if (error || !bitmap) { + slot.reject(new Error(error ?? "decode failed")); + } else { + slot.resolve({ frameNumber, bitmap }); + } + }; + worker.onerror = (e) => { + // Surface worker-thread errors to any pending request on this + // worker. We don't know which id failed, so reject the oldest. + const first = this.pending.entries().next().value; + if (first) { + const [id, slot] = first; + this.pending.delete(id); + slot.reject(new Error(`worker error: ${e.message}`)); + } + }; + this.workers.push(worker); + } + } + + /** + * Decode a packed-frame buffer. The buffer ownership transfers to the + * worker — do not read it after this call returns. + */ + decode(buffer: ArrayBuffer): Promise { + return new Promise((resolve, reject) => { + const id = this.nextId++; + this.pending.set(id, { resolve, reject }); + const worker = this.workers[this.next]; + this.next = (this.next + 1) % this.workers.length; + worker.postMessage({ id, buffer }, [buffer]); + }); + } + + destroy() { + for (const w of this.workers) w.terminate(); + this.workers = []; + for (const slot of this.pending.values()) { + slot.reject(new Error("decoder pool destroyed")); + } + this.pending.clear(); + } +} + +/** + * Shared pool used by ExtractMode (and later, AnnotateMode). Two workers is + * enough for sequential playback decode; raising it helps the seek path + * when several scrub requests overlap. + */ +export const sharedFrameDecoderPool = new FrameDecoderPool(2);