0.18 round 2: thread Halt through DiscStream construction
# Conflicts: # src/mux/disc.rs
This commit is contained in:
+53
@@ -34,6 +34,26 @@ impl Halt {
|
||||
Self(Arc::new(AtomicBool::new(false)))
|
||||
}
|
||||
|
||||
/// Wrap an existing `Arc<AtomicBool>` as a `Halt`. Useful as a
|
||||
/// bridge during the 0.18 deprecation window: callers that already
|
||||
/// hold an `Arc<AtomicBool>` (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<AtomicBool>) -> Self {
|
||||
Self(flag)
|
||||
}
|
||||
|
||||
/// Borrow the underlying `Arc<AtomicBool>`. Used at boundaries with
|
||||
/// pre-`Halt` APIs that still take an `Arc<AtomicBool>` directly
|
||||
/// (`CopyOptions::halt`, the deprecated `DiscStream::set_halt`).
|
||||
/// Round 3 deletes those boundaries and this accessor with them.
|
||||
pub fn as_arc(&self) -> &Arc<AtomicBool> {
|
||||
&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<AtomicBool> 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());
|
||||
}
|
||||
}
|
||||
|
||||
+79
-13
@@ -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::{DecryptingSectorSource, SectorReader, SectorSource};
|
||||
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.
|
||||
@@ -130,11 +131,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<Arc<AtomicBool>>,
|
||||
/// 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<Halt>,
|
||||
event_fn: Option<Box<dyn Fn(Event) + Send>>,
|
||||
eof: bool,
|
||||
|
||||
@@ -233,19 +236,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<AtomicBool>` 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<AtomicBool>) {
|
||||
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)
|
||||
}
|
||||
|
||||
@@ -645,4 +668,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<AtomicBool>)` 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<AtomicBool>) bridge must observe Arc-side flips"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user