diff --git a/src/disc/mod.rs b/src/disc/mod.rs index 3af4c11..7a12e34 100644 --- a/src/disc/mod.rs +++ b/src/disc/mod.rs @@ -13,7 +13,7 @@ mod dvd; mod encrypt; pub mod mapfile; pub mod read_error; -mod sweep_pipeline; +mod sweep; use crate::drive::{Drive, extract_scsi_context}; use crate::error::{Error, Result}; @@ -1370,10 +1370,8 @@ impl Disc { path: &std::path::Path, opts: &SweepOptions, ) -> Result { - use sweep_pipeline::{ - ConsumerInputs, ProgressSnapshot, WorkItem, send_or_abort, spawn_consumer, - try_recv_progress, try_request_stats, - }; + use crate::io::{DEFAULT_PIPELINE_DEPTH, Pipeline}; + use sweep::{ProgressSnapshot, SweepSink, WorkItem, try_recv_progress}; let total_bytes = self.capacity_sectors as u64 * 2048; let keys = if opts.decrypt { @@ -1438,11 +1436,20 @@ impl Disc { // Spawn the consumer. It owns WritebackFile + Mapfile; the producer // (this thread) keeps `reader`, `read_ctx`, halt + set_speed. - let (work_tx, prog_rx, consumer_handle) = spawn_consumer(ConsumerInputs { - file, - map, - is_regular, - }); + // The thread name is preserved from the 0.17.x sweep_pipeline so it + // stays identifiable in stack traces / `top -H`. + let (sink, prog_rx) = SweepSink::new(file, map, is_regular); + let pipe: Pipeline = + Pipeline::spawn_named("freemkv-sweep-consumer", DEFAULT_PIPELINE_DEPTH, sink)?; + + // Translate `Pipeline::send` failure (consumer gone) into the + // same `Error` shape the 0.17.x `send_or_abort` produced, so + // the producer-error semantics are unchanged. + fn consumer_gone() -> Error { + Error::IoError { + source: std::io::Error::other("sweep consumer terminated unexpectedly"), + } + } let mut buf = vec![0u8; batch as usize * 2048]; let mut bytes_done = 0u64; @@ -1533,10 +1540,8 @@ impl Disc { // owned Vec. The producer's `buf` is reused // for the next read. let send_buf = buf[..block_bytes as usize].to_vec(); - if let Err(e) = - send_or_abort(&work_tx, WorkItem::Good { pos, buf: send_buf }) - { - producer_err = Some(e); + if pipe.send(WorkItem::Good { pos, buf: send_buf }).is_err() { + producer_err = Some(consumer_gone()); break 'outer; } bytes_done = bytes_done.saturating_add(block_bytes); @@ -1595,14 +1600,14 @@ impl Disc { 0, )?; } - if let Err(e) = send_or_abort( - &work_tx, - WorkItem::BisectGood { + if pipe + .send(WorkItem::BisectGood { pos: write_pos, buf: Box::new(sector_buf), - }, - ) { - producer_err = Some(e); + }) + .is_err() + { + producer_err = Some(consumer_gone()); bisect_aborted = true; break; } @@ -1612,11 +1617,11 @@ impl Disc { &inner_err, &mut read_ctx, ); - if let Err(e) = send_or_abort( - &work_tx, - WorkItem::BisectBad { pos: write_pos }, - ) { - producer_err = Some(e); + if pipe + .send(WorkItem::BisectBad { pos: write_pos }) + .is_err() + { + producer_err = Some(consumer_gone()); bisect_aborted = true; break; } @@ -1632,14 +1637,14 @@ impl Disc { pos += block_bytes; } read_error::ReadAction::SkipBlock { pause_secs } => { - if let Err(e) = send_or_abort( - &work_tx, - WorkItem::SkipFill { + if pipe + .send(WorkItem::SkipFill { pos, len: block_bytes, - }, - ) { - producer_err = Some(e); + }) + .is_err() + { + producer_err = Some(consumer_gone()); break 'outer; } bytes_done = bytes_done.saturating_add(block_bytes); @@ -1652,14 +1657,14 @@ impl Disc { sectors, pause_secs, } => { - if let Err(e) = send_or_abort( - &work_tx, - WorkItem::SkipFill { + if pipe + .send(WorkItem::SkipFill { pos, len: block_bytes, - }, - ) { - producer_err = Some(e); + }) + .is_err() + { + producer_err = Some(consumer_gone()); break 'outer; } bytes_done = bytes_done.saturating_add(block_bytes); @@ -1679,14 +1684,14 @@ impl Disc { let gap_start = pos + block_bytes; let gap_bytes = jump_pos.saturating_sub(gap_start); if gap_bytes > 0 { - if let Err(e) = send_or_abort( - &work_tx, - WorkItem::GapFill { + if pipe + .send(WorkItem::GapFill { pos: gap_start, len: gap_bytes, - }, - ) { - producer_err = Some(e); + }) + .is_err() + { + producer_err = Some(consumer_gone()); break 'outer; } bytes_done = bytes_done.saturating_add(gap_bytes); @@ -1741,8 +1746,11 @@ impl Disc { "Disc::sweep inner iter" ); } - // Throttled stats refresh request. - try_request_stats(&work_tx); + // Throttled stats refresh request — best-effort + // try_send so a busy consumer doesn't stall the + // producer; the cached snapshot stays current + // enough for one more iteration. + let _ = pipe.try_send(WorkItem::StatsRequest); } if let Some(reporter) = opts.progress { @@ -1792,24 +1800,25 @@ impl Disc { } } - // Tell the consumer we're done. Even on producer error, send - // Finish so the consumer drains cleanly and we get a summary. - let _ = work_tx.send(WorkItem::Finish); - drop(work_tx); - - let summary = consumer_handle.join().map_err(|_| Error::IoError { - source: std::io::Error::other("sweep consumer thread panicked"), - })?; + // Producer side is done. Drop the channel and let the + // consumer drain whatever's still in flight, then run its + // close() (drain writeback, fsync, mapfile.flush) and return + // the final stats. On consumer panic `pipe.finish` returns + // the wrapped panic message via Error::IoError — same shape + // the previous `consumer_handle.join().map_err(...)` produced. + let summary = pipe.finish(); // Producer-side error wins over consumer-side (the read failure // is what motivated quitting; the consumer's flush error, if // any, is downstream). if let Some(e) = producer_err { + // Drop the consumer's result if we already have a producer + // error, but propagate consumer-panic on top of nothing + // since that's strictly informative. + let _ = summary; return Err(e); } - if let Some(e) = summary.error { - return Err(e); - } + let summary = summary?; let stats = summary.stats; tracing::debug!( @@ -3465,4 +3474,47 @@ mod tests { patch_result.err() ); } + + /// Synthetic regression test for the 0.18 SweepSink + Pipeline + /// migration. ~100 batches of clean reads (6000 sectors at the + /// default 60-sector single-pass batch size); verifies all bytes + /// land in the ISO and the consumer's final stats match the input. + /// The throughput regression check (vs 0.17.13) is a separate + /// manual / live-drive concern; here we only assert correctness. + #[test] + fn sweep_pipeline_full_good_100_batches() { + let tmp = tempfile::tempdir().unwrap(); + let iso_path = tmp.path().join("test.iso"); + // 6000 sectors / 60-sector default batch = exactly 100 + // produce/consume cycles through the pipeline. + let sectors: u32 = 6000; + let mut reader = MockReader { + total_sectors: sectors, + bad_sectors: std::collections::HashSet::new(), + }; + let disc = make_test_disc(sectors, "TPipeline100"); + let opts = CopyOptions { + decrypt: false, + multipass: false, + progress: None, + halt: None, + }; + let result = disc.copy(&mut reader, &iso_path, &opts); + let r = result.expect("100-batch clean sweep should succeed"); + assert!(r.complete, "complete=true expected"); + assert!(!r.halted, "halted=false expected"); + assert_eq!( + r.bytes_good, + sectors as u64 * 2048, + "all sectors must be marked good after a 100% clean sweep" + ); + assert_eq!( + r.bytes_pending, 0, + "no pending bytes expected after a clean sweep" + ); + // The ISO file must end up the right size — the consumer + // wrote everything before fsync. + let meta = std::fs::metadata(&iso_path).unwrap(); + assert_eq!(meta.len(), sectors as u64 * 2048); + } } diff --git a/src/disc/sweep.rs b/src/disc/sweep.rs new file mode 100644 index 0000000..f41c325 --- /dev/null +++ b/src/disc/sweep.rs @@ -0,0 +1,239 @@ +//! `Disc::sweep`'s consumer-side `Sink`. +//! +//! Background: the original sweep loop runs strictly serialised — +//! SCSI read → decrypt → seek + write → mapfile.record → next iter. +//! On a healthy disc the SCSI read costs ~5-12 ms per 64 KB batch and +//! the post-read work (decrypt 1-3 ms + file write + mapfile fsync +//! 5-15 ms) adds another batch's worth of latency. The drive idles +//! during the post-read work; throughput tops out at the *sum* of +//! both costs. +//! +//! 0.17.11 introduced a bespoke producer/consumer split (the now- +//! removed `disc/sweep_pipeline.rs`) to overlap the two stages. 0.18 +//! collapses that split — together with the analogous splits patch +//! and mux need — onto the generic [`crate::io::Pipeline`] + +//! [`crate::io::Sink`] primitive. This module is the sweep-specific +//! `Sink` impl; the producer-side state machine (read_error context, +//! decrypt, set_speed, halt) stays in `Disc::sweep` in `disc/mod.rs`. +//! +//! Correctness invariants preserved (same as 0.17.11): +//! - Mapfile is single-writer (consumer-only). No locking. +//! - All `read_error::ReadCtx` state stays on the producer thread. +//! - `set_speed` calls happen on the producer thread (same thread that +//! owns the `SectorReader`). No new SCSI concurrency. +//! - Per-iteration ordering of file-write → mapfile-record is kept +//! intact in the consumer (write before record), so the on-disk +//! invariant "mapfile only marks Finished what the file has +//! received" survives a crash mid-pass. +//! - The BU40N+Initio bridge wedge concern is unchanged: only one +//! SCSI command in flight at a time, error-path timing identical, +//! no new retry logic. + +use std::io::{Seek, SeekFrom, Write}; +use std::sync::mpsc::{Receiver, SyncSender, sync_channel}; + +use crate::error::Error; +use crate::io::{Flow, Sink}; + +use super::mapfile::{MapStats, Mapfile, SectorStatus}; + +/// Reusable zero buffer for SkipFill / GapFill / BisectBad. 64 KB +/// matches the existing zero_gap chunk size used by the pre-split +/// sweep loop. +const ZERO_CHUNK: usize = 65 * 1024; + +/// Producer → Consumer messages. The consumer applies these in FIFO +/// order; ordering of file writes and mapfile records across items is +/// preserved. +pub(super) enum WorkItem { + /// Successful batch read. Producer has already decrypted `buf` if + /// `opts.decrypt` was set. Consumer writes `buf` at `pos` and + /// records the range as `Finished`. + Good { pos: u64, buf: Vec }, + + /// Bisect inner-loop good single sector (already decrypted by the + /// producer). 2048 bytes. + BisectGood { pos: u64, buf: Box<[u8; 2048]> }, + + /// Bisect inner-loop bad single sector. Consumer writes 2048 + /// zeros at `pos` and records the sector as `NonTrimmed`. + BisectBad { pos: u64 }, + + /// Whole-batch zero-fill (failed batch on `SkipBlock`, or the + /// failed batch portion of `JumpAhead`). Consumer streams zeros + /// across `[pos, pos+len)` and records the range as `NonTrimmed`. + SkipFill { pos: u64, len: u64 }, + + /// Gap fill following a `JumpAhead`. Same effect as `SkipFill`; + /// distinguished only so future logging / instrumentation can + /// tell them apart without parsing a flag. + GapFill { pos: u64, len: u64 }, + + /// Producer wants the latest mapfile stats for the progress + /// callback. Consumer responds on `prog_tx` with a fresh + /// [`ProgressSnapshot`]. Best-effort: if the producer hasn't + /// drained the previous snapshot, the new one is silently + /// dropped — the producer's local cache stays current enough. + StatsRequest, +} + +/// Snapshot the consumer sends back to the producer for the progress +/// callback. +pub(super) struct ProgressSnapshot { + pub stats: MapStats, + pub bad_ranges: Vec<(u64, u64)>, +} + +/// Final summary returned by the consumer thread on shutdown — what +/// `SweepSink::close` produces, surfaced to the producer via +/// `Pipeline::finish`. +pub(super) struct ConsumerSummary { + pub stats: MapStats, +} + +/// Drain any pending progress snapshots from the consumer. Returns +/// the most recent one, if any. The producer caches it and uses it +/// for subsequent progress callbacks until a fresh one arrives. +pub(super) fn try_recv_progress(rx: &Receiver) -> Option { + let mut latest = None; + while let Ok(snap) = rx.try_recv() { + latest = Some(snap); + } + latest +} + +/// `Sink` for sweep. Owns the writeback file + mapfile + +/// progress back-channel. `apply` carries the file-write + +/// mapfile.record per item; `close` drains the writeback pipeline, +/// fsyncs the ISO, and flushes the mapfile. +pub(super) struct SweepSink { + file: crate::io::WritebackFile, + map: Mapfile, + /// `sync_all`-on-failure-is-an-error iff the output is a regular + /// file. `/dev/null` and pipes always fail `sync_all`; that's not + /// a real error. + is_regular: bool, + /// Back-channel for `StatsRequest` responses. The producer caches + /// the latest snapshot and uses it for the progress callback; + /// dropped sends on a full channel are by design. + prog_tx: SyncSender, + /// Reusable zero buffer for SkipFill / GapFill / BisectBad. Held + /// in the sink so each apply call doesn't reallocate. + zero: Box<[u8; ZERO_CHUNK]>, +} + +impl SweepSink { + /// Construct a new `SweepSink` plus the matching progress + /// receiver. Channel depth on the back-channel is `1` — the + /// producer's cache is the source of truth between snapshots. + pub(super) fn new( + file: crate::io::WritebackFile, + map: Mapfile, + is_regular: bool, + ) -> (Self, Receiver) { + let (prog_tx, prog_rx) = sync_channel::(1); + let sink = SweepSink { + file, + map, + is_regular, + prog_tx, + zero: Box::new([0u8; ZERO_CHUNK]), + }; + (sink, prog_rx) + } +} + +impl Sink for SweepSink { + type Output = ConsumerSummary; + + fn apply(&mut self, item: WorkItem) -> Result { + match item { + WorkItem::Good { pos, buf } => { + // Decrypt is on the producer; consumer assumes plaintext. + let len = buf.len() as u64; + self.file + .seek(SeekFrom::Start(pos)) + .map_err(|e| Error::IoError { source: e })?; + self.file + .write_all(&buf) + .map_err(|e| Error::IoError { source: e })?; + self.map + .record(pos, len, SectorStatus::Finished) + .map_err(|e| Error::IoError { source: e })?; + } + WorkItem::BisectGood { pos, buf } => { + self.file + .seek(SeekFrom::Start(pos)) + .map_err(|e| Error::IoError { source: e })?; + self.file + .write_all(&buf[..]) + .map_err(|e| Error::IoError { source: e })?; + self.map + .record(pos, 2048, SectorStatus::Finished) + .map_err(|e| Error::IoError { source: e })?; + } + WorkItem::BisectBad { pos } => { + self.file + .seek(SeekFrom::Start(pos)) + .map_err(|e| Error::IoError { source: e })?; + self.file + .write_all(&self.zero[..2048]) + .map_err(|e| Error::IoError { source: e })?; + self.map + .record(pos, 2048, SectorStatus::NonTrimmed) + .map_err(|e| Error::IoError { source: e })?; + } + WorkItem::SkipFill { pos, len } | WorkItem::GapFill { pos, len } => { + self.file + .seek(SeekFrom::Start(pos)) + .map_err(|e| Error::IoError { source: e })?; + // Subsequent writes are sequential; `WritebackFile`'s + // seek-elision keeps them on the writeback pipeline path. + let mut filled = 0u64; + while filled < len { + let chunk = (len - filled).min(self.zero.len() as u64) as usize; + self.file + .write_all(&self.zero[..chunk]) + .map_err(|e| Error::IoError { source: e })?; + filled += chunk as u64; + } + self.map + .record(pos, len, SectorStatus::NonTrimmed) + .map_err(|e| Error::IoError { source: e })?; + } + WorkItem::StatsRequest => { + let stats = self.map.stats(); + let bad_ranges = self.map.ranges_with(&[ + SectorStatus::NonTrimmed, + SectorStatus::Unreadable, + SectorStatus::NonScraped, + SectorStatus::NonTried, + ]); + // Best-effort: drop on backpressure; producer's cache + // stays current enough. + let _ = self + .prog_tx + .try_send(ProgressSnapshot { stats, bad_ranges }); + } + } + Ok(Flow::Continue) + } + + fn close(mut self) -> Result { + // Drain the writeback pipeline + fsync the ISO, then persist + // any pending mapfile state. Same finalisation order as the + // pre-Pipeline consumer loop. + if let Err(e) = self.file.sync_all() { + if self.is_regular { + return Err(Error::IoError { source: e }); + } + // Non-regular outputs (/dev/null, pipes) always fail + // sync_all; that's not a real error. + } + self.map.flush().map_err(|e| Error::IoError { source: e })?; + + Ok(ConsumerSummary { + stats: self.map.stats(), + }) + } +} diff --git a/src/disc/sweep_pipeline.rs b/src/disc/sweep_pipeline.rs deleted file mode 100644 index eff66c8..0000000 --- a/src/disc/sweep_pipeline.rs +++ /dev/null @@ -1,313 +0,0 @@ -//! Producer / consumer split for `Disc::sweep`. -//! -//! Background: the original sweep loop runs strictly serialised — -//! SCSI read → decrypt → seek + write → mapfile.record → next iter. -//! On a healthy disc the SCSI read costs ~5-12 ms per 64 KB batch and -//! the post-read work (decrypt 1-3 ms + file write + mapfile fsync -//! 5-15 ms) adds another batch's worth of latency. The drive idles -//! during the post-read work; throughput tops out at the *sum* of -//! both costs. -//! -//! This module decouples them. A consumer thread owns the -//! [`crate::io::WritebackFile`] (the ISO file) and the -//! [`super::mapfile::Mapfile`]. The producer thread (the caller of -//! `Disc::sweep`) keeps the [`crate::sector::SectorReader`], the -//! [`super::read_error`] state machine, and decrypt — so what enters -//! the channel is already-clean cleartext bytes, matching the "disc -//! hands plaintext to its consumers" semantic that -//! [`crate::mux::DiscStream`] already follows. Producer and consumer -//! run concurrently; with a healthy disc the drive can read the next -//! batch while the previous one is being written and recorded. -//! -//! Correctness invariants preserved: -//! - Mapfile is single-writer (consumer-only). No locking. -//! - All `read_error::ReadCtx` state stays on the producer thread. -//! - `set_speed` calls happen on the producer thread (same thread that -//! owns the `SectorReader`). No new SCSI concurrency. -//! - Per-iteration ordering of file-write → mapfile-record is kept -//! intact in the consumer (write before record, same as today), so -//! the on-disk invariant "mapfile only marks Finished what the file -//! has received" survives a crash mid-pass. -//! - The BU40N+Initio bridge wedge concern is unchanged: only one -//! SCSI command in flight at a time, error-path timing identical, -//! no new retry logic. - -use std::io::{Seek, SeekFrom, Write}; -use std::sync::mpsc::{Receiver, SyncSender, TrySendError, sync_channel}; -use std::thread::{self, JoinHandle}; - -use crate::error::{Error, Result}; - -use super::mapfile::{MapStats, Mapfile, SectorStatus}; - -/// Channel depth for in-flight work items. 4 is enough to absorb a -/// mapfile-flush burst on the consumer without growing memory -/// unboundedly. Producer back-pressure is the natural rate limiter: -/// `SyncSender::send` blocks when the channel is full. -const CHANNEL_DEPTH: usize = 4; - -/// Reusable zero buffer for SkipFill / GapFill / BisectBad. 64 KB -/// matches the existing zero_gap chunk size used by the pre-split -/// sweep loop. -const ZERO_CHUNK: usize = 65 * 1024; - -/// Producer → Consumer messages. The consumer applies these in FIFO -/// order; ordering of file writes and mapfile records across items is -/// preserved. -pub(super) enum WorkItem { - /// Successful batch read. Producer has already decrypted `buf` if - /// `opts.decrypt` was set. Consumer writes `buf` at `pos` and - /// records the range as `Finished`. - Good { pos: u64, buf: Vec }, - - /// Bisect inner-loop good single sector (already decrypted by the - /// producer). 2048 bytes. - BisectGood { pos: u64, buf: Box<[u8; 2048]> }, - - /// Bisect inner-loop bad single sector. Consumer writes 2048 - /// zeros at `pos` and records the sector as `NonTrimmed`. - BisectBad { pos: u64 }, - - /// Whole-batch zero-fill (failed batch on `SkipBlock`, or the - /// failed batch portion of `JumpAhead`). Consumer streams zeros - /// across `[pos, pos+len)` and records the range as `NonTrimmed`. - SkipFill { pos: u64, len: u64 }, - - /// Gap fill following a `JumpAhead`. Same effect as `SkipFill`; - /// distinguished only so future logging / instrumentation can - /// tell them apart without parsing a flag. - GapFill { pos: u64, len: u64 }, - - /// Producer wants the latest mapfile stats for the progress - /// callback. Consumer responds on `prog_tx` with a fresh - /// [`ProgressSnapshot`]. Best-effort: if the producer hasn't - /// drained the previous snapshot, the new one is silently - /// dropped — the producer's local cache stays current enough. - StatsRequest, - - /// Producer is done. Consumer drains, runs `sync_all` on the - /// file, and exits. Final stats are returned via the - /// `JoinHandle` from `spawn_consumer`. - Finish, -} - -/// Snapshot the consumer sends back to the producer for the progress -/// callback. -pub(super) struct ProgressSnapshot { - pub stats: MapStats, - pub bad_ranges: Vec<(u64, u64)>, -} - -/// Final summary returned by the consumer thread on shutdown. The -/// producer reads it via the `JoinHandle` returned from -/// [`spawn_consumer`]. -pub(super) struct ConsumerSummary { - pub stats: MapStats, - /// First mapfile/write error the consumer hit, if any. The - /// producer treats this as fatal on the way back up. - pub error: Option, -} - -/// Owned bundle the consumer thread takes ownership of. Decrypt -/// happens on the producer side before send, so the consumer never -/// sees keys. -pub(super) struct ConsumerInputs { - pub file: crate::io::WritebackFile, - pub map: Mapfile, - /// `sync_all`-on-failure-is-an-error iff the output is a regular - /// file. `/dev/null` and pipes always fail `sync_all`; that's not - /// a real error. - pub is_regular: bool, -} - -/// Spawn the consumer thread. The producer keeps the work-tx and -/// prog-rx; the join handle yields the final summary on `Finish` (or -/// on channel close). -pub(super) fn spawn_consumer( - inputs: ConsumerInputs, -) -> ( - SyncSender, - Receiver, - JoinHandle, -) { - let (work_tx, work_rx) = sync_channel::(CHANNEL_DEPTH); - let (prog_tx, prog_rx) = sync_channel::(1); - - let handle = thread::Builder::new() - .name("freemkv-sweep-consumer".into()) - .spawn(move || consumer_loop(inputs, work_rx, prog_tx)) - .expect("spawning sweep consumer thread should not fail"); - - (work_tx, prog_rx, handle) -} - -/// Send a work item, translating a `SendError` (consumer thread died -/// / panicked) into a useful library error so the caller can -/// propagate cleanly. -pub(super) fn send_or_abort(tx: &SyncSender, item: WorkItem) -> Result<()> { - tx.send(item).map_err(|_| Error::IoError { - source: std::io::Error::other("sweep consumer terminated unexpectedly"), - }) -} - -/// Best-effort `StatsRequest` send. If the channel is full, skip — -/// the producer's cached snapshot is fine for one more iteration. -pub(super) fn try_request_stats(tx: &SyncSender) { - if let Err(TrySendError::Full(_)) = tx.try_send(WorkItem::StatsRequest) { - // expected when the consumer is busy; cached snapshot is - // still recent enough. - } -} - -/// Drain any pending progress snapshots from the consumer. Returns -/// the most recent one, if any. The producer caches it and uses it -/// for subsequent progress callbacks until a fresh one arrives. -pub(super) fn try_recv_progress(rx: &Receiver) -> Option { - let mut latest = None; - while let Ok(snap) = rx.try_recv() { - latest = Some(snap); - } - latest -} - -fn consumer_loop( - mut inputs: ConsumerInputs, - work_rx: Receiver, - prog_tx: SyncSender, -) -> ConsumerSummary { - let zero = [0u8; ZERO_CHUNK]; - let mut first_error: Option = None; - - // Channel closed without a Finish == treat as Finish (producer - // dropped tx without explicit teardown — should not happen in - // normal operation but be defensive). - while let Ok(item) = work_rx.recv() { - // Once we have an error, drain remaining items without - // applying side-effects so the producer never blocks on a - // dead consumer. Loop until Finish or channel close. - if first_error.is_some() { - if matches!(item, WorkItem::Finish) { - break; - } - continue; - } - - match apply_item(&mut inputs, item, &zero, &prog_tx) { - Ok(true) => {} - Ok(false) => break, // Finish received - Err(e) => first_error = Some(e), - } - } - - // Final flush — drain the writeback pipeline + fsync the ISO, - // then persist any pending mapfile state. - if first_error.is_none() { - if let Err(e) = inputs.file.sync_all() { - if inputs.is_regular { - first_error = Some(Error::IoError { source: e }); - } - // Non-regular outputs (/dev/null, pipes) always fail - // sync_all; that's not a real error. - } - if let Err(e) = inputs.map.flush() { - first_error = Some(Error::IoError { source: e }); - } - } - - ConsumerSummary { - stats: inputs.map.stats(), - error: first_error, - } -} - -/// Apply a single `WorkItem`. Returns `Ok(true)` to continue, `Ok(false)` -/// to break the consumer loop on `Finish`, `Err(_)` on first failure -/// (caller captures and continues draining). -fn apply_item( - inputs: &mut ConsumerInputs, - item: WorkItem, - zero: &[u8; ZERO_CHUNK], - prog_tx: &SyncSender, -) -> Result { - match item { - WorkItem::Good { pos, buf } => { - // Decrypt is on the producer; consumer assumes plaintext. - let len = buf.len() as u64; - inputs - .file - .seek(SeekFrom::Start(pos)) - .map_err(|e| Error::IoError { source: e })?; - inputs - .file - .write_all(&buf) - .map_err(|e| Error::IoError { source: e })?; - inputs - .map - .record(pos, len, SectorStatus::Finished) - .map_err(|e| Error::IoError { source: e })?; - } - WorkItem::BisectGood { pos, buf } => { - inputs - .file - .seek(SeekFrom::Start(pos)) - .map_err(|e| Error::IoError { source: e })?; - inputs - .file - .write_all(&buf[..]) - .map_err(|e| Error::IoError { source: e })?; - inputs - .map - .record(pos, 2048, SectorStatus::Finished) - .map_err(|e| Error::IoError { source: e })?; - } - WorkItem::BisectBad { pos } => { - inputs - .file - .seek(SeekFrom::Start(pos)) - .map_err(|e| Error::IoError { source: e })?; - inputs - .file - .write_all(&zero[..2048]) - .map_err(|e| Error::IoError { source: e })?; - inputs - .map - .record(pos, 2048, SectorStatus::NonTrimmed) - .map_err(|e| Error::IoError { source: e })?; - } - WorkItem::SkipFill { pos, len } | WorkItem::GapFill { pos, len } => { - inputs - .file - .seek(SeekFrom::Start(pos)) - .map_err(|e| Error::IoError { source: e })?; - // Subsequent writes are sequential; `crate::io::WritebackFile`'s - // seek-elision keeps them on the writeback pipeline path. - let mut filled = 0u64; - while filled < len { - let chunk = (len - filled).min(zero.len() as u64) as usize; - inputs - .file - .write_all(&zero[..chunk]) - .map_err(|e| Error::IoError { source: e })?; - filled += chunk as u64; - } - inputs - .map - .record(pos, len, SectorStatus::NonTrimmed) - .map_err(|e| Error::IoError { source: e })?; - } - WorkItem::StatsRequest => { - let stats = inputs.map.stats(); - let bad_ranges = inputs.map.ranges_with(&[ - SectorStatus::NonTrimmed, - SectorStatus::Unreadable, - SectorStatus::NonScraped, - SectorStatus::NonTried, - ]); - // Best-effort: drop on backpressure; producer's cache - // stays current enough. - let _ = prog_tx.try_send(ProgressSnapshot { stats, bad_ranges }); - } - WorkItem::Finish => return Ok(false), - } - Ok(true) -} diff --git a/src/io/mod.rs b/src/io/mod.rs index a119dbd..8d806ed 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -21,9 +21,11 @@ pub mod pipeline; pub(crate) use writeback_file::WritebackFile; -// Re-exports for the 0.18 redesign. Currently flagged unused because -// no in-tree call site has been migrated yet (sweep is still on -// `disc/sweep_pipeline.rs`; patch and mux have no pipeline). The next -// 0.18 slice removes this allow as it wires up the first consumer. +// Re-exports for the 0.18 redesign. Sweep is wired up in +// `disc/sweep.rs`; patch and mux migrate in later 0.18 slices. +// `WRITE_THROUGH_DEPTH` is patch-only and currently has no in-tree +// caller — the targeted `#[allow]` keeps the re-export visible +// without dragging the rest of the module under `dead_code`. #[allow(unused_imports)] -pub use pipeline::{DEFAULT_PIPELINE_DEPTH, Flow, Pipeline, Sink, WRITE_THROUGH_DEPTH}; +pub use pipeline::WRITE_THROUGH_DEPTH; +pub use pipeline::{DEFAULT_PIPELINE_DEPTH, Flow, Pipeline, Sink}; diff --git a/src/io/pipeline.rs b/src/io/pipeline.rs index 6690fdb..9b0f137 100644 --- a/src/io/pipeline.rs +++ b/src/io/pipeline.rs @@ -5,9 +5,10 @@ //! The consumer's behaviour is supplied by a [`Sink`] implementation: //! `apply` is called once per item, `close` is called once at the end. //! -//! Three call sites in libfreemkv have grown a producer/consumer split -//! independently — sweep already has one (in `disc/sweep_pipeline.rs`), -//! patch and mux do not. 0.18 collapses all three onto this primitive. +//! Three call sites in libfreemkv want a producer/consumer split — +//! sweep (migrated to `disc/sweep.rs::SweepSink`), patch, and mux. +//! 0.18 collapses all three onto this primitive; sweep is in, +//! patch and mux migrate in later 0.18 slices. //! See `(internal)/memory/0_18_redesign.md` for the full picture. //! //! ## Cancellation and error semantics @@ -28,15 +29,10 @@ //! //! ## Dead-code suppression //! -//! The `Pipeline` / `Sink` / `Flow` / `DEFAULT_PIPELINE_DEPTH` / -//! `WRITE_THROUGH_DEPTH` items are crate-internal API today (the -//! parent `io` module is `pub(crate)`) but have no in-tree callers -//! in this slice — sweep is still on `disc/sweep_pipeline.rs`, patch -//! and mux still have no pipeline at all. Wiring them up is the -//! next slice of the 0.18 redesign. The `#[allow]` below is removed -//! once any of those three call sites lands on this primitive. - -#![allow(dead_code)] +//! `WRITE_THROUGH_DEPTH` has no in-tree caller yet — patch is the +//! intended consumer (write-through depth=1) and migrates in a later +//! 0.18 slice. The constant ships now so the contract for that slice +//! is fixed; the targeted `#[allow]` is removed when patch lands. use std::io; use std::sync::mpsc::{SyncSender, sync_channel}; @@ -50,8 +46,9 @@ use crate::error::Error; /// Empirically tuned for sweep and mux — both want enough slack that /// short consumer stalls don't immediately back up onto the producer, /// but not so much that a producer outpacing the consumer accumulates -/// arbitrary buffered work. `4` matches the depth `disc/sweep_pipeline.rs` -/// has used since 0.17.11. Patch should usually use +/// arbitrary buffered work. `4` matches the depth sweep has used +/// since 0.17.11 (originally in `disc/sweep_pipeline.rs`, now in +/// `disc/sweep.rs::SweepSink`). Patch should usually use /// [`WRITE_THROUGH_DEPTH`] (`1`) instead — write-through gives clean /// back-pressure between every read attempt and the matching write, /// which matters when the consumer is updating the mapfile in lockstep. @@ -61,13 +58,21 @@ pub const DEFAULT_PIPELINE_DEPTH: usize = 4; /// drains before the next can enqueue. Use this when the producer /// must observe consumer side-effects (e.g. mapfile state) before /// emitting the next item. +#[allow(dead_code)] pub const WRITE_THROUGH_DEPTH: usize = 1; /// Outcome of [`Sink::apply`]: either keep feeding items /// ([`Flow::Continue`]), or stop the pipeline early and run `close()` /// ([`Flow::Stop`]). +/// +/// `Stop` has no in-tree caller in this slice — sweep never returns +/// it (it always processes the producer's full work-list before the +/// channel is dropped). Patch and mux are the intended consumers and +/// migrate in later 0.18 slices. The variant ships now so the contract +/// is fixed; the targeted `#[allow]` is removed when patch lands. pub enum Flow { Continue, + #[allow(dead_code)] Stop, } @@ -106,14 +111,35 @@ impl Pipeline { /// [`Sink`]. /// /// The thread is named `freemkv-pipeline-consumer` so it shows up - /// distinctly in stack traces and `top -H`. Returns an - /// `Error::IoError` if the OS refuses the thread spawn (resource - /// exhaustion); callers already operate in fallible context, so - /// this is propagated rather than panicked. + /// distinctly in stack traces and `top -H`. Callers that want a + /// more specific name (e.g. `freemkv-sweep-consumer`) should use + /// [`Pipeline::spawn_named`] instead. Returns an `Error::IoError` + /// if the OS refuses the thread spawn (resource exhaustion); + /// callers already operate in fallible context, so this is + /// propagated rather than panicked. + /// + /// Sweep uses [`Pipeline::spawn_named`] directly so the consumer + /// thread shows up as `freemkv-sweep-consumer`; this function has + /// no in-tree caller yet. Patch and mux migrate in later 0.18 + /// slices. The targeted `#[allow]` is removed when one of them + /// lands on the default name. + #[allow(dead_code)] pub fn spawn>(depth: usize, sink: S) -> Result { + Self::spawn_named("freemkv-pipeline-consumer", depth, sink) + } + + /// Like [`Pipeline::spawn`] but lets the caller supply the + /// consumer thread's name. Useful when several pipelines run in + /// the same process and stack traces / `top -H` need to tell them + /// apart (e.g. `freemkv-sweep-consumer`, `freemkv-mux-consumer`). + pub fn spawn_named>( + name: &str, + depth: usize, + sink: S, + ) -> Result { let (tx, rx) = sync_channel::(depth); let handle = thread::Builder::new() - .name("freemkv-pipeline-consumer".into()) + .name(name.into()) .spawn(move || -> Result { let mut sink = sink; let mut first_err: Option = None; @@ -162,6 +188,15 @@ impl Pipeline { self.tx.send(item).map_err(|e| e.0) } + /// Non-blocking variant of [`Pipeline::send`]. If the channel is + /// full or the consumer has hung up, the item is returned in + /// `Err`. Useful for best-effort signalling (e.g. sweep's + /// throttled `StatsRequest`) where dropping the message is + /// preferable to blocking the producer. + pub fn try_send(&self, item: I) -> Result<(), std::sync::mpsc::TrySendError> { + self.tx.try_send(item) + } + /// Drop the producer-side channel and wait for the consumer /// thread to finish. Returns whatever the consumer's `close()` /// produced, or the first `apply` error, or — on consumer panic —