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

0.18 round 2: refactor Disc::sweep onto Pipeline + SweepSink (delete sweep_pipeline.rs)

Browse Source
This commit is contained in:
matthew
2026-05-09 10:32:05 -07:00
parent 59dd19aaaa 24140dfa55
commit c7838f5aed
5 changed files with 408 additions and 393 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/disc/mod.rs
+108 -56
View File
@@ -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<CopyResult> {
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<WorkItem, sweep::ConsumerSummary> =
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);
}
}
src/disc/sweep.rs
+239
View File
@@ -0,0 +1,239 @@
//! `Disc::sweep`'s consumer-side `Sink<WorkItem>`.
//!
//! 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<u8> },
/// 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<ProgressSnapshot>) -> Option<ProgressSnapshot> {
let mut latest = None;
while let Ok(snap) = rx.try_recv() {
latest = Some(snap);
}
latest
}
/// `Sink<WorkItem>` 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<ProgressSnapshot>,
/// 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<ProgressSnapshot>) {
let (prog_tx, prog_rx) = sync_channel::<ProgressSnapshot>(1);
let sink = SweepSink {
file,
map,
is_regular,
prog_tx,
zero: Box::new([0u8; ZERO_CHUNK]),
};
(sink, prog_rx)
}
}
impl Sink<WorkItem> for SweepSink {
type Output = ConsumerSummary;
fn apply(&mut self, item: WorkItem) -> Result<Flow, Error> {
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<Self::Output, Error> {
// 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(),
})
}
}
src/disc/sweep_pipeline.rs
-313
View File
@@ -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<u8> },
/// 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<ConsumerSummary>` 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<Error>,
}
/// 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<WorkItem>,
Receiver<ProgressSnapshot>,
JoinHandle<ConsumerSummary>,
) {
let (work_tx, work_rx) = sync_channel::<WorkItem>(CHANNEL_DEPTH);
let (prog_tx, prog_rx) = sync_channel::<ProgressSnapshot>(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<WorkItem>, 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<WorkItem>) {
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<ProgressSnapshot>) -> Option<ProgressSnapshot> {
let mut latest = None;
while let Ok(snap) = rx.try_recv() {
latest = Some(snap);
}
latest
}
fn consumer_loop(
mut inputs: ConsumerInputs,
work_rx: Receiver<WorkItem>,
prog_tx: SyncSender<ProgressSnapshot>,
) -> ConsumerSummary {
let zero = [0u8; ZERO_CHUNK];
let mut first_error: Option<Error> = 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<ProgressSnapshot>,
) -> Result<bool> {
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)
}
src/io/mod.rs
+7 -5
View File
@@ -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};
src/io/pipeline.rs
+54 -19
View File
@@ -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 `freemkv-private/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<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
/// [`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<S: Sink<I, Output = R>>(depth: usize, sink: S) -> Result<Self, Error> {
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<S: Sink<I, Output = R>>(
name: &str,
depth: usize,
sink: S,
) -> Result<Self, Error> {
let (tx, rx) = sync_channel::<I>(depth);
let handle = thread::Builder::new()
.name("freemkv-pipeline-consumer".into())
.name(name.into())
.spawn(move || -> Result<R, Error> {
let mut sink = sink;
let mut first_err: Option<Error> = None;
@@ -162,6 +188,15 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
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<I>> {
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 —