From 4da559e39f3ed7e296c4bff8c4a0f6e3a5442ef3 Mon Sep 17 00:00:00 2001 From: MattJackson <1085847+MattJackson@users.noreply.github.com> Date: Tue, 19 May 2026 14:19:37 -0700 Subject: [PATCH] mux: PrefetchedSectorSource event_fn + delete DiscStream::new_pipeline * `PrefetchedSectorSource::new_with_events` adds an optional `event_fn` callback that fires `BytesRead` after every successful batch from the producer thread. The original `new()` becomes a thin no-events wrapper. Lets autorip wire the highway and still get UI progress events without polling the consumer side. * `build_iso_pipeline` grows an `event_fn` arg so the autorip multipass mux can pipe BytesRead straight through to its progress UI. * Stream trait gains a default `errors() -> u64` method (= 0) so Box callers (autorip's mux loop) can read the skip-on-error counter without downcasting. `DiscStream` overrides to return its `errors` field. * Delete `DiscStream::new_pipeline` and the pipeline-mode fields (`demux_thread`, `demux_rx`) plus the `read_pipeline` helper. All pipeline construction now goes through `PipelinedPesStream` via `build_iso_pipeline`; `DiscStream` becomes the single-thread-only inline path used by the drive single-pass read. * `lib.rs` re-exports `build_iso_pipeline`. --- src/lib.rs | 1 + src/mux/disc.rs | 233 ++------------------------------------- src/mux/resolve.rs | 5 +- src/pes.rs | 10 ++ src/sector/prefetched.rs | 37 ++++++- 5 files changed, 60 insertions(+), 226 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index adc42f8..daae233 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -211,6 +211,7 @@ pub use mux::{InputOptions, StreamUrl, input, output, parse_url}; // are the ISO-on-disk implementations. [`DecryptingSectorSource`] is the // single decrypt-on-read decorator (AACS / CSS / none) — wrap any // `SectorSource` to get plaintext sectors out. +pub use mux::build_iso_pipeline; pub use scsi::{DriveInfo, ScsiSense, ScsiTransport, drive_has_disc, list_drives}; pub use sector::{ DecryptingSectorSource, FileSectorSink, FileSectorSource, PrefetchedSectorSource, SectorSink, diff --git a/src/mux/disc.rs b/src/mux/disc.rs index b45070a..18e74c2 100644 --- a/src/mux/disc.rs +++ b/src/mux/disc.rs @@ -149,28 +149,16 @@ pub struct DiscStream { // a percent without a separate API call. bytes_total_extents: u64, - // PES output. `ts_demuxer` and `ps_demuxer` are `None` when the - // stream is in pipeline mode — the demux state lives inside a - // [`super::demux_thread::DemuxThread`] and PesPackets arrive - // through `demux_rx` already parsed. + // PES output — single-threaded inline demux + codec parse. The + // pipeline-mode mux (3-stage threaded) lives in + // [`super::pipelined_stream::PipelinedPesStream`]; this type is + // the legacy in-thread path for live-disc reads where adaptive + // batch retry on bad sectors lives in `fill_extents`. ts_demuxer: Option, ps_demuxer: Option, parsers: Vec<(u16, Box)>, pending_frames: std::collections::VecDeque, pid_to_track: Vec<(u16, usize)>, - - // Pipeline mode: when `Some`, the read+decrypt+demux pipeline - // runs on a dedicated thread; this stream's `read()` just pulls - // PesPacket batches from `demux_rx` and runs codec parse on the - // caller thread. See [`super::demux_thread`]. - // - // `demux_thread` is kept solely so that `Drop` joins the worker - // before this stream is dropped — direct reads happen through - // `demux_rx`. The `allow(dead_code)` keeps the optimizer happy - // since the field is only used at drop time. - #[allow(dead_code)] - demux_thread: Option, - demux_rx: Option>, } impl DiscStream { @@ -251,118 +239,6 @@ impl DiscStream { parsers, pending_frames: std::collections::VecDeque::new(), pid_to_track, - demux_thread: None, - demux_rx: None, - } - } - - /// Pipeline-mode constructor. Moves the read+decrypt+demux work - /// onto a [`super::demux_thread::DemuxThread`] so the caller's - /// `read()` thread only does codec parse + frame emission. - /// - /// `reader` is a [`crate::sector::PrefetchedSectorSource`] — the - /// prefetched producer thread already runs read+decrypt on its - /// own thread; this constructor peels off its channels for the - /// demux thread to consume in zero-copy mode (no buffer memcpy - /// across thread boundary, recycled-pool of two buffers, no - /// allocator activity in the hot loop). - /// - /// Pipeline mode is the preferred wiring for ISO file mux on a - /// multi-core host; it gives a ~2× consumer throughput in the - /// `null://` benchmark vs the single-thread inline path. - pub fn new_pipeline( - reader: crate::sector::PrefetchedSectorSource, - title: DiscTitle, - decrypt_keys: crate::decrypt::DecryptKeys, - batch_sectors: u16, - content_format: crate::disc::ContentFormat, - halt: Option, - ) -> Self { - let extents = title.extents.clone(); - let bytes_total_extents: u64 = extents.iter().map(|e| e.sector_count as u64 * 2048).sum(); - - let mut pids = Vec::new(); - let mut parsers = Vec::new(); - let mut pid_to_track = Vec::new(); - for (idx, s) in title.streams.iter().enumerate() { - let (pid, codec) = match s { - crate::disc::Stream::Video(v) => (v.pid, v.codec), - crate::disc::Stream::Audio(a) => (a.pid, a.codec), - crate::disc::Stream::Subtitle(s) => (s.pid, s.codec), - }; - pids.push(pid); - pid_to_track.push((pid, idx)); - parsers.push((pid, super::codec::parser_for_codec(codec, None))); - } - - let (ts, ps) = match content_format { - crate::disc::ContentFormat::MpegPs => (None, Some(super::ps::PsDemuxer::new())), - crate::disc::ContentFormat::BdTs => { - let ts_pids: Vec = pids.clone(); - if ts_pids.is_empty() { - (None, None) - } else { - (Some(super::ts::TsDemuxer::new(&ts_pids)), None) - } - } - }; - - let (prefetch_rx, recycle_tx, shell) = reader.into_channels(); - let (handle, rx) = super::demux_thread::DemuxThread::spawn_zero_copy( - prefetch_rx, - recycle_tx, - shell, - halt.clone(), - ts, - ps, - ); - - // The DiscStream's own reader is a no-op pass-through — the - // real reader lives inside the demux thread. We need *some* - // Box to satisfy the field type; use a - // tiny stub. fill_extents won't be called in pipeline mode. - struct NullSource; - impl SectorSource for NullSource { - fn capacity_sectors(&self) -> u32 { - 0 - } - fn read_sectors( - &mut self, - _: u32, - _: u16, - _: &mut [u8], - _: bool, - ) -> crate::error::Result { - Ok(0) - } - } - let dummy: Box = Box::new(NullSource); - - Self { - reader: DecryptingSectorSource::new(dummy, crate::decrypt::DecryptKeys::None), - title, - disc: None, - decrypt_keys, - extents, - current_extent: 0, - current_offset: 0, - read_buf: Vec::new(), - buf_valid: 0, - adaptive: AdaptiveBatch::new(batch_sectors), - errors: 0, - skip_errors: false, - halt, - event_fn: None, - eof: false, - bytes_read_total: 0, - bytes_total_extents, - ts_demuxer: None, - ps_demuxer: None, - parsers, - pending_frames: std::collections::VecDeque::new(), - pid_to_track, - demux_thread: Some(handle), - demux_rx: Some(rx), } } @@ -596,88 +472,6 @@ fn prof_tick(stage: &str, ns: u128, bytes: u64) { }); } -impl DiscStream { - /// Pipeline-mode `read()` helper: pull one PesPacket batch from - /// the demux thread, run codec parse on each PES, enqueue the - /// resulting PesFrames, return the first one. - fn read_pipeline(&mut self) -> io::Result> { - use super::demux_thread::DemuxBatch; - let rx = self.demux_rx.as_ref().expect("read_pipeline without rx"); - match rx.recv() { - Ok(DemuxBatch::Ts(packets)) => { - let skip_parse = std::env::var_os("FREEMKV_SKIP_PARSE").is_some(); - for pes in packets { - if let Some((_, track)) = self - .pid_to_track - .iter() - .find(|(pid, _)| *pid == pes.pid) - .copied() - { - if skip_parse { - self.pending_frames.push_back(crate::pes::PesFrame { - track, - pts: pes.pts.map(super::codec::pts_to_ns).unwrap_or(0), - keyframe: false, - data: pes.data, - }); - } else if let Some((_, parser)) = - self.parsers.iter_mut().find(|(pid, _)| *pid == pes.pid) - { - for frame in parser.parse(&pes) { - self.pending_frames.push_back( - crate::pes::PesFrame::from_codec_frame(track, frame), - ); - } - } - } - } - Ok(self.pending_frames.pop_front()) - } - Ok(DemuxBatch::Ps(packets)) => { - for ps in packets { - let track = match ps.stream_id { - 0xE0..=0xEF => 0, - 0xC0..=0xDF => 1, - 0xBD => ps - .sub_stream_id - .map(|s| (s & 0x1F) as usize + 1) - .unwrap_or(1), - _ => continue, - }; - if track >= self.title.streams.len() { - continue; - } - let pid = self - .pid_to_track - .iter() - .find(|(_, idx)| *idx == track) - .map(|(p, _)| *p) - .unwrap_or(0); - let pes = super::ts::PesPacket { - pid, - pts: ps.pts.map(|p| p as i64), - dts: ps.dts.map(|d| d as i64), - data: ps.data, - }; - if let Some((_, parser)) = self.parsers.iter_mut().find(|(p, _)| *p == pid) { - for frame in parser.parse(&pes) { - self.pending_frames - .push_back(crate::pes::PesFrame::from_codec_frame(track, frame)); - } - } - } - Ok(self.pending_frames.pop_front()) - } - Ok(DemuxBatch::Err(e)) => Err(e), - Err(_) => { - // Channel closed → demux thread finished. EOF. - self.eof = true; - Ok(None) - } - } - } -} - impl crate::pes::Stream for DiscStream { fn read(&mut self) -> io::Result> { if let Some(frame) = self.pending_frames.pop_front() { @@ -688,19 +482,6 @@ impl crate::pes::Stream for DiscStream { return Ok(None); } - // Pipeline mode: read+decrypt+demux all happen on the demux - // thread. Pull a batch of PesPackets and run codec parse on - // this thread; loop until we have at least one frame OR EOF. - if self.demux_rx.is_some() { - loop { - match self.read_pipeline()? { - Some(f) => return Ok(Some(f)), - None if self.eof => return Ok(None), - None => continue, - } - } - } - loop { let t0 = std::time::Instant::now(); if !self.fill_extents()? { @@ -897,6 +678,10 @@ impl crate::pes::Stream for DiscStream { } true } + + fn errors(&self) -> u64 { + self.errors + } } #[cfg(test)] diff --git a/src/mux/resolve.rs b/src/mux/resolve.rs index c6aa0cd..367566b 100644 --- a/src/mux/resolve.rs +++ b/src/mux/resolve.rs @@ -234,6 +234,7 @@ pub fn input(url: &str, opts: &InputOptions) -> io::Result( batch_sectors: u16, format: ContentFormat, halt: Option, + event_fn: Option, ) -> PipelinedPesStream { let extents = title.extents.clone(); let decrypting = crate::sector::DecryptingSectorSource::new(Box::new(reader) as Box, keys); - let prefetched = crate::sector::PrefetchedSectorSource::new( + let prefetched = crate::sector::PrefetchedSectorSource::new_with_events( decrypting, extents, batch_sectors, halt.clone(), + event_fn, ); let (rx, recycle_tx, shell) = prefetched.into_channels(); diff --git a/src/pes.rs b/src/pes.rs index e03c000..249c2e0 100644 --- a/src/pes.rs +++ b/src/pes.rs @@ -115,6 +115,16 @@ pub trait Stream: Send { fn headers_ready(&self) -> bool { true } + + /// Cumulative count of read errors the stream skipped past (e.g. + /// zero-filled bad sectors on a live drive). Default `0` for + /// streams that don't have a notion of skip-on-error (file ISO, + /// network, stdio, the pipeline highway, etc.); concrete impls + /// with adaptive retry (`DiscStream` on the drive single-pass + /// path) override. + fn errors(&self) -> u64 { + 0 + } } /// Wraps any output stream and counts bytes written. diff --git a/src/sector/prefetched.rs b/src/sector/prefetched.rs index 1a555ee..5bceb24 100644 --- a/src/sector/prefetched.rs +++ b/src/sector/prefetched.rs @@ -41,11 +41,18 @@ //! reading the underlying source directly. use crate::error::Result; +use crate::event::{Event, EventKind}; use crate::halt::Halt; use crate::sector::SectorSource; use crossbeam_channel::{Receiver, Sender, bounded}; use std::thread::JoinHandle; +/// Producer-thread event callback. Fires `BytesRead` after every +/// successful batch read so the consumer side can update a UI +/// progress indicator without polling. `Send + 'static` because the +/// callback runs on the producer thread. +pub type EventFn = Box; + const PREFETCH_CHANNEL_DEPTH: usize = 2; /// Smallest sector source the producer will issue per read. AACS @@ -85,15 +92,33 @@ impl PrefetchedSectorSource { /// — every byte the producer emits is what the consumer's demux /// will feed to its codec parsers. pub fn new( - mut reader: S, + reader: S, extents: Vec, batch_sectors: u16, halt: Option, ) -> Self + where + S: SectorSource + Send + 'static, + { + Self::new_with_events(reader, extents, batch_sectors, halt, None) + } + + /// Same as [`new`] but with a callback fired from the producer + /// thread after each successful batch — used by autorip's mux + /// path to surface `BytesRead` progress to the UI without the + /// consumer thread having to poll. + pub fn new_with_events( + mut reader: S, + extents: Vec, + batch_sectors: u16, + halt: Option, + event_fn: Option, + ) -> Self where S: SectorSource + Send + 'static, { let total_sectors: u32 = extents.iter().map(|e| e.sector_count).sum(); + let bytes_total_extents: u64 = extents.iter().map(|e| e.sector_count as u64 * 2048).sum(); let (tx, rx) = bounded::(PREFETCH_CHANNEL_DEPTH); let (recycle_tx, recycle_rx) = bounded::>(PREFETCH_CHANNEL_DEPTH + 1); let batch_bytes = batch_sectors as usize * 2048; @@ -111,6 +136,7 @@ impl PrefetchedSectorSource { .spawn(move || { let mut ext_idx = 0usize; let mut offset: u32 = 0; + let mut bytes_read_total: u64 = 0; while ext_idx < extents.len() { if halt.as_ref().map(|h| h.is_cancelled()).unwrap_or(false) { return; @@ -144,6 +170,15 @@ impl PrefetchedSectorSource { match reader.read_sectors(lba, sectors, &mut buf[..bytes], false) { Ok(n) => { buf.truncate(n); + bytes_read_total = bytes_read_total.saturating_add(n as u64); + if let Some(ref f) = event_fn { + f(Event { + kind: EventKind::BytesRead { + bytes: bytes_read_total, + total: bytes_total_extents, + }, + }); + } if tx.send(Ok(buf)).is_err() { return; // consumer dropped }