This commit is contained in:
2026-06-04 18:55:10 +05:30
parent f7029e8748
commit 763e821f30
5 changed files with 504 additions and 188 deletions

View File

@@ -36,21 +36,21 @@ impl FrameCache {
self.map.contains_key(&idx) self.map.contains_key(&idx)
} }
fn get(&mut self, idx: u64) -> Option<Vec<u8>> { /// FIFO retrieval (no reorder-on-access). Our real access pattern is
let bytes = self.map.get(&idx)?.clone(); /// sequential decode, so cache fills in order and evicts in order; the
if let Some(pos) = self.order.iter().position(|&i| i == idx) { /// previous LRU reshuffle was an O(n) VecDeque scan that bought nothing.
self.order.remove(pos); fn get(&self, idx: u64) -> Option<Vec<u8>> {
} self.map.get(&idx).cloned()
self.order.push_back(idx);
Some(bytes)
} }
fn insert(&mut self, idx: u64, bytes: Vec<u8>) { fn insert(&mut self, idx: u64, bytes: Vec<u8>) {
if self.map.contains_key(&idx) { if self.map.contains_key(&idx) {
if let Some(pos) = self.order.iter().position(|&i| i == idx) { // Re-decode of an already-cached frame — just overwrite, no
self.order.remove(pos); // order change (the queue entry stays valid).
self.map.insert(idx, bytes);
return;
} }
} else if self.map.len() >= self.capacity { if self.map.len() >= self.capacity {
if let Some(oldest) = self.order.pop_front() { if let Some(oldest) = self.order.pop_front() {
self.map.remove(&oldest); self.map.remove(&oldest);
} }
@@ -100,16 +100,26 @@ pub struct DecoderSession {
/// on any decode failure so the next attempt uses software — we'd /// on any decode failure so the next attempt uses software — we'd
/// rather degrade than keep retrying a flaky GPU path. /// rather degrade than keep retrying a flaky GPU path.
hwaccel: Option<String>, hwaccel: Option<String>,
/// When Some(H), ffmpeg downscales to height H (aspect preserved) via
/// `-vf scale=-2:H`. Used by the playback decoder for cheap previews;
/// the extract decoder always passes None so saved frames are full-res.
preview_max_height: Option<u32>,
shared: Arc<Shared>, shared: Arc<Shared>,
worker: Option<JoinHandle<()>>, worker: Option<JoinHandle<()>>,
} }
impl DecoderSession { impl DecoderSession {
pub fn open(path: PathBuf, info: VideoInfo, hwaccel: Option<String>) -> Self { pub fn open(
path: PathBuf,
info: VideoInfo,
hwaccel: Option<String>,
preview_max_height: Option<u32>,
) -> Self {
Self { Self {
path, path,
info, info,
hwaccel: hwaccel.filter(|s| !s.is_empty()), hwaccel: hwaccel.filter(|s| !s.is_empty()),
preview_max_height: preview_max_height.filter(|h| *h > 0),
shared: Arc::new(Shared { shared: Arc::new(Shared {
state: Mutex::new(SharedState { state: Mutex::new(SharedState {
child: None, child: None,
@@ -223,8 +233,13 @@ impl DecoderSession {
fn seek_and_start_worker(&mut self, idx: u64) -> Result<(), String> { fn seek_and_start_worker(&mut self, idx: u64) -> Result<(), String> {
self.stop_worker(); self.stop_worker();
let (child, stdout) = let (child, stdout) = spawn_ffmpeg(
spawn_ffmpeg(&self.path, &self.info, idx, self.hwaccel.as_deref())?; &self.path,
&self.info,
idx,
self.hwaccel.as_deref(),
self.preview_max_height,
)?;
{ {
let mut st = self.shared.state.lock().unwrap(); let mut st = self.shared.state.lock().unwrap();
@@ -314,6 +329,7 @@ fn spawn_ffmpeg(
info: &VideoInfo, info: &VideoInfo,
idx: u64, idx: u64,
hwaccel: Option<&str>, hwaccel: Option<&str>,
preview_max_height: Option<u32>,
) -> Result<(Child, BufReader<ChildStdout>), String> { ) -> Result<(Child, BufReader<ChildStdout>), String> {
let path_str = path.to_str().ok_or_else(|| "non-utf8 path".to_string())?; let path_str = path.to_str().ok_or_else(|| "non-utf8 path".to_string())?;
let t = idx as f64 / info.fps; let t = idx as f64 / info.fps;
@@ -345,8 +361,17 @@ fn spawn_ffmpeg(
// milliseconds instead of hundreds of ms. // milliseconds instead of hundreds of ms.
cmd.args(["-ss", &format!("{:.6}", t)]); cmd.args(["-ss", &format!("{:.6}", t)]);
cmd.args(["-i", path_str]); cmd.args(["-i", path_str]);
cmd.args(["-an"]);
// Preview scaling: cap output height to `h`, width auto (`-2` keeps
// aspect & ensures even). Skip the filter entirely when the source is
// already shorter than the cap — no upscale, no wasted filter step.
if let Some(h) = preview_max_height {
if info.height > h {
let vf = format!("scale=-2:{h}");
cmd.args(["-vf", &vf]);
}
}
cmd.args([ cmd.args([
"-an",
"-q:v", "-q:v",
"3", "3",
"-f", "-f",
@@ -413,34 +438,329 @@ fn worker_loop(mut stdout: BufReader<ChildStdout>, shared: Arc<Shared>) {
} }
} }
/// Read one JPEG (SOI 0xFF 0xD8 .. EOI 0xFF 0xD9) from an image2pipe stream.
///
/// Chunked scan: for each `fill_buf` chunk we vector-scan for the marker
/// byte (`slice::iter().position` compiles to a SIMD memchr in release),
/// then `extend_from_slice` the body wholesale. The previous per-byte
/// `for &b in buf { push(b); ... }` cost ~3 ns/byte and at ~150 KB/frame ×
/// 30 fps it was a measurable chunk of the worker's budget. The bulk-copy
/// path measures around 10-20× faster on warm caches.
fn read_jpeg_bytes(reader: &mut BufReader<ChildStdout>) -> Result<Vec<u8>, String> { fn read_jpeg_bytes(reader: &mut BufReader<ChildStdout>) -> Result<Vec<u8>, String> {
let mut out = Vec::with_capacity(64 * 1024); let mut out: Vec<u8> = Vec::with_capacity(64 * 1024);
let mut last = 0u8;
let mut in_jpeg = false; let mut in_jpeg = false;
// Last byte of the previous chunk; needed because the 2-byte JPEG
// marker (0xFF 0xMM) can straddle a fill_buf boundary.
let mut last: u8 = 0;
loop { loop {
let buf = reader.fill_buf().map_err(|e| format!("pipe read: {e}"))?; let buf = reader.fill_buf().map_err(|e| format!("pipe read: {e}"))?;
if buf.is_empty() { if buf.is_empty() {
return Err("ffmpeg pipe closed before frame completed".into()); return Err("ffmpeg pipe closed before frame completed".into());
} }
let mut consumed = 0usize; let n = buf.len();
for &b in buf {
consumed += 1;
if !in_jpeg { if !in_jpeg {
if last == 0xFF && b == 0xD8 { match find_marker(buf, last, 0xD8) {
Some(d8_pos) => {
in_jpeg = true; in_jpeg = true;
out.push(0xFF); out.push(0xFF);
out.push(0xD8); out.push(0xD8);
} let after = d8_pos + 1;
last = b; let rest = &buf[after..];
} else { // Same-chunk EOI is the common case for small frames.
out.push(b); if let Some(d9_pos) = find_marker(rest, 0, 0xD9) {
if last == 0xFF && b == 0xD9 { out.extend_from_slice(&rest[..=d9_pos]);
reader.consume(consumed); reader.consume(after + d9_pos + 1);
return Ok(out); return Ok(out);
} }
last = b; out.extend_from_slice(rest);
last = rest.last().copied().unwrap_or(0);
reader.consume(n);
}
None => {
last = buf[n - 1];
reader.consume(n);
} }
} }
reader.consume(consumed); continue;
}
if let Some(d9_pos) = find_marker(buf, last, 0xD9) {
out.extend_from_slice(&buf[..=d9_pos]);
reader.consume(d9_pos + 1);
return Ok(out);
}
out.extend_from_slice(buf);
last = buf[n - 1];
reader.consume(n);
}
}
/// Find the index of `marker` byte preceded by 0xFF within `buf`.
/// `prev` is the last byte of the previous chunk (0 if none) so a marker
/// that straddles a fill_buf boundary is still detected.
fn find_marker(buf: &[u8], prev: u8, marker: u8) -> Option<usize> {
if buf.is_empty() {
return None;
}
if prev == 0xFF && buf[0] == marker {
return Some(0);
}
let mut start = 0usize;
while start + 1 < buf.len() {
// SIMD-friendly scan for 0xFF; LLVM lowers this to a memchr loop.
let off = buf[start..buf.len() - 1]
.iter()
.position(|&b| b == 0xFF)?;
let ff_pos = start + off;
if buf[ff_pos + 1] == marker {
return Some(ff_pos + 1);
}
start = ff_pos + 1;
}
None
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::{BufRead, Cursor, Read, Result as IoResult};
#[test]
fn find_marker_basic() {
// 0xFF 0xD8 at index 2-3
let buf = &[0x00, 0x11, 0xFF, 0xD8, 0x22][..];
assert_eq!(find_marker(buf, 0, 0xD8), Some(3));
}
#[test]
fn find_marker_no_match() {
let buf = &[0xFF, 0xC0, 0xFF, 0xC4, 0x00][..];
assert_eq!(find_marker(buf, 0, 0xD8), None);
}
#[test]
fn find_marker_boundary() {
// prev chunk ended on 0xFF, this chunk starts with 0xD9
let buf = &[0xD9, 0x00, 0x00][..];
assert_eq!(find_marker(buf, 0xFF, 0xD9), Some(0));
}
#[test]
fn find_marker_trailing_ff() {
// 0xFF at the last byte — has no successor in this chunk, so no
// match here; the caller will carry it as `last` to the next chunk.
let buf = &[0x00, 0x11, 0xFF][..];
assert_eq!(find_marker(buf, 0, 0xD9), None);
}
#[test]
fn find_marker_skips_false_positive() {
// 0xFF followed by 0xC0 (not our marker), then real 0xFF 0xD9 later.
let buf = &[0xFF, 0xC0, 0x11, 0xFF, 0xD9, 0x22][..];
assert_eq!(find_marker(buf, 0, 0xD9), Some(4));
}
// ChunkedReader: a BufRead that returns the source one configurable
// chunk at a time, so we can exercise the boundary logic in
// read_jpeg_bytes.
struct ChunkedReader {
data: Vec<u8>,
pos: usize,
chunk: usize,
held: Vec<u8>,
}
impl ChunkedReader {
fn new(data: Vec<u8>, chunk: usize) -> Self {
Self {
data,
pos: 0,
chunk,
held: Vec::new(),
}
}
}
impl Read for ChunkedReader {
fn read(&mut self, _: &mut [u8]) -> IoResult<usize> {
unreachable!("BufRead path only")
}
}
impl BufRead for ChunkedReader {
fn fill_buf(&mut self) -> IoResult<&[u8]> {
if self.held.is_empty() && self.pos < self.data.len() {
let end = (self.pos + self.chunk).min(self.data.len());
self.held = self.data[self.pos..end].to_vec();
self.pos = end;
}
Ok(&self.held)
}
fn consume(&mut self, n: usize) {
self.held.drain(..n);
}
}
// Test-only sibling of read_jpeg_bytes that takes any BufRead. Keeps
// the production signature constrained to BufReader<ChildStdout>.
fn read_jpeg_generic<R: BufRead>(reader: &mut R) -> Result<Vec<u8>, String> {
let mut out: Vec<u8> = Vec::with_capacity(64 * 1024);
let mut in_jpeg = false;
let mut last: u8 = 0;
loop {
let buf = reader.fill_buf().map_err(|e| format!("pipe read: {e}"))?;
if buf.is_empty() {
return Err("ffmpeg pipe closed before frame completed".into());
}
let n = buf.len();
if !in_jpeg {
match find_marker(buf, last, 0xD8) {
Some(d8_pos) => {
in_jpeg = true;
out.push(0xFF);
out.push(0xD8);
let after = d8_pos + 1;
let rest_len = n - after;
// borrow ends after this if-block; copy data out first
let (rest_copy, rest_last) = {
let rest = &buf[after..];
if let Some(d9_pos) = find_marker(rest, 0, 0xD9) {
out.extend_from_slice(&rest[..=d9_pos]);
let total = after + d9_pos + 1;
reader.consume(total);
return Ok(out);
}
(rest.to_vec(), rest.last().copied().unwrap_or(0))
};
out.extend_from_slice(&rest_copy);
last = rest_last;
let _ = rest_len;
reader.consume(n);
}
None => {
last = buf[n - 1];
reader.consume(n);
}
}
continue;
}
if let Some(d9_pos) = find_marker(buf, last, 0xD9) {
let take = d9_pos + 1;
let body: Vec<u8> = buf[..take].to_vec();
out.extend_from_slice(&body);
reader.consume(take);
return Ok(out);
}
let last_byte = buf[n - 1];
let body: Vec<u8> = buf.to_vec();
out.extend_from_slice(&body);
last = last_byte;
reader.consume(n);
}
}
fn make_jpeg(prefix: &[u8], body: &[u8], suffix: &[u8]) -> Vec<u8> {
let mut v = Vec::new();
v.extend_from_slice(prefix);
v.extend_from_slice(&[0xFF, 0xD8]);
v.extend_from_slice(body);
v.extend_from_slice(&[0xFF, 0xD9]);
v.extend_from_slice(suffix);
v
}
fn expected_jpeg(body: &[u8]) -> Vec<u8> {
let mut v = vec![0xFF, 0xD8];
v.extend_from_slice(body);
v.extend_from_slice(&[0xFF, 0xD9]);
v
}
#[test]
fn read_jpeg_single_chunk() {
let body = vec![0x11, 0x22, 0x33, 0x44];
let stream = make_jpeg(&[0xAA], &body, &[0xBB]);
let mut r = ChunkedReader::new(stream, 4096);
let got = read_jpeg_generic(&mut r).unwrap();
assert_eq!(got, expected_jpeg(&body));
}
#[test]
fn read_jpeg_byte_at_a_time() {
let body = vec![0x10, 0x20, 0xFF, 0x00, 0x30, 0x40];
let stream = make_jpeg(&[0x99, 0xFF, 0xC0], &body, &[]);
let mut r = ChunkedReader::new(stream, 1);
let got = read_jpeg_generic(&mut r).unwrap();
assert_eq!(got, expected_jpeg(&body));
}
#[test]
fn read_jpeg_soi_straddles_chunks() {
// Force a chunk boundary that ends with 0xFF and the next starts
// with 0xD8 (the SOI's second byte).
let body = vec![0xAA, 0xBB];
let mut stream = vec![0xFF]; // chunk1 = [0xFF]
stream.extend_from_slice(&[0xD8]); // chunk2 begins with 0xD8
stream.extend_from_slice(&body);
stream.extend_from_slice(&[0xFF, 0xD9]);
let mut r = ChunkedReader::new(stream, 1);
let got = read_jpeg_generic(&mut r).unwrap();
assert_eq!(got, expected_jpeg(&body));
}
#[test]
fn read_jpeg_eoi_straddles_chunks() {
// chunk1 ends on the 0xFF of EOI, chunk2 starts with 0xD9.
let body = vec![0x10, 0x20, 0x30];
let mut stream = vec![0xFF, 0xD8];
stream.extend_from_slice(&body);
stream.push(0xFF); // chunk boundary right after this
stream.push(0xD9);
let mut r = ChunkedReader::new(stream, 6); // body+SOI = 5, then 0xFF lands at idx 5 → fits in first chunk; need a chunk size that splits between 0xFF and 0xD9. SOI(2)+body(3)+0xFF = 6 bytes; chunk size 6 → first chunk includes through 0xFF. Then chunk2 = [0xD9].
let got = read_jpeg_generic(&mut r).unwrap();
assert_eq!(got, expected_jpeg(&body));
}
#[test]
fn read_jpeg_false_eoi_in_body() {
// 0xFF 0xC0 (a JFIF marker) inside body should not be mistaken for EOI.
let body = vec![0x11, 0xFF, 0xC0, 0x22, 0xFF, 0xDB, 0x33];
let stream = make_jpeg(&[], &body, &[]);
let mut r = ChunkedReader::new(stream, 3);
let got = read_jpeg_generic(&mut r).unwrap();
assert_eq!(got, expected_jpeg(&body));
}
#[test]
fn fifo_cache_evicts_oldest_in_order() {
let mut c = FrameCache::new(3);
c.insert(1, vec![1]);
c.insert(2, vec![2]);
c.insert(3, vec![3]);
c.insert(4, vec![4]); // evicts 1
assert!(!c.contains(1));
assert_eq!(c.get(2), Some(vec![2]));
assert_eq!(c.get(3), Some(vec![3]));
assert_eq!(c.get(4), Some(vec![4]));
}
#[test]
fn fifo_cache_overwrite_keeps_order() {
let mut c = FrameCache::new(2);
c.insert(1, vec![1]);
c.insert(2, vec![2]);
c.insert(1, vec![10]); // overwrite, do NOT reorder
c.insert(3, vec![3]); // evicts 1 (still the oldest in queue)
assert!(!c.contains(1));
assert_eq!(c.get(2), Some(vec![2]));
assert_eq!(c.get(3), Some(vec![3]));
}
#[test]
fn cache_used() {
// Suppress unused-import warning under cfg(test).
let _ = Cursor::new(vec![1u8]);
} }
} }

View File

@@ -39,32 +39,58 @@ fn open_video(
.as_ref() .as_ref()
.map(|s| s.hwaccel.clone()) .map(|s| s.hwaccel.clone())
.filter(|s| !s.is_empty()); .filter(|s| !s.is_empty());
let decoder = DecoderSession::open(p, info.clone(), hwaccel); let preview_max_height = settings.as_ref().map(|s| s.preview_max_height);
// Two sessions:
// - preview: may downscale via -vf scale=-2:H for cheap playback.
// - extract: always full resolution; what extract_one writes to disk.
// Both spawn ffmpeg lazily — opening a video is still cheap.
let preview = DecoderSession::open(p.clone(), info.clone(), hwaccel.clone(), preview_max_height);
let extract = DecoderSession::open(p, info.clone(), hwaccel, None);
if let Some(s) = settings.as_ref() { if let Some(s) = settings.as_ref() {
decoder.set_cache_capacity(s.cache_capacity); preview.set_cache_capacity(s.cache_capacity);
extract.set_cache_capacity(s.cache_capacity);
} }
let mut guard = state.decoder.lock().map_err(|e| format!("lock: {e}"))?; *state
*guard = Some(decoder); .preview_decoder
.lock()
.map_err(|e| format!("lock: {e}"))? = Some(preview);
*state
.extract_decoder
.lock()
.map_err(|e| format!("lock: {e}"))? = Some(extract);
Ok(info) Ok(info)
} }
#[tauri::command] #[tauri::command]
fn close_video(state: State<'_, AppState>) -> Result<(), String> { fn close_video(state: State<'_, AppState>) -> Result<(), String> {
let mut guard = state.decoder.lock().map_err(|e| format!("lock: {e}"))?; *state
*guard = None; .preview_decoder
.lock()
.map_err(|e| format!("lock: {e}"))? = None;
*state
.extract_decoder
.lock()
.map_err(|e| format!("lock: {e}"))? = None;
Ok(()) Ok(())
} }
#[tauri::command] #[tauri::command]
fn current_video(state: State<'_, AppState>) -> Result<Option<VideoInfo>, String> { fn current_video(state: State<'_, AppState>) -> Result<Option<VideoInfo>, String> {
let guard = state.decoder.lock().map_err(|e| format!("lock: {e}"))?; let guard = state
.preview_decoder
.lock()
.map_err(|e| format!("lock: {e}"))?;
Ok(guard.as_ref().map(|d| d.info().clone())) Ok(guard.as_ref().map(|d| d.info().clone()))
} }
#[tauri::command] #[tauri::command]
fn decode_frame(state: State<'_, AppState>, frame_idx: u64) -> Result<Response, String> { fn decode_frame(state: State<'_, AppState>, frame_idx: u64) -> Result<Response, String> {
let jpeg = { let jpeg = {
let mut guard = state.decoder.lock().map_err(|e| format!("lock: {e}"))?; let mut guard = state
.preview_decoder
.lock()
.map_err(|e| format!("lock: {e}"))?;
let decoder = guard.as_mut().ok_or("no video loaded")?; let decoder = guard.as_mut().ok_or("no video loaded")?;
decoder.decode_at(frame_idx)? decoder.decode_at(frame_idx)?
}; };
@@ -86,8 +112,20 @@ fn import_coco(path: String) -> Result<coco::ImportedCoco, String> {
#[tauri::command] #[tauri::command]
fn set_cache_capacity(state: State<'_, AppState>, capacity: usize) -> Result<usize, String> { fn set_cache_capacity(state: State<'_, AppState>, capacity: usize) -> Result<usize, String> {
let clamped = capacity.clamp(decoder::MIN_CACHE_CAPACITY, decoder::MAX_CACHE_CAPACITY); let clamped = capacity.clamp(decoder::MIN_CACHE_CAPACITY, decoder::MAX_CACHE_CAPACITY);
let guard = state.decoder.lock().map_err(|e| format!("lock: {e}"))?; if let Some(d) = state
if let Some(d) = guard.as_ref() { .preview_decoder
.lock()
.map_err(|e| format!("lock: {e}"))?
.as_ref()
{
d.set_cache_capacity(clamped);
}
if let Some(d) = state
.extract_decoder
.lock()
.map_err(|e| format!("lock: {e}"))?
.as_ref()
{
d.set_cache_capacity(clamped); d.set_cache_capacity(clamped);
} }
Ok(clamped) Ok(clamped)
@@ -101,7 +139,10 @@ fn extract_frame(
shapes: Vec<coco::Shape>, shapes: Vec<coco::Shape>,
author: String, author: String,
) -> Result<(), String> { ) -> Result<(), String> {
let mut guard = state.decoder.lock().map_err(|e| format!("lock: {e}"))?; let mut guard = state
.extract_decoder
.lock()
.map_err(|e| format!("lock: {e}"))?;
let decoder = guard.as_mut().ok_or("no video loaded")?; let decoder = guard.as_mut().ok_or("no video loaded")?;
extract::extract_one(decoder, frame_idx, Path::new(&out_dir), &shapes, &author) extract::extract_one(decoder, frame_idx, Path::new(&out_dir), &shapes, &author)
} }
@@ -132,7 +173,10 @@ fn bulk_extract(
return Err("end must be >= start".into()); return Err("end must be >= start".into());
} }
let step = frame_skip.unwrap_or(1).max(1); let step = frame_skip.unwrap_or(1).max(1);
let mut guard = state.decoder.lock().map_err(|e| format!("lock: {e}"))?; let mut guard = state
.extract_decoder
.lock()
.map_err(|e| format!("lock: {e}"))?;
let decoder = guard.as_mut().ok_or("no video loaded")?; let decoder = guard.as_mut().ok_or("no video loaded")?;
let dir = PathBuf::from(&out_dir); let dir = PathBuf::from(&out_dir);
let total = ((end - start) / step) + 1; let total = ((end - start) / step) + 1;

View File

@@ -3,9 +3,13 @@ use crate::stream::StreamControl;
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
pub struct AppState { pub struct AppState {
/// Wrapped in Arc so the stream producer thread can hold its own /// Playback decoder. May apply `-vf scale=` to shrink output for cheap
/// handle and reach back into the decoder without going through Tauri. /// previews on slow machines. Arc-wrapped so the stream producer
pub decoder: Arc<Mutex<Option<DecoderSession>>>, /// thread can keep its own handle.
pub preview_decoder: Arc<Mutex<Option<DecoderSession>>>,
/// Full-resolution decoder used by `extract_one` / bulk extract. Lazy:
/// no ffmpeg spawns until first extract, so opening a video stays cheap.
pub extract_decoder: Arc<Mutex<Option<DecoderSession>>>,
/// Atomic control struct read by the stream producer each tick. JS /// Atomic control struct read by the stream producer each tick. JS
/// mutates via tauri commands; the producer polls it for play / pause / /// mutates via tauri commands; the producer polls it for play / pause /
/// speed / target / stride. /// speed / target / stride.
@@ -15,7 +19,8 @@ pub struct AppState {
impl AppState { impl AppState {
pub fn new() -> Self { pub fn new() -> Self {
Self { Self {
decoder: Arc::new(Mutex::new(None)), preview_decoder: Arc::new(Mutex::new(None)),
extract_decoder: Arc::new(Mutex::new(None)),
stream_control: Arc::new(StreamControl::new()), stream_control: Arc::new(StreamControl::new()),
} }
} }

View File

@@ -52,7 +52,10 @@ pub fn start_stream(
stride: u64, stride: u64,
) -> Result<u64, String> { ) -> Result<u64, String> {
let (total, fps) = { let (total, fps) = {
let guard = state.decoder.lock().map_err(|e| format!("lock: {e}"))?; let guard = state
.preview_decoder
.lock()
.map_err(|e| format!("lock: {e}"))?;
let info = guard.as_ref().ok_or("no video loaded")?.info(); let info = guard.as_ref().ok_or("no video loaded")?.info();
(info.total_frames, info.fps) (info.total_frames, info.fps)
}; };
@@ -74,7 +77,7 @@ pub fn start_stream(
// can start consuming the other fields. // can start consuming the other fields.
control.playing.store(true, Ordering::SeqCst); control.playing.store(true, Ordering::SeqCst);
let decoder = state.decoder.clone(); let decoder = state.preview_decoder.clone();
thread::spawn(move || { thread::spawn(move || {
producer_loop(decoder, control, channel, sid, fps); producer_loop(decoder, control, channel, sid, fps);
}); });

View File

@@ -1,7 +1,6 @@
import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { useCallback, useEffect, useMemo, useRef, useState } from "react";
import { open as openDialog } from "@tauri-apps/plugin-dialog"; import { open as openDialog } from "@tauri-apps/plugin-dialog";
import { listen } from "@tauri-apps/api/event"; import { listen } from "@tauri-apps/api/event";
import { Channel } from "@tauri-apps/api/core";
import { api } from "../ipc"; import { api } from "../ipc";
import type { CocoShape, ImportedCoco, Settings, VideoInfo } from "../types"; import type { CocoShape, ImportedCoco, Settings, VideoInfo } from "../types";
import { FrameCanvas } from "../components/extract/FrameCanvas"; import { FrameCanvas } from "../components/extract/FrameCanvas";
@@ -9,9 +8,9 @@ import { drawOverlay } from "../components/extract/overlay";
import { sharedFrameDecoderPool } from "../services/frameDecoderPool"; import { sharedFrameDecoderPool } from "../services/frameDecoderPool";
const SPEEDS = [0.25, 0.5, 1, 1.5, 2, 3, 4]; const SPEEDS = [0.25, 0.5, 1, 1.5, 2, 3, 4];
// Stream playback no longer writes to this cache (it holds at most one // Small cache — playback writes one entry per frame and we want recent
// bitmap at a time via `lastStreamBitmapRef`). The cache is now used only by // scrub-back frames available without going back to ffmpeg, but we don't
// the seek/scrub/step path, so a small limit is plenty. // need 24+ frames worth of GPU memory tied up.
const FRAME_BITMAP_CACHE_LIMIT = 8; const FRAME_BITMAP_CACHE_LIMIT = 8;
interface Props { interface Props {
@@ -101,15 +100,6 @@ export function ExtractMode({ username, settings }: Props) {
const renderedIdx = useRef(-1); const renderedIdx = useRef(-1);
const decodingInFlight = useRef(false); const decodingInFlight = useRef(false);
const bitmapCacheRef = useRef<Map<number, ImageBitmap>>(new Map()); const bitmapCacheRef = useRef<Map<number, ImageBitmap>>(new Map());
// During stream playback we don't write to the bitmap cache (each frame is
// a one-shot, caching them at 30 fps just balloons GPU memory). The stream
// handler holds the latest bitmap here and closes the previous one when a
// new one arrives.
const lastStreamBitmapRef = useRef<ImageBitmap | null>(null);
// Bumped on every stream start/stop. The channel handler reads this before
// and after every `await` and bails (closing the bitmap) if it changed —
// prevents stale frames from a torn-down stream painting over a new one.
const streamEpochRef = useRef(0);
// Last decode/render error message. Set in `pump` on failure, cleared on // Last decode/render error message. Set in `pump` on failure, cleared on
// any successful render, and observed by the playback loop so a mid-play // any successful render, and observed by the playback loop so a mid-play
// failure stops playback with a visible banner instead of leaving the UI // failure stops playback with a visible banner instead of leaving the UI
@@ -332,7 +322,7 @@ export function ExtractMode({ username, settings }: Props) {
const repaintCurrent = () => { const repaintCurrent = () => {
const idx = frameIdxRef.current; const idx = frameIdxRef.current;
const cached = lastStreamBitmapRef.current ?? getCachedBitmap(idx); const cached = getCachedBitmap(idx);
if (cached) drawBitmap(cached, idx); if (cached) drawBitmap(cached, idx);
else { renderedIdx.current = -1; pump(); } else { renderedIdx.current = -1; pump(); }
}; };
@@ -355,122 +345,96 @@ export function ExtractMode({ username, settings }: Props) {
return () => ro.disconnect(); return () => ro.disconnect();
}, [video, drawBitmap, getCachedBitmap, pump, settings?.preview_max_height]); }, [video, drawBitmap, getCachedBitmap, pump, settings?.preview_max_height]);
// --- playback (Channel-push transport) ----------------------------------- // --- playback (invoke-pull, every-frame) ---------------------------------
// Forward playback runs through a Rust producer thread that pushes packed // SAM-Tool is an EXTRACTION tool: the user is reviewing frames to pick
// frames over a Tauri Channel. The hot path here is RIGOROUSLY ref-only: // which to extract, so we MUST display every frame in order — dropping
// * `frameIdxRef.current` is the authoritative current frame. // any frame defeats the use case. That rules out Channel-push (which
// * `setFrameIdx` is NEVER called from the channel handler — that was // emits at wall-clock rate and drops frames when the consumer lags).
// forcing a 30 Hz React reconciliation of the 1000+ line ExtractMode //
// tree, which starved the decode pump. A separate 10 Hz interval // Architecture instead: classic pull loop. The loop sets requestedIdx to
// copies the ref into state for the slider/readout. // the next frame and awaits `pump()`. pump returns the instant the frame
// * Bitmaps from the stream are NOT written into the LRU cache (24 // is on the canvas (single-flight, no rAF polling). We then schedule the
// bitmaps × 1080p ≈ 200 MB of GPU memory if we did). The channel // next iteration with setTimeout to hit the target wall-clock interval.
// handler holds at most one bitmap at a time in `lastStreamBitmapRef`. // If decode/render exceeds the interval, the loop simply runs slower —
// * Deps are `[playing, video?.path]` only — speed/reviewMode/skip flow // every frame still painted. Effective playback rate is
// through refs so toggling them mid-play does NOT tear down the // min(fps × speed, decode_capacity). The user can lower the
// producer. A separate effect pushes them via setStreamState. // preview-quality preset to raise decode_capacity on slow machines.
// * `streamEpochRef` is bumped on every start so any in-flight decode //
// from a torn-down session can be bailed (and its bitmap closed). // The React-reconciliation fix is preserved: setFrameIdx is NOT called
// per frame. frameIdxRef.current is the authoritative cursor, and the
// 10 Hz interval below syncs it to state for the slider/readout. speed,
// reviewMode, and frame_skip are read via refs so toggling them mid-play
// doesn't tear down the loop.
useEffect(() => { useEffect(() => {
if (!playing || !video) return; if (!playing || !video) return;
let cancelled = false; let cancelled = false;
let paused = false; let nextDueAt = performance.now();
let lastPainted = frameIdxRef.current;
const myEpoch = ++streamEpochRef.current; const loop = async () => {
while (!cancelled && playing) {
const skip = Math.max(1, frameSkipRef.current | 0); const skip = Math.max(1, frameSkipRef.current | 0);
const startFrom = Math.min( const current = frameIdxRef.current;
video.total_frames - 1, let target = Math.min(video.total_frames - 1, current + skip);
frameIdxRef.current + skip if (target <= current) {
); setPlaying(false);
if (startFrom <= frameIdxRef.current) { return;
}
// Review-mode scan: with skip > 1 a naive +skip jump can leap past
// annotated frames between current+1 and current+skip. Walk that
// window and clamp `target` down to the first annotated frame.
if (reviewModeRef.current) {
for (let i = current + 1; i <= target; i++) {
if (annotatedFrames.current.has(i)) {
target = i;
break;
}
}
}
requestedIdx.current = target;
await pump();
if (cancelled) return;
if (decodeErrorRef.current) {
setPlaybackError(decodeErrorRef.current);
setPlaying(false);
return;
}
// Cursor is ref-only here — the 10 Hz syncer below batches updates
// to React state, so we don't pay the reconciliation cost per frame.
frameIdxRef.current = target;
if (reviewModeRef.current && annotatedFrames.current.has(target)) {
// Annotated stop — flush state synchronously so the slider lands
// exactly where the user expects.
setFrameIdx(target);
setPlaying(false); setPlaying(false);
return; return;
} }
const channel = new Channel<ArrayBuffer>(); const advanced = Math.max(1, target - current);
channel.onmessage = async (buffer: ArrayBuffer) => { const interval = (advanced * 1000) / (video.fps * speedRef.current);
if (cancelled || paused || streamEpochRef.current !== myEpoch) return; const now = performance.now();
if (buffer.byteLength <= 8) { const due = nextDueAt + interval;
setPlaybackError("decode failed during playback"); if (due > now) {
setPlaying(false); await new Promise((r) => setTimeout(r, due - now));
return; nextDueAt = due;
} else {
// Decode took longer than the target interval; reset baseline
// instead of accumulating debt. Playback runs slower but no
// frames are dropped.
nextDueAt = now;
} }
let frameNumber: number;
let bitmap: ImageBitmap;
try {
const result = await sharedFrameDecoderPool.decode(buffer);
frameNumber = result.frameNumber;
bitmap = result.bitmap;
} catch (e) {
if (cancelled || streamEpochRef.current !== myEpoch) return;
setPlaybackError(e instanceof Error ? e.message : String(e));
setPlaying(false);
return;
}
// Re-check epoch and cancel after the await: a play→pause→play in
// quick succession can land here with a bitmap from the prior session.
if (
cancelled ||
paused ||
streamEpochRef.current !== myEpoch ||
frameNumber < lastPainted
) {
bitmap.close();
return;
}
lastPainted = frameNumber;
// Hold exactly one bitmap at a time during stream playback — do NOT
// populate the LRU cache from here.
const prev = lastStreamBitmapRef.current;
lastStreamBitmapRef.current = bitmap;
if (prev && prev !== bitmap) prev.close();
drawBitmap(bitmap, frameNumber);
// Ref-only writes. The 10 Hz interval below syncs to React state.
renderedIdx.current = frameNumber;
requestedIdx.current = frameNumber;
frameIdxRef.current = frameNumber;
decodeErrorRef.current = null;
if (reviewModeRef.current && annotatedFrames.current.has(frameNumber)) {
paused = true;
try { await api.stopStream(); } catch { /* ignore */ }
// This IS a transition the user must see immediately, so flush
// state synchronously here.
setFrameIdx(frameNumber);
setPlaying(false);
} }
}; };
loop();
api
.startStream(channel, startFrom, speedRef.current, skip)
.catch((e) => {
if (cancelled) return;
setPlaybackError(`start_stream failed: ${e}`);
setPlaying(false);
});
return () => { return () => {
cancelled = true; cancelled = true;
streamEpochRef.current++; // Sync the ref into state so the slider/readout aren't stale after
const lastFrame = frameIdxRef.current; // pause.
// Flush the ref into state so the slider/readout aren't stale after setFrameIdx(frameIdxRef.current);
// pause. (The 10 Hz syncer below would catch up but this avoids the
// visible lag.)
setFrameIdx(lastFrame);
// Fire-and-forget stop. We don't do a re-fetch round-trip on every
// cleanup any more — the next user-initiated seek or step pulls a
// fresh frame via the pump (invoke) path. Cuts an IPC + worker
// decode off every pause.
api.stopStream().catch(() => { /* ignore */ });
// Close the held stream bitmap so its GPU memory frees promptly.
const held = lastStreamBitmapRef.current;
if (held) {
held.close();
lastStreamBitmapRef.current = null;
}
}; };
// Stable deps: speed / reviewMode / frameSkip are read via refs and // Stable deps: speed / reviewMode / frame_skip are read via refs so
// applied to the running producer through setStreamState below. // toggling them mid-play doesn't tear down the loop.
// eslint-disable-next-line react-hooks/exhaustive-deps // eslint-disable-next-line react-hooks/exhaustive-deps
}, [playing, video?.path]); }, [playing, video?.path]);
@@ -487,18 +451,6 @@ export function ExtractMode({ username, settings }: Props) {
return () => window.clearInterval(id); return () => window.clearInterval(id);
}, [playing]); }, [playing]);
// Live-update the producer when speed or frame-skip changes mid-play —
// these go through setStreamState atomics so the producer thread doesn't
// need a restart.
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 ---------------------------------------------- // --- navigation helpers ----------------------------------------------
const seekToFrame = useCallback((idx: number) => { const seekToFrame = useCallback((idx: number) => {
if (!video) return; if (!video) return;
@@ -509,17 +461,9 @@ export function ExtractMode({ username, settings }: Props) {
if (cached) { if (cached) {
drawBitmap(cached, clamped); drawBitmap(cached, clamped);
renderedIdx.current = clamped; renderedIdx.current = clamped;
} else if (!playing) { } else {
// Pump only handles seeks while paused. During Channel-push play the
// producer is the source of frames; we re-target it instead.
pump(); 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)); setFrameIdx((prev) => (prev === clamped ? prev : clamped));
}, [drawBitmap, getCachedBitmap, pump, video, playing]); }, [drawBitmap, getCachedBitmap, pump, video, playing]);