MattJackson
90ab00ed45
Auto-fmt nit on the multi-line map.flush() expression that landed when the sweep + patch + DiscStream-FrameSource branches were merged together. No semantic change.
244 lines
10 KiB
Rust
244 lines
10 KiB
Rust
//! 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(),
|
|
})
|
|
}
|
|
}
|