pq/libfreemkv
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

0.18 round 2: refactor Disc::patch onto Pipeline + PatchSink

Browse Source 
# Conflicts:
#	src/io/mod.rs
#	src/io/pipeline.rs
This commit is contained in:
matthew
2026-05-09 10:32:51 -07:00
parent c7838f5aed 3f486b6aa8
commit b958768d74
5 changed files with 530 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/disc/mod.rs
+166 -128
View File
@@ -12,6 +12,7 @@ mod bluray;
mod dvd;
mod encrypt;
pub mod mapfile;
mod patch;
pub mod read_error;
mod sweep;
@@ -1957,7 +1958,8 @@ impl Disc {
path: &std::path::Path,
opts: &PatchOpts,
) -> Result<PatchOutcome> {
use std::io::{Seek, SeekFrom, Write};
use crate::io::pipeline::{Pipeline, WRITE_THROUGH_DEPTH};
use patch::{PatchItem, PatchSink};
const BRIDGE_DEGRADATION_PAUSE_SECS: u64 = 10;
const POST_FAILURE_PAUSE_SECS: u64 = 1;
@@ -1976,7 +1978,7 @@ impl Disc {
}
let mapfile_path = self.mapfile_for(path);
let mut map =
let map =
mapfile::Mapfile::load(&mapfile_path).map_err(|e| Error::IoError { source: e })?;
let total_bytes = map.total_size();
let keys = if opts.decrypt {
@@ -1988,11 +1990,69 @@ impl Disc {
let is_regular = std::fs::metadata(path)
.map(|m| m.file_type().is_file())
.unwrap_or(false);
// Same bounded-cache `WritebackFile` sweep uses, so patch's
// recovery writes (sparse but can be many across a damaged region)
// get the burst-flush protection on slow / NFS-backed staging.
let mut file =
crate::io::WritebackFile::open(path).map_err(|e| Error::IoError { source: e })?;
// Snapshot fields we need from the mapfile *before* it moves into
// the consumer thread: bytes_good baseline, total entries, the
// initial `bad_ranges` work list, and the start-of-patch
// diagnostic dump. The shared state (`shared`) republishes these
// throughout the pass; the consumer owns the live `Mapfile`.
let bytes_good_before = map.stats().bytes_good;
let bytes_good_start = bytes_good_before;
let initial_stats = map.stats();
let initial_entries: Vec<_> = map.entries().to_vec();
// Every retry pass acts on every non-Finished range. Including
// Unreadable means a sector that failed in pass N gets a fresh
// shot in pass N+1 — drive state evolves, the same read can
// succeed later. Each pass owns its own jumps/skips; if pass 5
// jumps over the same zone as pass 2, fine.
let mut bad_ranges = map.ranges_with(&[
mapfile::SectorStatus::NonTrimmed,
mapfile::SectorStatus::NonScraped,
mapfile::SectorStatus::Unreadable,
]);
if opts.reverse {
bad_ranges.reverse();
}
let work_total: u64 = bad_ranges.iter().map(|(_, sz)| *sz).sum();
// Spawn the consumer. The `WritebackFile` (same bounded-cache
// wrapper sweep uses, so patch's recovery writes — sparse but
// can be many across a damaged region — get the burst-flush
// protection on slow / NFS-backed staging) and the `Mapfile`
// both move into the sink. We hold an `Arc<Mutex<…>>` snapshot
// the sink republishes after every record so producer-side
// stall guards / progress callbacks can read consumer side-
// effects.
let (sink, shared) = PatchSink::new(path, map, is_regular)?;
// Why: WRITE_THROUGH_DEPTH (=1) — patch reads ONE sector per
// recovery decision and the producer's stall / damage-window
// logic checks consumer-published stats inline. Sweep's
// DEFAULT_PIPELINE_DEPTH (=4) would let several sectors of
// recovered bytes queue up between producer decisions and
// writes, which conflicts with the per-sector lockstep this
// loop was written against.
let pipe = Pipeline::<PatchItem, _>::spawn(WRITE_THROUGH_DEPTH, sink)?;
// Send a `PatchItem` and translate a `SendError` (consumer
// thread died / panicked) into a useful library error so the
// caller can propagate cleanly. Mirrors `sweep_pipeline.rs`'s
// `send_or_abort`.
let send_or_abort = |pipe: &Pipeline<PatchItem, _>, item: PatchItem| -> Result<()> {
pipe.send(item).map_err(|_| Error::IoError {
source: std::io::Error::other("patch consumer terminated unexpectedly"),
})
};
// Snapshot helper for producer-side stats reads. Holds the
// mutex briefly; we never read across operations so a fresh
// snapshot per call is fine.
let read_shared = |shared: &std::sync::Mutex<patch::SharedPatchState>| -> (
mapfile::MapStats,
Vec<(u64, u64)>,
) {
let g = shared.lock().expect("PatchSink shared state mutex poisoned");
(g.stats, g.bad_ranges.clone())
};
// Log ISO file size at patch start for write monitoring
if let Ok(metadata) = std::fs::metadata(path) {
@@ -2007,8 +2067,6 @@ impl Disc {
let block_sectors = opts.block_sectors.unwrap_or(1);
let recovery = opts.full_recovery;
let bytes_good_before = map.stats().bytes_good;
let bytes_good_start = bytes_good_before;
let mut halted = false;
let mut wedged_exit = false;
let mut blocks_attempted: u64 = 0;
@@ -2068,24 +2126,23 @@ impl Disc {
tracing::info!(
target: "freemkv::disc",
phase = "patch_mapfile_snapshot",
total_entries = map.entries().len(),
total_entries = initial_entries.len(),
bytes_good_before,
bytes_retryable = map.stats().bytes_retryable,
bytes_unreadable = map.stats().bytes_unreadable,
bytes_nontried = map.stats().bytes_nontried,
bytes_retryable = initial_stats.bytes_retryable,
bytes_unreadable = initial_stats.bytes_unreadable,
bytes_nontried = initial_stats.bytes_nontried,
"Mapfile state snapshot at patch start"
);
// Log first 10 and last 10 entries for inspection
let entries = map.entries();
if !entries.is_empty() {
if !initial_entries.is_empty() {
tracing::info!(
target: "freemkv::disc",
phase = "patch_mapfile_entries_start",
num_to_log = (entries.len().min(10)) as u32,
num_to_log = (initial_entries.len().min(10)) as u32,
"First 10 entries"
);
for entry in entries.iter().take(10) {
for entry in initial_entries.iter().take(10) {
tracing::debug!(
target: "freemkv::disc",
phase = "patch_mapfile_entry_start",
@@ -2096,14 +2153,14 @@ impl Disc {
);
}
}
if entries.len() > 10 {
if initial_entries.len() > 10 {
tracing::info!(
target: "freemkv::disc",
phase = "patch_mapfile_entries_end",
num_to_log = (entries.len().min(10)) as u32,
num_to_log = (initial_entries.len().min(10)) as u32,
"Last 10 entries"
);
for entry in entries.iter().skip(entries.len() - 10) {
for entry in initial_entries.iter().skip(initial_entries.len() - 10) {
tracing::debug!(
target: "freemkv::disc",
phase = "patch_mapfile_entry_end",
@@ -2115,21 +2172,6 @@ impl Disc {
}
}
// Every retry pass acts on every non-Finished range. Including
// Unreadable means a sector that failed in pass N gets a fresh
// shot in pass N+1 — drive state evolves, the same read can
// succeed later. Each pass owns its own jumps/skips; if pass 5
// jumps over the same zone as pass 2, fine.
let mut bad_ranges = map.ranges_with(&[
mapfile::SectorStatus::NonTrimmed,
mapfile::SectorStatus::NonScraped,
mapfile::SectorStatus::Unreadable,
]);
if opts.reverse {
bad_ranges.reverse();
}
let work_total: u64 = bad_ranges.iter().map(|(_, sz)| *sz).sum();
tracing::info!(
target: "freemkv::disc",
phase = "patch_bad_ranges",
@@ -2220,7 +2262,7 @@ impl Disc {
break;
}
let bytes_good_now = map.stats().bytes_good;
let bytes_good_now = read_shared(&shared).0.bytes_good;
if bytes_good_now > range_bytes_good {
range_bytes_good = bytes_good_now;
range_start = std::time::Instant::now();
@@ -2257,21 +2299,23 @@ impl Disc {
// patch-marked Unreadable sectors are actually readable.
let unmarked_bytes = block_end.saturating_sub(*range_pos);
if opts.reverse {
map.record(
*range_pos,
unmarked_bytes,
mapfile::SectorStatus::NonTrimmed,
)
.map_err(|e| Error::IoError { source: e })?;
send_or_abort(
&pipe,
PatchItem::NonTrimmed {
pos: *range_pos,
len: unmarked_bytes,
},
)?;
} else {
let remaining_start = *range_pos + (end - block_end);
if remaining_start < end {
map.record(
remaining_start,
end - remaining_start,
mapfile::SectorStatus::NonTrimmed,
)
.map_err(|e| Error::IoError { source: e })?;
send_or_abort(
&pipe,
PatchItem::NonTrimmed {
pos: remaining_start,
len: end - remaining_start,
},
)?;
}
}
// Continue to next range (break inner loop only)
@@ -2372,8 +2416,6 @@ impl Disc {
crate::decrypt::decrypt_sectors(&mut buf[..bytes], &keys, 0)?;
}
let write_start = std::time::Instant::now();
file.seek(SeekFrom::Start(pos))
.map_err(|e| Error::IoError { source: e })?;
tracing::debug!(
target: "freemkv::disc",
phase = "patch_write_start",
@@ -2381,8 +2423,18 @@ impl Disc {
bytes,
"Starting ISO write"
);
file.write_all(&buf[..bytes])
.map_err(|e| Error::IoError { source: e })?;
// Hand the recovered bytes off to the consumer:
// seek + write + mapfile.record(Finished) all
// happen on the consumer thread, so the producer
// can immediately move on to the next read while
// these bytes are being committed.
send_or_abort(
&pipe,
PatchItem::Recovered {
pos,
buf: buf[..bytes].to_vec(),
},
)?;
let write_duration_ms = write_start.elapsed().as_millis();
tracing::info!(
target: "freemkv::disc",
@@ -2392,21 +2444,21 @@ impl Disc {
write_duration_ms,
"ISO write succeeded"
);
let mapfile_record_start = std::time::Instant::now();
map.record(pos, block_bytes, mapfile::SectorStatus::Finished)
.map_err(|e| Error::IoError { source: e })?;
let mapfile_record_duration_ms = mapfile_record_start.elapsed().as_millis();
tracing::info!(
target: "freemkv::disc",
phase = "patch_mapfile_record_ok",
pos,
block_bytes,
mapfile_record_duration_ms,
"Mapfile record written"
"Mapfile record dispatched"
);
// Stall guard: watch bytes_good (real progress), not pos (advances on skips)
let bytes_good_now = map.stats().bytes_good;
// Stall guard: watch bytes_good (real progress),
// not pos (advances on skips). With the consumer
// running in its own thread, this read can lag
// by up to one item; the watchdog operates at
// STALL_SECS=3600 granularity so single-item lag
// is irrelevant.
let bytes_good_now = read_shared(&shared).0.bytes_good;
if bytes_good_now > bytes_good_last {
stall_start = std::time::Instant::now();
bytes_good_last = bytes_good_now;
@@ -2458,25 +2510,23 @@ impl Disc {
0,
)?;
}
file.seek(SeekFrom::Start(bt_pos))
.map_err(|e| Error::IoError { source: e })?;
file.write_all(&buf[..bt_bytes])
.map_err(|e| Error::IoError { source: e })?;
map.record(
bt_pos,
span,
mapfile::SectorStatus::Finished,
)
.map_err(|e| Error::IoError { source: e })?;
send_or_abort(
&pipe,
PatchItem::Recovered {
pos: bt_pos,
buf: buf[..bt_bytes].to_vec(),
},
)?;
}
Err(_err) => {
blocks_read_failed += 1;
map.record(
bt_pos,
span,
mapfile::SectorStatus::Unreadable,
)
.map_err(|e| Error::IoError { source: e })?;
send_or_abort(
&pipe,
PatchItem::Unreadable {
pos: bt_pos,
len: span,
},
)?;
tracing::info!(
target: "freemkv::disc",
phase = "patch_backtrack_stop",
@@ -2610,8 +2660,6 @@ impl Disc {
)?;
}
let write_start = std::time::Instant::now();
file.seek(SeekFrom::Start(pos))
.map_err(|e| Error::IoError { source: e })?;
tracing::debug!(
target: "freemkv::disc",
phase = "patch_write_start",
@@ -2619,8 +2667,13 @@ impl Disc {
bytes,
"Starting ISO write"
);
file.write_all(&buf[..bytes])
.map_err(|e| Error::IoError { source: e })?;
send_or_abort(
&pipe,
PatchItem::Recovered {
pos,
buf: buf[..bytes].to_vec(),
},
)?;
let write_duration_ms = write_start.elapsed().as_millis();
tracing::info!(
target: "freemkv::disc",
@@ -2630,22 +2683,16 @@ impl Disc {
write_duration_ms,
"ISO write succeeded"
);
let mapfile_record_start = std::time::Instant::now();
map.record(pos, block_bytes, mapfile::SectorStatus::Finished)
.map_err(|e| Error::IoError { source: e })?;
let mapfile_record_duration_ms =
mapfile_record_start.elapsed().as_millis();
tracing::info!(
target: "freemkv::disc",
phase = "patch_mapfile_record_ok",
pos,
block_bytes,
mapfile_record_duration_ms,
"Mapfile record written"
"Mapfile record dispatched"
);
// Stall guard after successful retry
let bytes_good_now = map.stats().bytes_good;
let bytes_good_now = read_shared(&shared).0.bytes_good;
if bytes_good_now > bytes_good_last {
stall_start = std::time::Instant::now();
bytes_good_last = bytes_good_now;
@@ -2665,8 +2712,13 @@ impl Disc {
}
// All retries exhausted - mark as Unreadable
map.record(pos, block_bytes, mapfile::SectorStatus::Unreadable)
.map_err(|e| Error::IoError { source: e })?;
send_or_abort(
&pipe,
PatchItem::Unreadable {
pos,
len: block_bytes,
},
)?;
damage_window.push(false);
if damage_window.len() > PASSN_DAMAGE_WINDOW {
@@ -2674,7 +2726,7 @@ impl Disc {
}
// Stall guard: check on failures too, not just successes
let bytes_good_now = map.stats().bytes_good;
let bytes_good_now = read_shared(&shared).0.bytes_good;
if bytes_good_now > bytes_good_last {
stall_start = std::time::Instant::now();
bytes_good_last = bytes_good_now;
@@ -2779,19 +2831,25 @@ impl Disc {
}
}
// Log mapfile record for Unreadable status
let mapfile_record_start = std::time::Instant::now();
map.record(pos, block_bytes, mapfile::SectorStatus::Unreadable)
.map_err(|e| Error::IoError { source: e })?;
let mapfile_record_duration_ms = mapfile_record_start.elapsed().as_millis();
// Redundant second Unreadable mark — preserved
// bit-for-bit from the pre-split loop (`record`
// is idempotent for same-status replacement of
// the same range). Routes through the consumer
// like every other state change.
send_or_abort(
&pipe,
PatchItem::Unreadable {
pos,
len: block_bytes,
},
)?;
tracing::info!(
target: "freemkv::disc",
phase = "patch_mapfile_record_unreadable",
pos,
block_bytes,
consecutive_failures,
mapfile_record_duration_ms,
"Mapfile record written as Unreadable"
"Mapfile record dispatched as Unreadable"
);
let pause_secs = if err.is_bridge_degradation() {
@@ -2905,7 +2963,7 @@ impl Disc {
work_done = work_done.saturating_add(block_bytes);
if let Some(reporter) = opts.progress {
let s = map.stats();
let (s, bad_ranges_now) = read_shared(&shared);
let kind = if block_sectors == 1 {
crate::progress::PassKind::Scrape {
reverse: opts.reverse,
@@ -2915,16 +2973,10 @@ impl Disc {
reverse: opts.reverse,
}
};
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))
.map(|t| bytes_bad_in_title(t, &bad_ranges_now))
.unwrap_or(0);
let main_title = self.titles.first();
let pp = crate::progress::PassProgress {
@@ -2948,26 +3000,13 @@ impl Disc {
}
}
if let Err(e) = file.sync_all() {
if is_regular {
tracing::warn!(
target: "freemkv::disc",
phase = "patch_sync_failed",
error = %e,
os_error = e.raw_os_error(),
error_kind = ?e.kind(),
"patch: sync_all failed"
);
return Err(Error::IoError { source: e });
}
tracing::debug!(
target: "freemkv::disc",
phase = "patch_sync_skipped",
error = %e,
"patch: sync_all failed for non-regular file; ignoring"
);
}
map.flush().map_err(|e| Error::IoError { source: e })?;
// Drain the consumer thread: drop tx, wait for `close` to run
// sync_all + mapfile.flush, then take the final stats from the
// sink's summary. `close` failing on a regular-file sync_all is
// surfaced here as `Error::IoError`, matching pre-split
// behaviour.
let summary = pipe.finish()?;
let stats = summary.stats;
// Log final ISO file size for write verification
if let Ok(metadata) = std::fs::metadata(path) {
@@ -2975,12 +3014,11 @@ impl Disc {
target: "freemkv::disc",
phase = "patch_iso_size_end",
iso_bytes = metadata.len(),
bytes_recovered = map.stats().bytes_good.saturating_sub(bytes_good_before),
bytes_recovered = stats.bytes_good.saturating_sub(bytes_good_before),
"ISO file size at patch end"
);
}
let stats = map.stats();
tracing::info!(
target: "freemkv::disc",
phase = "patch_done",
src/disc/patch.rs
+241
View File
@@ -0,0 +1,241 @@
//! Producer / consumer split for `Disc::patch`.
//!
//! Background: pre-0.18 patch ran strictly serial — single-sector
//! recovery read → seek + write recovered bytes → mapfile.record →
//! next iteration. The drive sat idle while the previous block's
//! recovered bytes were committed. On a damaged disc with many bad
//! sectors that adds up: per-sector write + mapfile.record costs a
//! handful of milliseconds each, which the drive could be using to
//! issue the next per-sector retry.
//!
//! This module decouples them. A consumer thread owns the
//! [`crate::io::WritebackFile`] (the ISO file) and the
//! [`super::mapfile::Mapfile`]. The producer thread (`Disc::patch`)
//! keeps the [`crate::sector::SectorReader`], the wedge / damage-window
//! state, the per-range watchdog, decrypt — so what enters the channel
//! is already-clean cleartext bytes (or an "Unreadable" terminal mark).
//!
//! Producer and consumer run concurrently; the channel uses
//! [`crate::io::pipeline::WRITE_THROUGH_DEPTH`] (=1) so back-pressure
//! kicks in immediately. We want the drive's per-sector retry budget
//! to stay in lockstep with the writer — sweep's `DEFAULT_PIPELINE_DEPTH`
//! (4) would let several sectors of recovered bytes queue up between
//! the producer's retry decisions and the writer, and patch's recovery
//! loop reads stats (`bytes_good`, range progress) inline to drive its
//! skip / wedge decisions. WRITE_THROUGH_DEPTH gives "read N+1 while
//! writing N", no further pipelining — exactly the model the producer
//! logic was written against.
//!
//! Correctness invariants preserved:
//! - Mapfile is single-writer (consumer-only). No locking on it.
//! - All recovery state (damage window, consecutive_failures, skip
//! escalation, range watchdog) 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. The threading primitive only overlaps the
//! *write* with the *next read*; the per-sector single-shot read
//! budget that the bridge wedge concern was originally about is
//! untouched.
//!
//! Per-range watchdog (`MAX_RANGE_SECS` / `RANGE_BUDGET_CAP_SECS`)
//! checks `bytes_good` for forward progress. With work in flight on
//! the consumer, the producer would otherwise see stale values; the
//! sink publishes a [`SharedPatchState`] snapshot after every record
//! so the producer's stall guards observe consumer side-effects with
//! at most one item of lag (which is fine — the watchdog uses minute-
//! scale budgets, not single-record latency).
use std::io::{Seek, SeekFrom, Write};
use std::sync::{Arc, Mutex};
use crate::error::{Error, Result};
use crate::io::pipeline::{Flow, Sink};
use super::mapfile::{MapStats, Mapfile, SectorStatus};
/// Item the producer hands to the patch consumer. One per per-sector
/// recovery decision.
pub(super) enum PatchItem {
/// Sector / small batch successfully recovered (and decrypted on the
/// producer side if `opts.decrypt` was set). Consumer seeks to
/// `pos`, writes `buf`, records the range as `Finished`.
Recovered { pos: u64, buf: Vec<u8> },
/// Producer exhausted retries on `[pos, pos+len)`. Consumer records
/// the range as `Unreadable`. No file write — the existing zero-fill
/// from sweep is preserved in place.
Unreadable { pos: u64, len: u64 },
/// Producer hit the per-range skip limit and is leaving the
/// remaining bytes as `NonTrimmed` for a future pass. CRITICAL:
/// this is not the same as `Unreadable` — sectors we never tried
/// stay hopeful. (See the comment at the skip-limit branch in
/// `Disc::patch`: ~36% of patch-marked Unreadable sectors are
/// actually readable on a later pass.) No file write.
NonTrimmed { pos: u64, len: u64 },
}
/// Mapfile snapshot the sink republishes after every record so the
/// producer can drive its stall / progress logic without holding the
/// mapfile lock for long. `bad_ranges` mirrors what
/// `Mapfile::ranges_with(&[NonTrimmed, Unreadable, NonScraped, NonTried])`
/// would return — same set the pre-split patch loop computed inline
/// for the progress callback.
pub(super) struct SharedPatchState {
pub stats: MapStats,
pub bad_ranges: Vec<(u64, u64)>,
}
impl SharedPatchState {
fn from_map(map: &Mapfile) -> Self {
Self {
stats: map.stats(),
bad_ranges: map.ranges_with(&[
SectorStatus::NonTrimmed,
SectorStatus::Unreadable,
SectorStatus::NonScraped,
SectorStatus::NonTried,
]),
}
}
}
/// Final summary returned by [`Sink::close`] when the consumer drains
/// cleanly. Mirrors what the pre-split patch loop computed at the end
/// of the function — final mapfile stats plus whether `sync_all`
/// failed on a regular file (the only kind of fsync error patch ever
/// surfaced; `/dev/null` and pipes always fail `sync_all`, that's not
/// a real error).
pub(super) struct PatchSummary {
pub stats: MapStats,
}
/// Consumer-side of the patch pipeline. Owns the ISO writeback file
/// and the mapfile; publishes a shared snapshot after every record so
/// the producer can read `bytes_good` for stall detection and
/// progress reporting.
pub(super) struct PatchSink {
file: crate::io::WritebackFile,
map: Mapfile,
/// Whether the output is a regular file (so a `sync_all` failure
/// is real). `/dev/null` etc. always fail `sync_all`; ignore those.
is_regular: bool,
/// Snapshot the producer reads. Updated after every successful
/// `record()` call. `Mutex` rather than separate atomics because
/// the producer wants stats + bad_ranges as a coherent pair.
shared: Arc<Mutex<SharedPatchState>>,
}
impl PatchSink {
/// Open `path` as a [`crate::io::WritebackFile`] and pair it with
/// `map` for the consumer. The producer holds onto the returned
/// `Arc<Mutex<SharedPatchState>>` so it can poll mapfile state
/// while the consumer is mutating it.
pub(super) fn new(
path: &std::path::Path,
map: Mapfile,
is_regular: bool,
) -> Result<(Self, Arc<Mutex<SharedPatchState>>)> {
let file = crate::io::WritebackFile::open(path).map_err(|e| Error::IoError { source: e })?;
let shared = Arc::new(Mutex::new(SharedPatchState::from_map(&map)));
let shared_clone = shared.clone();
Ok((
Self {
file,
map,
is_regular,
shared,
},
shared_clone,
))
}
fn republish(&self) {
// Best-effort lock — only the producer reads, only the consumer
// writes; contention is single-acquire so the lock is never
// poisoned in practice. If it ever did get poisoned we'd want
// the underlying error surfaced rather than silently swallowed,
// so we propagate the poison panic. (Same posture as
// `sweep_pipeline.rs` — it never recovers from a poisoned
// mutex either.)
let mut guard = self.shared.lock().expect("PatchSink shared state mutex poisoned");
*guard = SharedPatchState::from_map(&self.map);
}
}
impl Sink<PatchItem> for PatchSink {
type Output = PatchSummary;
fn apply(&mut self, item: PatchItem) -> std::result::Result<Flow, Error> {
match item {
PatchItem::Recovered { pos, buf } => {
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 })?;
}
PatchItem::Unreadable { pos, len } => {
self.map
.record(pos, len, SectorStatus::Unreadable)
.map_err(|e| Error::IoError { source: e })?;
}
PatchItem::NonTrimmed { pos, len } => {
self.map
.record(pos, len, SectorStatus::NonTrimmed)
.map_err(|e| Error::IoError { source: e })?;
}
}
self.republish();
Ok(Flow::Continue)
}
fn close(mut self) -> std::result::Result<Self::Output, Error> {
// Drain in-flight writeback then issue a full fsync. A failure
// here matters only on regular files — pipes / `/dev/null` etc.
// always fail `sync_all`.
if let Err(e) = self.file.sync_all() {
if self.is_regular {
tracing::warn!(
target: "freemkv::disc",
phase = "patch_sync_failed",
error = %e,
os_error = e.raw_os_error(),
error_kind = ?e.kind(),
"patch: sync_all failed"
);
return Err(Error::IoError { source: e });
}
tracing::debug!(
target: "freemkv::disc",
phase = "patch_sync_skipped",
error = %e,
"patch: sync_all failed for non-regular file; ignoring"
);
}
self.map
.flush()
.map_err(|e| Error::IoError { source: e })?;
// Final republish so anyone reading the shared snapshot after
// `Pipeline::finish` sees the post-flush state. (The producer
// already has its own copy of the final `MapStats` in the
// returned `PatchSummary`, but the snapshot is part of the
// public-ish contract of the consumer: it stays current
// through close.)
self.republish();
Ok(PatchSummary {
stats: self.map.stats(),
})
}
}
src/io/mod.rs
+5 -5
View File
@@ -21,11 +21,11 @@ pub mod pipeline;
pub(crate) use writeback_file::WritebackFile;
// 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`.
// Re-exports for the 0.18 redesign. Sweep + patch are both wired up
// (disc/sweep.rs, disc/patch.rs); mux migrates separately in autorip.
// `WRITE_THROUGH_DEPTH` is patch-specific and has no other 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::WRITE_THROUGH_DEPTH;
pub use pipeline::{DEFAULT_PIPELINE_DEPTH, Flow, Pipeline, Sink};
src/io/pipeline.rs
+7 -4
View File
@@ -29,10 +29,13 @@
//!
//! ## Dead-code suppression
//!
//! `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.
//! Both `Disc::sweep` (`DEFAULT_PIPELINE_DEPTH`, `spawn_named`,
//! `try_send`) and `Disc::patch` (`WRITE_THROUGH_DEPTH`, `Flow::Stop`)
//! are now in-tree callers (0.18 round 2). The targeted `#[allow]`
//! markers below cover the corners of the API not exercised by either
//! call site (e.g. `Pipeline::spawn` without a name; `Flow::Stop`
//! return path from sweep's sink — sweep always consumes the producer's
//! full work-list). Drop the allows once mux migrates.
use std::io;
use std::sync::mpsc::{SyncSender, sync_channel};
tests/pass_n_size_aware_skip.rs
+111
View File
@@ -286,3 +286,114 @@ fn patch_recovers_multiple_good_middles() {
pr.bytes_total,
);
}
/// 0.18 Pass N pipeline split: exercises the new producer/consumer
/// path end-to-end on a synthetic patterned reader. Bad range layout
/// is small (5 bad LBAs surrounded by good middle) so the producer
/// emits a mix of `Recovered` and `Unreadable` items and the consumer
/// thread must apply both kinds. Verifies:
///
/// - `bytes_good` advances (good sectors flow producer→consumer→file
/// →mapfile with the data preserved).
/// - The recovered LBAs end up Finished; the bad LBAs end up Unreadable.
/// - Bytes written at the recovered offsets match what the producer
/// read from the patterned source (proves the channel hand-off
/// didn't drop or reorder buffers, and the consumer's seek+write
/// landed at the right offsets).
#[test]
fn patch_pipeline_split_recovers_and_records_correctly() {
let capacity_sectors: u32 = 512;
let total_bytes: u64 = capacity_sectors as u64 * SECTOR_SIZE as u64;
// Layout: LBAs 200-204 inclusive are bad (5 sectors), 205-249 good.
// The pre-existing range is LBAs 200-249 NonTrimmed (100 KB).
let mut bad_lbas = HashSet::new();
for lba in 200..205 {
bad_lbas.insert(lba);
}
let (mut reader, _trace) = PatternedSectorReader::new(capacity_sectors, bad_lbas.clone());
let disc = synthetic_disc(capacity_sectors);
let tmp = tempfile::NamedTempFile::new().unwrap();
let iso_path = tmp.path().to_path_buf();
drop(tmp);
let finished = [
(0, 200 * 2048),
(250 * 2048, (capacity_sectors as u64 - 250) * 2048),
];
let nontrimmed = [(200 * 2048, 50 * 2048)];
prep_iso_and_mapfile(&iso_path, total_bytes, &finished, &nontrimmed);
let opts = CopyOptions {
decrypt: false,
multipass: true,
..Default::default()
};
let pr = disc
.copy(&mut reader, &iso_path, &opts)
.expect("copy returns Ok");
// Bytes_good_total should advance — the good LBAs in the bad range
// (205-249, 45 sectors) are all reachable via per-sector retry.
// Initial bytes_good = 200 * 2048 + (512-250) * 2048 = 462 sectors.
// After patch, bytes_good should be ≥ 462 + 45 = 507 sectors worth.
let initial_good_sectors: u64 = 200 + (capacity_sectors as u64 - 250);
let min_expected_good_bytes = (initial_good_sectors + 30) * 2048;
assert!(
pr.bytes_good >= min_expected_good_bytes,
"patch should have recovered most good LBAs in the bad range via the pipeline. \
bytes_good={} (expected ≥ {}); bytes_total={}",
pr.bytes_good,
min_expected_good_bytes,
pr.bytes_total,
);
// Verify the mapfile records: every good LBA is Finished, every
// bad LBA is either Unreadable or NonTrimmed (not Finished).
let map_path = libfreemkv::disc::mapfile_path_for(&iso_path);
let map = Mapfile::load(&map_path).unwrap();
let finished_ranges = map.ranges_with(&[SectorStatus::Finished]);
let in_finished = |lba: u32| -> bool {
let pos = lba as u64 * 2048;
finished_ranges
.iter()
.any(|&(p, sz)| pos >= p && pos < p + sz)
};
for lba in 205..250 {
assert!(
in_finished(lba),
"good LBA {lba} should be Finished after pipeline patch run"
);
}
for lba in 200..205 {
assert!(
!in_finished(lba),
"bad LBA {lba} should NOT be Finished after pipeline patch run"
);
}
// Verify the consumer wrote the producer's bytes at the right
// offsets. PatternedSectorReader fills each sector with `(lba & 0xff)
// as u8` — picking LBA 220 (well inside the recovered region) gives
// a clean signature byte to check.
use std::io::{Read, Seek, SeekFrom};
let mut iso = std::fs::File::open(&iso_path).unwrap();
iso.seek(SeekFrom::Start(220 * 2048)).unwrap();
let mut sector = [0u8; 2048];
iso.read_exact(&mut sector).unwrap();
let expected_byte = (220u32 & 0xff) as u8;
let _ = std::fs::remove_file(&iso_path);
let _ = std::fs::remove_file(&map_path);
assert!(
sector.iter().all(|&b| b == expected_byte),
"consumer should have written PatternedSectorReader's pattern \
(byte {expected_byte:#x} for LBA 220) to the recovered offset; \
got first 8 bytes = {:?}",
&sector[..8]
);
}