diff --git a/CHANGELOG.md b/CHANGELOG.md index fbcc6d5..787e829 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,60 @@ # Changelog +## 0.17.11 (2026-05-09) + +### Sweep producer/consumer split — overlap drive read with file write + +Pre-0.17.11 the sweep loop ran strictly serialised: SCSI read → decrypt +→ seek + write → mapfile.record → next read. The drive idled for the +post-read work; throughput capped at the **sum** of both costs. On a +healthy disc that's ~7-12 ms read + ~5-15 ms write/record per 64 KB +batch — limiting sustained throughput to ~10-12 MB/s on the test bed +(BU40N + UHD inner zone), well below the drive's ceiling of ~14-16 MB/s. + +This release decouples them with a producer / consumer split: + +- **Producer** (caller's thread): owns the `SectorReader`, the entire + `read_error` state machine (Retry, Bisect, SkipBlock, JumpAhead, + AbortPass), `set_speed` damage-zone transitions, halt check, and + decrypt. Hands plaintext bytes to the consumer. +- **Consumer** (one spawned thread): owns the `crate::io::Writer` and + the `Mapfile`. Receives `WorkItem` messages and applies the file + write + mapfile record per item. +- **Channel**: bounded `mpsc::sync_channel(4)` — natural back-pressure + via blocking `send` when consumer falls behind. + +While the consumer is writing batch N to disk and updating the mapfile, +the producer is already reading batch N+1 from the drive. Steady-state +throughput is now bound by the slower of the two pipelines (the drive, +on a healthy disc) instead of their sum. + +Side effects of the refactor: + +- **Bisect path now decrypts.** Pre-0.17.11 the bisect inner loop wrote + raw cyphertext when `decrypt=true` and a single sector was recovered + via single-sector retry — a quiet correctness bug exercised only by + the (rare) batch-fail-then-bisect-succeed path on encrypted discs. + The new producer-side decrypt covers both the main success path and + the bisect inner success path. +- All `read_error::ReadCtx` state stays single-threaded on the producer + (damage window, jump multiplier, consecutive-good count, etc.). No + locking added. +- Mapfile remains single-writer (consumer-only). No locking. +- Halt-flag responsiveness unchanged: producer breaks the loop on + signal, sends `Finish`, consumer drains its ≤4 in-flight items and + exits within ~1 batch (~12 ms typical). +- BU40N + Initio bridge wedge concern unchanged: still one SCSI command + in flight, error-path timing identical, no new retry logic. + +New module: `src/disc/sweep_pipeline.rs` (`WorkItem`, `ProgressSnapshot`, +`ConsumerInputs`, `spawn_consumer`, `consumer_loop`, send/recv helpers). +Public API surface unchanged — `Disc::copy` / `CopyOptions` / +`CopyResult` look identical to callers. + +Patch (Pass N) is **not** affected by this release. Patch is bound by +drive recovery time (60 s timeouts on bad sectors), not the read↔write +serialisation; a similar split there would yield negligible benefit. + ## 0.17.10 (2026-05-09) ### Bounded-cache writeback for big sequential writes diff --git a/Cargo.toml b/Cargo.toml index ac848e4..f09ff81 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "libfreemkv" -version = "0.17.10" +version = "0.17.11" edition = "2024" rust-version = "1.86" license = "AGPL-3.0-only" diff --git a/src/disc/mod.rs b/src/disc/mod.rs index 18ffe76..faf7507 100644 --- a/src/disc/mod.rs +++ b/src/disc/mod.rs @@ -13,6 +13,7 @@ mod dvd; mod encrypt; pub mod mapfile; pub mod read_error; +mod sweep_pipeline; use crate::drive::{Drive, extract_scsi_context}; use crate::error::{Error, Result}; @@ -1369,7 +1370,10 @@ impl Disc { path: &std::path::Path, opts: &SweepOptions, ) -> Result { - use std::io::{Seek, SeekFrom, Write}; + use sweep_pipeline::{ + ConsumerInputs, ProgressSnapshot, WorkItem, send_or_abort, spawn_consumer, + try_recv_progress, try_request_stats, + }; let total_bytes = self.capacity_sectors as u64 * 2048; let keys = if opts.decrypt { @@ -1383,7 +1387,7 @@ impl Disc { if !opts.resume { let _ = std::fs::remove_file(&mapfile_path); } - let mut map = mapfile::Mapfile::open_or_create( + let map = mapfile::Mapfile::open_or_create( &mapfile_path, total_bytes, concat!("libfreemkv v", env!("CARGO_PKG_VERSION")), @@ -1414,17 +1418,31 @@ impl Disc { f }; - // Wrap the raw `File` in our bounded-cache writer so the - // kernel's writeback queue drains continuously instead of - // accumulating hundreds of MB of dirty pages and then bursting - // a flush that blocks app writes (see `crate::io`). - let mut file = crate::io::Writer::new(file).map_err(|e| Error::IoError { source: e })?; + // Wrap the raw `File` in our bounded-cache writer (drains + // dirty pages continuously instead of bursting; see + // `crate::io`). The Writer moves into the consumer thread. + let file = crate::io::Writer::new(file).map_err(|e| Error::IoError { source: e })?; let batch: u16 = match opts.batch_sectors { Some(b) => b, None if opts.skip_on_error => ecc_sectors(self.format), None => DEFAULT_BATCH_SECTORS, }; + // Pre-compute the list of NonTried regions before handing the + // mapfile to the consumer thread. Each region is processed by + // the producer in order; the consumer mutates the mapfile per + // work-item. Any regions left as NonTrimmed/Unreadable after + // sweep finishes are the patch pass's job. + let regions: Vec<(u64, u64)> = map.ranges_with(&[mapfile::SectorStatus::NonTried]); + + // Spawn the consumer. It owns Writer + 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, + }); + let mut buf = vec![0u8; batch as usize * 2048]; let mut bytes_done = 0u64; let mut halt_requested = false; @@ -1433,49 +1451,23 @@ impl Disc { let mut read_ok_count: u64 = 0; let mut read_err_count: u64 = 0; let mut last_log_iter: u64 = 0; - // ALL read state lives in one place. The single error-handling - // entry point (`read_error::handle_read_error`) owns the - // counters, retry budgets, and damage-window updates. let mut read_ctx = read_error::ReadCtx::for_sweep(batch); - // Speed control derives from damage-zone state. We track the - // transition locally so we only call set_speed on edges, not - // every iteration. let mut in_damage_zone = false; const DAMAGE_ZONE_EXIT_THRESHOLD: u64 = 16; + let mut cached_snapshot: Option = None; + let mut producer_err: Option = None; + tracing::trace!( target: "freemkv::disc", phase = "copy_start", total_bytes, batch, skip_on_error = opts.skip_on_error, - "Disc::copy entered" + regions = regions.len(), + "Disc::sweep entered (producer/consumer)" ); - 'outer: loop { - // Every pass retries every non-Finished range. Includes - // Unreadable so each pass gets its own shot at sectors prior - // passes gave up on — drive state may have changed (cooled - // down, bridge stabilized, etc.). Mapfile is binary in - // intent: Finished or not-yet-good. - let regions_to_do = map.ranges_with(&[ - mapfile::SectorStatus::NonTried, - mapfile::SectorStatus::NonTrimmed, - mapfile::SectorStatus::NonScraped, - mapfile::SectorStatus::Unreadable, - ]); - tracing::trace!( - target: "freemkv::disc", - phase = "outer_loop", - regions_remaining = regions_to_do.len(), - "Disc::copy outer iter" - ); - if regions_to_do.is_empty() { - break; - } - let Some((region_pos, region_size)) = map.next_with(0, mapfile::SectorStatus::NonTried) - else { - break; - }; + 'outer: for (region_pos, region_size) in regions { let region_end = region_pos + region_size; let mut pos = region_pos; tracing::trace!( @@ -1509,12 +1501,9 @@ impl Disc { match read_result { Ok(_) => { - // === SUCCESS PATH === read_ok_count += 1; read_ctx.on_success(); - // Damage-zone exit: after enough consecutive good - // reads, restore max speed and reset jump multiplier. if read_ctx.consecutive_good >= DAMAGE_ZONE_EXIT_THRESHOLD { read_ctx.jump_multiplier = 1; if in_damage_zone { @@ -1530,6 +1519,7 @@ impl Disc { } read_ctx.bridge_degradation_count = 0; + // Decrypt on producer; consumer expects plaintext. if opts.decrypt { crate::decrypt::decrypt_sectors( &mut buf[..block_bytes as usize], @@ -1537,31 +1527,30 @@ impl Disc { 0, )?; } - file.seek(SeekFrom::Start(pos)) - .map_err(|e| Error::IoError { source: e })?; - file.write_all(&buf[..block_bytes as usize]) - .map_err(|e| Error::IoError { source: e })?; - map.record(pos, block_bytes, mapfile::SectorStatus::Finished) - .map_err(|e| Error::IoError { source: e })?; + + // Move the batch into the channel via fresh + // 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); + break 'outer; + } bytes_done = bytes_done.saturating_add(block_bytes); pos += block_bytes; } Err(err) if !opts.skip_on_error => { - // Caller asked us not to skip. Surface the error verbatim. let (status, sense) = extract_scsi_context(&err); - return Err(Error::DiscRead { + producer_err = Some(Error::DiscRead { sector: block_lba as u64, status: Some(status), sense, }); + break 'outer; } Err(err) => { - // === ERROR PATH — single source of truth === - // ALL errors flow through handle_read_error. New - // error class = one new arm in that function. - // Logging, counter updates, retry budgets all live - // there. The dispatch below is purely the I/O side - // of each action. read_err_count += 1; let action = read_error::handle_read_error(&err, &mut read_ctx); @@ -1570,22 +1559,18 @@ impl Disc { if pause_secs > 0 { std::thread::sleep(std::time::Duration::from_secs(pause_secs)); } - // Don't advance pos — same LBA next iteration. } read_error::ReadAction::Bisect => { - // Re-issue the failed batch as single-sector reads. - // ctx.bisecting=true so the inner failures don't - // recursively request another bisect. read_ctx.bisecting = true; let saved_batch = read_ctx.batch; read_ctx.batch = 1; + let mut bisect_aborted = false; for sector_offset in 0..block_count { if let Some(ref h) = opts.halt { if h.load(std::sync::atomic::Ordering::Relaxed) { halt_requested = true; - read_ctx.bisecting = false; - read_ctx.batch = saved_batch; - break 'outer; + bisect_aborted = true; + break; } } let sector_lba = block_lba + (sector_offset as u32); @@ -1599,54 +1584,63 @@ impl Disc { ) { Ok(_) => { read_ctx.on_success(); - file.seek(SeekFrom::Start(write_pos)) - .map_err(|e| Error::IoError { source: e })?; - file.write_all(§or_buf) - .map_err(|e| Error::IoError { source: e })?; - map.record( - write_pos, - 2048, - mapfile::SectorStatus::Finished, - ) - .map_err(|e| Error::IoError { source: e })?; + // Decrypt single sector before send. Pre-split + // bisect path silently skipped this — encrypted + // bytes were written for bisect-recovered sectors. + if opts.decrypt { + crate::decrypt::decrypt_sectors( + &mut sector_buf, + &keys, + 0, + )?; + } + if let Err(e) = send_or_abort( + &work_tx, + WorkItem::BisectGood { + pos: write_pos, + buf: Box::new(sector_buf), + }, + ) { + producer_err = Some(e); + bisect_aborted = true; + break; + } } Err(inner_err) => { - // Inner failure goes through the same handler — it'll - // see bisecting=true and won't recurse. We only honour - // the SkipBlock action here (the bisect by definition - // can't return another Bisect, and JumpAhead inside a - // single-sector retry doesn't make sense). let _ = read_error::handle_read_error( &inner_err, &mut read_ctx, ); - let zero = [0u8; 2048]; - file.seek(SeekFrom::Start(write_pos)) - .map_err(|e| Error::IoError { source: e })?; - file.write_all(&zero) - .map_err(|e| Error::IoError { source: e })?; - map.record( - write_pos, - 2048, - mapfile::SectorStatus::NonTrimmed, - ) - .map_err(|e| Error::IoError { source: e })?; + if let Err(e) = send_or_abort( + &work_tx, + WorkItem::BisectBad { pos: write_pos }, + ) { + producer_err = Some(e); + bisect_aborted = true; + break; + } } } } read_ctx.bisecting = false; read_ctx.batch = saved_batch; + if bisect_aborted { + break 'outer; + } bytes_done = bytes_done.saturating_add(block_bytes); pos += block_bytes; } read_error::ReadAction::SkipBlock { pause_secs } => { - let zero = vec![0u8; block_bytes as usize]; - file.seek(SeekFrom::Start(pos)) - .map_err(|e| Error::IoError { source: e })?; - file.write_all(&zero) - .map_err(|e| Error::IoError { source: e })?; - map.record(pos, block_bytes, mapfile::SectorStatus::NonTrimmed) - .map_err(|e| Error::IoError { source: e })?; + if let Err(e) = send_or_abort( + &work_tx, + WorkItem::SkipFill { + pos, + len: block_bytes, + }, + ) { + producer_err = Some(e); + break 'outer; + } bytes_done = bytes_done.saturating_add(block_bytes); if pause_secs > 0 { std::thread::sleep(std::time::Duration::from_secs(pause_secs)); @@ -1657,17 +1651,18 @@ impl Disc { sectors, pause_secs, } => { - // Mark the failed batch + the gap up to jump_pos NonTrimmed. - let zero_batch = vec![0u8; block_bytes as usize]; - file.seek(SeekFrom::Start(pos)) - .map_err(|e| Error::IoError { source: e })?; - file.write_all(&zero_batch) - .map_err(|e| Error::IoError { source: e })?; - map.record(pos, block_bytes, mapfile::SectorStatus::NonTrimmed) - .map_err(|e| Error::IoError { source: e })?; + if let Err(e) = send_or_abort( + &work_tx, + WorkItem::SkipFill { + pos, + len: block_bytes, + }, + ) { + producer_err = Some(e); + break 'outer; + } bytes_done = bytes_done.saturating_add(block_bytes); - // Damage-zone enter: drop to minimum read speed. if !in_damage_zone { in_damage_zone = true; reader.set_speed(0x0000); @@ -1683,22 +1678,16 @@ impl Disc { let gap_start = pos + block_bytes; let gap_bytes = jump_pos.saturating_sub(gap_start); if gap_bytes > 0 { - let zero_gap = vec![0u8; 65536]; - let mut filled: u64 = 0; - while filled < gap_bytes { - let chunk = (gap_bytes - filled).min(zero_gap.len() as u64); - file.seek(SeekFrom::Start(gap_start + filled)) - .map_err(|e| Error::IoError { source: e })?; - file.write_all(&zero_gap[..chunk as usize]) - .map_err(|e| Error::IoError { source: e })?; - filled += chunk; + if let Err(e) = send_or_abort( + &work_tx, + WorkItem::GapFill { + pos: gap_start, + len: gap_bytes, + }, + ) { + producer_err = Some(e); + break 'outer; } - map.record( - gap_start, - gap_bytes, - mapfile::SectorStatus::NonTrimmed, - ) - .map_err(|e| Error::IoError { source: e })?; bytes_done = bytes_done.saturating_add(gap_bytes); } tracing::warn!( @@ -1716,11 +1705,12 @@ impl Disc { } read_error::ReadAction::AbortPass => { let (status, sense) = extract_scsi_context(&err); - return Err(Error::DiscRead { + producer_err = Some(Error::DiscRead { sector: block_lba as u64, status: Some(status), sense, }); + break 'outer; } } } @@ -1728,45 +1718,65 @@ impl Disc { iter_count += 1; + // Drain any consumer-side stats snapshot. + if let Some(snap) = try_recv_progress(&prog_rx) { + cached_snapshot = Some(snap); + } + if iter_count - last_log_iter >= 100 { last_log_iter = iter_count; - let stats = map.stats(); - tracing::trace!( - target: "freemkv::disc", - phase = "iter_progress", - iter_count, - read_ok_count, - read_err_count, - pos, - region_end, - bytes_good = stats.bytes_good, - bytes_pending = stats.bytes_pending, - copy_elapsed_ms = copy_t0.elapsed().as_millis() as u64, - "Disc::copy inner iter" - ); + if let Some(ref snap) = cached_snapshot { + tracing::trace!( + target: "freemkv::disc", + phase = "iter_progress", + iter_count, + read_ok_count, + read_err_count, + pos, + region_end, + bytes_good = snap.stats.bytes_good, + bytes_pending = snap.stats.bytes_pending, + copy_elapsed_ms = copy_t0.elapsed().as_millis() as u64, + "Disc::sweep inner iter" + ); + } + // Throttled stats refresh request. + try_request_stats(&work_tx); } if let Some(reporter) = opts.progress { - let stats = map.stats(); - let bad_ranges = map.ranges_with(&[ - mapfile::SectorStatus::NonTrimmed, - mapfile::SectorStatus::Unreadable, - mapfile::SectorStatus::NonScraped, - mapfile::SectorStatus::NonTried, - ]); - let main_title_bad = self - .titles - .first() - .map(|t| bytes_bad_in_title(t, &bad_ranges)) - .unwrap_or(0); + // Use the latest consumer snapshot if we have + // one; otherwise synthesise a producer-side + // placeholder. On a fresh sweep, before the + // first stats round-trip lands, this means + // bytes_good ≈ bytes_done (producer's notion of + // good-so-far) and the bad-range list is empty — + // close enough for an early UI tick; the next + // real snapshot replaces it. let main_title = self.titles.first(); + let main_title_bad = match &cached_snapshot { + Some(snap) => self + .titles + .first() + .map(|t| bytes_bad_in_title(t, &snap.bad_ranges)) + .unwrap_or(0), + None => 0, + }; + let (bytes_good, bytes_unreadable, bytes_pending) = match &cached_snapshot { + Some(snap) => ( + snap.stats.bytes_good, + snap.stats.bytes_unreadable, + snap.stats.bytes_pending, + ), + None => (bytes_done, 0u64, total_bytes.saturating_sub(bytes_done)), + }; let pp = crate::progress::PassProgress { kind: crate::progress::PassKind::Sweep, work_done: pos, work_total: total_bytes, - bytes_good_total: stats.bytes_good, - bytes_unreadable_total: stats.bytes_unreadable, - bytes_pending_total: stats.bytes_pending, + bytes_good_total: bytes_good, + bytes_unreadable_total: bytes_unreadable, + bytes_pending_total: bytes_pending, bytes_total_disc: total_bytes, disc_duration_secs: main_title.map(|t| t.duration_secs), bytes_bad_in_main_title: main_title_bad, @@ -1781,32 +1791,26 @@ impl Disc { } } - tracing::debug!( - target: "freemkv::disc", - phase = "sweep_sync", - file_len = file.metadata().map(|m| m.len()).unwrap_or(0), - "sweep: calling sync_all" - ); - if let Err(e) = file.sync_all() { - if is_regular { - tracing::warn!( - target: "freemkv::disc", - phase = "sweep_sync_failed", - error = %e, - os_error = e.raw_os_error(), - error_kind = ?e.kind(), - "sweep: sync_all failed" - ); - return Err(Error::IoError { source: e }); - } - tracing::debug!( - target: "freemkv::disc", - phase = "sweep_sync_skipped", - error = %e, - "sweep: sync_all failed for non-regular file; ignoring" - ); + // 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 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 { + return Err(e); } - let stats = map.stats(); + if let Some(e) = summary.error { + return Err(e); + } + + let stats = summary.stats; tracing::debug!( target: "freemkv::disc", phase = "sweep_done", diff --git a/src/disc/sweep_pipeline.rs b/src/disc/sweep_pipeline.rs new file mode 100644 index 0000000..e0f25b6 --- /dev/null +++ b/src/disc/sweep_pipeline.rs @@ -0,0 +1,309 @@ +//! 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::Writer`] (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::Writer, + 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. + 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. + } + } + + 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::Writer`'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/writer.rs b/src/io/writer.rs index a9bf7fe..7844c0d 100644 --- a/src/io/writer.rs +++ b/src/io/writer.rs @@ -12,7 +12,7 @@ //! //! See `super::writeback::linux` for the pathology and the strategy. -use std::fs::{File, Metadata}; +use std::fs::File; use std::io::{self, Seek, SeekFrom, Write}; use super::writeback::WritebackPipeline; @@ -40,10 +40,6 @@ impl Writer { }) } - pub(crate) fn metadata(&self) -> io::Result { - self.file.metadata() - } - /// Drain in-flight writeback then issue a full fsync. Use this in /// place of `File::sync_all`. pub(crate) fn sync_all(&mut self) -> io::Result<()> {