diff --git a/src/halt.rs b/src/halt.rs index 7000627..e357b15 100644 --- a/src/halt.rs +++ b/src/halt.rs @@ -34,6 +34,26 @@ impl Halt { Self(Arc::new(AtomicBool::new(false))) } + /// Wrap an existing `Arc` as a `Halt`. Useful as a + /// bridge during the 0.18 deprecation window: callers that already + /// hold an `Arc` (e.g. `Drive::halt_flag()`, the + /// deprecated `DiscStream::set_halt`) can adopt the new token API + /// without changing the underlying flag. + /// + /// Cancelling either side flips the same bit — the wrapping `Halt` + /// and the original `Arc` are two views over one shared flag. + pub fn from_arc(flag: Arc) -> Self { + Self(flag) + } + + /// Borrow the underlying `Arc`. Used at boundaries with + /// pre-`Halt` APIs that still take an `Arc` directly + /// (`CopyOptions::halt`, the deprecated `DiscStream::set_halt`). + /// Round 3 deletes those boundaries and this accessor with them. + pub fn as_arc(&self) -> &Arc { + &self.0 + } + /// Flip the shared flag to cancelled. Idempotent. pub fn cancel(&self) { self.0.store(true, Ordering::Relaxed); @@ -110,4 +130,37 @@ mod tests { handle.join().unwrap(); assert!(h.is_cancelled()); } + + #[test] + fn from_arc_shares_state() { + // The 0.18 deprecation-window bridge: a Halt built from an + // existing Arc must be a *view* over the same bit, + // not a fresh copy. Cancelling either side flips both. + let arc = Arc::new(AtomicBool::new(false)); + let halt = Halt::from_arc(arc.clone()); + assert!(!halt.is_cancelled()); + assert!(!arc.load(Ordering::Relaxed)); + + // Cancel via the wrapping Halt; the original Arc observes it. + halt.cancel(); + assert!(arc.load(Ordering::Relaxed)); + + // Conversely: flip the Arc directly; the Halt view observes it. + let arc2 = Arc::new(AtomicBool::new(false)); + let halt2 = Halt::from_arc(arc2.clone()); + arc2.store(true, Ordering::Relaxed); + assert!(halt2.is_cancelled()); + } + + #[test] + fn as_arc_returns_backing_flag() { + // `as_arc()` must hand back the *same* Arc, not a clone of a + // different bit. Verified by writing through the borrowed Arc + // and observing through the Halt. + let halt = Halt::new(); + let arc = halt.as_arc().clone(); + assert!(!halt.is_cancelled()); + arc.store(true, Ordering::Relaxed); + assert!(halt.is_cancelled()); + } } diff --git a/src/mux/disc.rs b/src/mux/disc.rs index 96e6629..4eca5bc 100644 --- a/src/mux/disc.rs +++ b/src/mux/disc.rs @@ -8,10 +8,11 @@ use crate::disc::{Disc, DiscTitle, Extent}; use crate::drive::extract_scsi_context; use crate::event::{BatchSizeReason, Event, EventKind}; +use crate::halt::Halt; use crate::sector::SectorReader; use std::io; use std::sync::Arc; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::AtomicBool; /// Ramp back up to the preferred batch size after this many sectors /// of clean reading at the current (reduced) size. 100 MiB = 51,200 sectors. @@ -120,11 +121,13 @@ pub struct DiscStream { adaptive: AdaptiveBatch, pub errors: u64, pub skip_errors: bool, - /// When set and the flag is raised, fill_extents returns Err(Halted) at the - /// next retry boundary. Unlike skip_errors, this propagates the error up so - /// the rip terminates cleanly. Share the Arc with Drive::halt_flag() to get - /// unified Stop behavior across drive reads and sector processing. - halt: Option>, + /// When set and the token is cancelled, fill_extents returns Err(Halted) + /// at the next retry boundary. Unlike skip_errors, this propagates the + /// error up so the rip terminates cleanly. Construct with + /// [`DiscStream::with_halt`] (preferred) or set post-hoc via the + /// deprecated [`DiscStream::set_halt`] bridge — both populate this same + /// field and either entry point yields one source of truth. + halt: Option, event_fn: Option>, eof: bool, @@ -219,19 +222,39 @@ impl DiscStream { self.event_fn = Some(Box::new(f)); } - /// Share a halt flag — typically from `Drive::halt_flag()`. When raised, - /// the next read-retry boundary inside fill_extents returns Err(Halted) - /// instead of continuing. Required for Stop to work during dense bad-sector - /// regions (where read() loops internally waiting for enough clean data to - /// emit a frame and would otherwise never check an external stop signal). + /// Constructor-time builder: attach a [`Halt`] token so that when + /// any clone is cancelled, the next read-retry boundary inside + /// `fill_extents` returns `Err(Halted)`. Required for Stop to work + /// during dense bad-sector regions (where the outer PES read() loop + /// can spend minutes inside fill_extents before emitting a frame). + /// + /// Preferred over the post-hoc [`DiscStream::set_halt`] bridge — + /// pass the same `Halt` clone you hand to sweep / patch / mux so + /// every phase observes a single Stop signal. + pub fn with_halt(mut self, halt: Halt) -> Self { + self.halt = Some(halt); + self + } + + /// Bridge for callers that haven't migrated to the + /// [`DiscStream::with_halt`] constructor-time path yet. Wraps the + /// supplied `Arc` as a [`Halt`] (`Halt::from_arc`) and + /// stores it in the same internal slot, so a halt installed via + /// either entry point goes through one halt-check inside + /// `fill_extents`. Calling `set_halt` after `with_halt` (or vice + /// versa) replaces the previous token with the new one. + #[deprecated( + since = "0.18.0", + note = "use `DiscStream::with_halt(Halt)` at construction instead" + )] pub fn set_halt(&mut self, flag: Arc) { - self.halt = Some(flag); + self.halt = Some(Halt::from_arc(flag)); } fn is_halted(&self) -> bool { self.halt .as_ref() - .map(|h| h.load(Ordering::Relaxed)) + .map(|h| h.is_cancelled()) .unwrap_or(false) } @@ -629,4 +652,47 @@ mod tests { } assert_eq!(frames, 0); } + + /// `is_halted()` must observe a cancellation signal regardless of + /// which entry point installed the token. The deprecated + /// `set_halt(Arc)` and the new `with_halt(Halt)` are + /// two views over one slot — flipping either bit must cause the + /// next `fill_extents` retry boundary to bail. + #[test] + fn halt_via_with_halt_observed_by_is_halted() { + let halt = Halt::new(); + let stream = DiscStream::new( + Box::new(ZeroReader { capacity: 8 }), + synthetic_title(8), + crate::decrypt::DecryptKeys::None, + 8, + crate::disc::ContentFormat::BdTs, + ) + .with_halt(halt.clone()); + assert!(!stream.is_halted()); + halt.cancel(); + assert!( + stream.is_halted(), + "with_halt token cancellation must be observed by is_halted()" + ); + } + + #[test] + fn halt_via_set_halt_bridge_observed_by_is_halted() { + let arc = Arc::new(AtomicBool::new(false)); + let mut stream = DiscStream::new( + Box::new(ZeroReader { capacity: 8 }), + synthetic_title(8), + crate::decrypt::DecryptKeys::None, + 8, + crate::disc::ContentFormat::BdTs, + ); + stream.set_halt(arc.clone()); + assert!(!stream.is_halted()); + arc.store(true, std::sync::atomic::Ordering::Relaxed); + assert!( + stream.is_halted(), + "set_halt(Arc) bridge must observe Arc-side flips" + ); + } }