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<dyn Stream> 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

@@ -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,

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
+    // PES output — single-threaded inline demux + codec parse. The
-    // stream is in pipeline mode — the demux state lives inside a
+    // pipeline-mode mux (3-stage threaded) lives in
-    // [`super::demux_thread::DemuxThread`] and PesPackets arrive
+    // [`super::pipelined_stream::PipelinedPesStream`]; this type is
-    // through `demux_rx` already parsed.
+    // the legacy in-thread path for live-disc reads where adaptive
+    // batch retry on bad sectors lives in `fill_extents`.
     ts_demuxer: Option<super::ts::TsDemuxer>,
     ps_demuxer: Option<super::ps::PsDemuxer>,
     parsers: Vec<(u16, Box<dyn super::codec::CodecParser>)>,
     pending_frames: std::collections::VecDeque<crate::pes::PesFrame>,
     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<super::demux_thread::DemuxThread>,
-    demux_rx: Option<crossbeam_channel::Receiver<super::demux_thread::DemuxBatch>>,
 }
 
 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<Halt>,
-    ) -> 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<u16> = 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<dyn SectorSource> 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<usize> {
-                Ok(0)
-            }
-        }
-        let dummy: Box<dyn SectorSource> = 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<Option<crate::pes::PesFrame>> {
-        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<Option<crate::pes::PesFrame>> {
         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)]

src/mux/resolve.rs

@@ -234,6 +234,7 @@ pub fn input(url: &str, opts: &InputOptions) -> io::Result<Box<dyn crate::pes::S
                 ISO_MUX_BATCH_SECTORS,
                 format,
                 None,
+                None,
             );
             Ok(Box::new(stream))
         }
@@ -357,15 +358,17 @@ pub fn build_iso_pipeline<S: SectorSource + Send + 'static>(
     batch_sectors: u16,
     format: ContentFormat,
     halt: Option<crate::halt::Halt>,
+    event_fn: Option<crate::sector::prefetched::EventFn>,
 ) -> PipelinedPesStream {
     let extents = title.extents.clone();
     let decrypting =
         crate::sector::DecryptingSectorSource::new(Box::new(reader) as Box<dyn SectorSource>, 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();
 

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.

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<dyn Fn(Event) + Send + 'static>;
+
 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<S>(
-        mut reader: S,
+        reader: S,
         extents: Vec<crate::disc::Extent>,
         batch_sectors: u16,
         halt: Option<Halt>,
     ) -> 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<S>(
+        mut reader: S,
+        extents: Vec<crate::disc::Extent>,
+        batch_sectors: u16,
+        halt: Option<Halt>,
+        event_fn: Option<EventFn>,
+    ) -> 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::<Batch>(PREFETCH_CHANNEL_DEPTH);
         let (recycle_tx, recycle_rx) = bounded::<Vec<u8>>(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
                             }