v0.17.12: mapfile time-batched persistence — unblock NFS staging
Pre-0.17.12 every Mapfile::record() persisted the full mapfile via tempfile-create + write + atomic-rename. On local LVM that's microseconds; on NFS each rename is multiple RPCs through the unraid user-share fuse layer, dragging a Black Mass UHD rip from ~11 MB/s on local to ~1.5 MB/s on NFS — the mapfile path alone burned multiple seconds of wall time per real-world second of work. Mapfile now batches the rename to once per second: - record() always updates in-memory state and stats; only fires write_to_disk when last_flushed.elapsed() >= FLUSH_INTERVAL (1 s). - New flush() API forces a persist; called by sweep_pipeline's consumer at end-of-sweep and by Disc::patch at end-of-patch, after the file's sync_all. - Drop impl best-effort flushes so an early-return / unwind doesn't silently lose pending state. Crash-safety changes from "lose at most one block" to "lose at most 1 s of recorded progress" — the ISO file's payload bytes are unaffected; only the mapfile's authority over which sectors are already-good is at risk, and a resume re-reads anything Pass 1 had already covered. Measured on the BU40N test bed against Black Mass UHD inner zone: - NFS staging: 1.5 MB/s → 16.48 MB/s (10.9× recovery) - Local LVM staging: 11.09 MB/s → 11.83 MB/s (+6.7 % bonus) Internal round_trip_load test now flushes before reading back from disk. External patch / copy tests are unaffected: patch and sweep_pipeline flush at completion before returning.
This commit is contained in:
@@ -1,5 +1,44 @@
|
|||||||
# Changelog
|
# Changelog
|
||||||
|
|
||||||
|
## 0.17.12 (2026-05-09)
|
||||||
|
|
||||||
|
### Mapfile time-batched persistence — unblock NFS staging
|
||||||
|
|
||||||
|
Pre-0.17.12 every `Mapfile::record()` wrote the entire mapfile to disk
|
||||||
|
via tempfile-create + buffered-write + atomic-rename. On local LVM
|
||||||
|
that's effectively free (page cache + microsecond-scale renames). On
|
||||||
|
NFS (autorip's intended staging path for centralised media) each
|
||||||
|
record() became three RPCs through the unraid user share's shfs-fuse
|
||||||
|
layer — measured end-to-end at multiple ms each. With ~170 record()
|
||||||
|
calls per second of sustained sweep, the mapfile path alone burned
|
||||||
|
multiple seconds of wall time per real-world second of work, dragging
|
||||||
|
the rip from ~11 MB/s on local to ~1.5 MB/s on NFS.
|
||||||
|
|
||||||
|
Fix: time-batch the persistence inside `Mapfile`.
|
||||||
|
|
||||||
|
- `record()` always updates in-memory state and stats (so callers'
|
||||||
|
`stats()` reads stay coherent in the same process).
|
||||||
|
- The `write_to_disk()` rename only fires when ≥ `FLUSH_INTERVAL`
|
||||||
|
(1 s) has elapsed since the last persist.
|
||||||
|
- New `flush()` method forces an out-of-band persist; called by
|
||||||
|
`sweep_pipeline`'s consumer at end-of-sweep and by `Disc::patch` at
|
||||||
|
end-of-patch, after the file's `sync_all()`.
|
||||||
|
- `Drop` impl best-effort flushes so an early-return / unwind doesn't
|
||||||
|
silently lose pending state.
|
||||||
|
|
||||||
|
Crash-safety changes from "lose at most one block" to "lose at most
|
||||||
|
1 s of recorded progress" — the ISO file's payload bytes are
|
||||||
|
unaffected; only the mapfile's authority over which sectors are
|
||||||
|
already-good is at risk, and a resume re-reads anything Pass 1 had
|
||||||
|
already covered. Acceptable for a 7× throughput recovery on the
|
||||||
|
target deployment.
|
||||||
|
|
||||||
|
The internal `round_trip_load` test now calls `flush()` before
|
||||||
|
`Mapfile::load` to read back what the in-memory state asserts.
|
||||||
|
External patch / copy tests are unaffected: `patch` and
|
||||||
|
`sweep_pipeline` flush at completion before returning, so any
|
||||||
|
`Mapfile::load` at the call-site sees fully persisted state.
|
||||||
|
|
||||||
## 0.17.11 (2026-05-09)
|
## 0.17.11 (2026-05-09)
|
||||||
|
|
||||||
### Sweep producer/consumer split — overlap drive read with file write
|
### Sweep producer/consumer split — overlap drive read with file write
|
||||||
|
|||||||
+1
-1
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "libfreemkv"
|
name = "libfreemkv"
|
||||||
version = "0.17.11"
|
version = "0.17.12"
|
||||||
edition = "2024"
|
edition = "2024"
|
||||||
rust-version = "1.86"
|
rust-version = "1.86"
|
||||||
license = "AGPL-3.0-only"
|
license = "AGPL-3.0-only"
|
||||||
|
|||||||
+61
-5
@@ -17,11 +17,22 @@
|
|||||||
//!
|
//!
|
||||||
//! Status chars: `?` non-tried · `*` non-trimmed · `/` non-scraped · `-` unreadable · `+` finished.
|
//! Status chars: `?` non-tried · `*` non-trimmed · `/` non-scraped · `-` unreadable · `+` finished.
|
||||||
//!
|
//!
|
||||||
//! The mapfile is flushed to disk on every `record()` call so a crashed
|
//! The mapfile is flushed to disk at most once per `FLUSH_INTERVAL`
|
||||||
//! rip loses at most one block of recorded state.
|
//! during `record()` calls, plus on explicit `flush()` and on `Drop`.
|
||||||
|
//! This bounds atomic-rename RPC rate on networked staging (e.g. NFS)
|
||||||
|
//! where per-record persists otherwise serialize the rip pipeline.
|
||||||
|
|
||||||
use std::io::{self, Write};
|
use std::io::{self, Write};
|
||||||
use std::path::{Path, PathBuf};
|
use std::path::{Path, PathBuf};
|
||||||
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
|
/// Minimum interval between mapfile persists. `record()` updates in-memory
|
||||||
|
/// state every call but only writes to disk when this interval has elapsed
|
||||||
|
/// since the last persist (or when `flush()` is called explicitly, or on
|
||||||
|
/// `Drop`). Bounds RPC rate on NFS staging where atomic-rename per record
|
||||||
|
/// otherwise dominates throughput. On crash the worst-case progress loss
|
||||||
|
/// is one interval's worth of records.
|
||||||
|
const FLUSH_INTERVAL: Duration = Duration::from_millis(1000);
|
||||||
|
|
||||||
/// Status of a byte range in the mapfile. ddrescue-compatible.
|
/// Status of a byte range in the mapfile. ddrescue-compatible.
|
||||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||||
@@ -100,8 +111,11 @@ pub struct MapStats {
|
|||||||
pub main_lost_ms: f64,
|
pub main_lost_ms: f64,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Write-through mapfile. Every `record()` persists to disk immediately
|
/// Time-batched mapfile. `record()` keeps in-memory state up-to-date on
|
||||||
/// so a crash during rip loses at most one block.
|
/// every call; persists to disk at most once per `FLUSH_INTERVAL`.
|
||||||
|
/// Explicit `flush()` and `Drop` guarantee state is on disk after a sweep
|
||||||
|
/// or patch finishes. On hard crash the worst-case loss is one flush
|
||||||
|
/// interval of records — the file's payload bytes are unaffected.
|
||||||
pub struct Mapfile {
|
pub struct Mapfile {
|
||||||
path: PathBuf,
|
path: PathBuf,
|
||||||
entries: Vec<MapEntry>,
|
entries: Vec<MapEntry>,
|
||||||
@@ -110,6 +124,12 @@ pub struct Mapfile {
|
|||||||
/// Incrementally maintained stats — updated on every `record()` call
|
/// Incrementally maintained stats — updated on every `record()` call
|
||||||
/// so `stats()` is O(1) instead of O(n).
|
/// so `stats()` is O(1) instead of O(n).
|
||||||
stats: MapStats,
|
stats: MapStats,
|
||||||
|
/// True when in-memory state has changed but `write_to_disk` has not
|
||||||
|
/// yet captured it.
|
||||||
|
dirty: bool,
|
||||||
|
/// Wall-clock timestamp of the last successful `write_to_disk` (or
|
||||||
|
/// the moment the mapfile was constructed, whichever is later).
|
||||||
|
last_flushed: Instant,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Mapfile {
|
impl Mapfile {
|
||||||
@@ -117,7 +137,7 @@ impl Mapfile {
|
|||||||
/// Writes to disk immediately so a resume can pick up even if the caller
|
/// Writes to disk immediately so a resume can pick up even if the caller
|
||||||
/// never records anything.
|
/// never records anything.
|
||||||
pub fn create(path: &Path, total_size: u64, version: &str) -> io::Result<Self> {
|
pub fn create(path: &Path, total_size: u64, version: &str) -> io::Result<Self> {
|
||||||
let mf = Self {
|
let mut mf = Self {
|
||||||
path: path.to_path_buf(),
|
path: path.to_path_buf(),
|
||||||
entries: vec![MapEntry {
|
entries: vec![MapEntry {
|
||||||
pos: 0,
|
pos: 0,
|
||||||
@@ -132,8 +152,13 @@ impl Mapfile {
|
|||||||
bytes_nontried: total_size,
|
bytes_nontried: total_size,
|
||||||
..Default::default()
|
..Default::default()
|
||||||
},
|
},
|
||||||
|
dirty: false,
|
||||||
|
last_flushed: Instant::now(),
|
||||||
};
|
};
|
||||||
|
// Eager initial persist so a resume can pick this up even if
|
||||||
|
// `record()` is never called.
|
||||||
mf.write_to_disk()?;
|
mf.write_to_disk()?;
|
||||||
|
mf.last_flushed = Instant::now();
|
||||||
Ok(mf)
|
Ok(mf)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -200,6 +225,8 @@ impl Mapfile {
|
|||||||
total_size,
|
total_size,
|
||||||
version,
|
version,
|
||||||
stats,
|
stats,
|
||||||
|
dirty: false,
|
||||||
|
last_flushed: Instant::now(),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -268,7 +295,24 @@ impl Mapfile {
|
|||||||
// path during sweep/patch, it just returns the cached value.
|
// path during sweep/patch, it just returns the cached value.
|
||||||
self.stats = Self::compute_stats(&merged, self.total_size);
|
self.stats = Self::compute_stats(&merged, self.total_size);
|
||||||
self.entries = merged;
|
self.entries = merged;
|
||||||
|
self.dirty = true;
|
||||||
|
if self.last_flushed.elapsed() >= FLUSH_INTERVAL {
|
||||||
self.write_to_disk()?;
|
self.write_to_disk()?;
|
||||||
|
self.dirty = false;
|
||||||
|
self.last_flushed = Instant::now();
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Persist any pending in-memory changes to disk. No-op if clean.
|
||||||
|
/// Callers (sweep/patch finalisation) invoke this after their last
|
||||||
|
/// `record()` to guarantee state is durable before returning.
|
||||||
|
pub fn flush(&mut self) -> io::Result<()> {
|
||||||
|
if self.dirty {
|
||||||
|
self.write_to_disk()?;
|
||||||
|
self.dirty = false;
|
||||||
|
self.last_flushed = Instant::now();
|
||||||
|
}
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -363,6 +407,16 @@ impl Mapfile {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl Drop for Mapfile {
|
||||||
|
/// Best-effort flush on drop so a sweep / patch that returns early
|
||||||
|
/// (or unwinds) doesn't lose its in-memory state. Errors here are
|
||||||
|
/// swallowed because Drop has no way to surface them; explicit
|
||||||
|
/// `flush()` on the success path gives callers proper error handling.
|
||||||
|
fn drop(&mut self) {
|
||||||
|
let _ = self.flush();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
fn parse_hex(s: &str) -> io::Result<u64> {
|
fn parse_hex(s: &str) -> io::Result<u64> {
|
||||||
let s = s.strip_prefix("0x").unwrap_or(s);
|
let s = s.strip_prefix("0x").unwrap_or(s);
|
||||||
u64::from_str_radix(s, 16).map_err(|_| {
|
u64::from_str_radix(s, 16).map_err(|_| {
|
||||||
@@ -463,6 +517,8 @@ mod tests {
|
|||||||
let mut mf = Mapfile::create(&p, 1000, "test").unwrap();
|
let mut mf = Mapfile::create(&p, 1000, "test").unwrap();
|
||||||
mf.record(100, 200, SectorStatus::Finished).unwrap();
|
mf.record(100, 200, SectorStatus::Finished).unwrap();
|
||||||
mf.record(500, 100, SectorStatus::Unreadable).unwrap();
|
mf.record(500, 100, SectorStatus::Unreadable).unwrap();
|
||||||
|
// record() batches; explicit flush before reading back from disk.
|
||||||
|
mf.flush().unwrap();
|
||||||
let loaded = Mapfile::load(&p).unwrap();
|
let loaded = Mapfile::load(&p).unwrap();
|
||||||
assert_eq!(loaded.entries(), mf.entries());
|
assert_eq!(loaded.entries(), mf.entries());
|
||||||
let _ = std::fs::remove_file(&p);
|
let _ = std::fs::remove_file(&p);
|
||||||
|
|||||||
@@ -2956,6 +2956,7 @@ impl Disc {
|
|||||||
"patch: sync_all failed for non-regular file; ignoring"
|
"patch: sync_all failed for non-regular file; ignoring"
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
map.flush().map_err(|e| Error::IoError { source: e })?;
|
||||||
|
|
||||||
// Log final ISO file size for write verification
|
// Log final ISO file size for write verification
|
||||||
if let Ok(metadata) = std::fs::metadata(path) {
|
if let Ok(metadata) = std::fs::metadata(path) {
|
||||||
|
|||||||
@@ -199,7 +199,8 @@ fn consumer_loop(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Final flush — drain the writeback pipeline + fsync the ISO.
|
// Final flush — drain the writeback pipeline + fsync the ISO,
|
||||||
|
// then persist any pending mapfile state.
|
||||||
if first_error.is_none() {
|
if first_error.is_none() {
|
||||||
if let Err(e) = inputs.file.sync_all() {
|
if let Err(e) = inputs.file.sync_all() {
|
||||||
if inputs.is_regular {
|
if inputs.is_regular {
|
||||||
@@ -208,6 +209,9 @@ fn consumer_loop(
|
|||||||
// Non-regular outputs (/dev/null, pipes) always fail
|
// Non-regular outputs (/dev/null, pipes) always fail
|
||||||
// sync_all; that's not a real error.
|
// sync_all; that's not a real error.
|
||||||
}
|
}
|
||||||
|
if let Err(e) = inputs.map.flush() {
|
||||||
|
first_error = Some(Error::IoError { source: e });
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
ConsumerSummary {
|
ConsumerSummary {
|
||||||
|
|||||||
Reference in New Issue
Block a user