section_recover: per-rip handler scorecard + Bisect leads the scouts
Scorecard (ephemeral, reset each pass, no persistence): grades every handler by recovery rate (bytes/sec). run_handlers orders handlers best-first by that rate; an untried handler ranks top so each is calibrated once, then the ranking narrows to the winners. Logged at pass end (phase=scorecard) so the operator sees which handler is pulling the weight on this drive/disc and which is a dud. Tier 0 scouts are now [Bisect, Jump, Linear-fast x2], scorecard-ordered. Bisect leads because probing a range's MIDDLE lands on a readable island in one read, where Jump (linear from the front, big skip) can grind the dead front or overshoot a small range entirely. The scorecard confirms or overturns that order with real per-disc data.
This commit is contained in:
Matthew Jackson
+28
-24
@@ -59,7 +59,8 @@ use crate::io::pipeline::{Flow, Sink};
|
||||
|
||||
use super::mapfile::{self, MapStats, Mapfile, SectorStatus};
|
||||
use super::section_recover::{
|
||||
Bisect, HandlerCtx, HandlerOutcome, Jump, Linear, RecoverySink, SectionHandler, run_handlers,
|
||||
Bisect, HandlerCtx, HandlerOutcome, HandlerScoreboard, Jump, Linear, RecoverySink,
|
||||
SectionHandler, run_handlers,
|
||||
};
|
||||
|
||||
/// Wall-clock budget one recovery handler gets on a section before the chain
|
||||
@@ -746,6 +747,10 @@ struct PatchCtx<'a, 'o> {
|
||||
total_bytes: u64,
|
||||
decrypt_is_aacs: bool,
|
||||
state: PatchLoopState,
|
||||
/// Per-rip handler scorecard: grades handlers by recovery rate so the
|
||||
/// coordinator runs the winners first and lets duds fall back. Reset per
|
||||
/// pass (ephemeral, no persistence).
|
||||
scoreboard: HandlerScoreboard,
|
||||
}
|
||||
|
||||
impl PatchCtx<'_, '_> {
|
||||
@@ -834,30 +839,28 @@ impl PatchCtx<'_, '_> {
|
||||
// itself via its `fast` flag.
|
||||
self.reader.set_speed(0xFFFF);
|
||||
|
||||
// Tier 0: the fast handlers only — sweep the readable bulk of EVERY range
|
||||
// before any slow grind. Tier 1: slow deep-recovery + bisect on what tier
|
||||
// 0 left. Adding a recovery idea is one more entry in the right tier (#55).
|
||||
// Tier 0 = FAST scouts, ordered best-first by the rip scorecard (see
|
||||
// run_handlers). Bisect leads by default because probing the MIDDLE of a
|
||||
// range finds a readable island in one read, where Jump has to grind the
|
||||
// dead front to reach it; Jump then blows through large dead runs, and
|
||||
// the fast linear sweeps mop up. The scorecard re-orders these as it
|
||||
// learns which one is actually winning on THIS disc. Tier 1 = slow deep
|
||||
// recovery on the small residue tier 0 leaves.
|
||||
let mut handlers: Vec<Box<dyn SectionHandler>> = if tier == 0 {
|
||||
// Tier 0 = a SINGLE fast scout (Jump only). It streams the big
|
||||
// readable wins back and skips dead runs in seconds — so the pass
|
||||
// sweeps every range fast, recovers the recoverable bulk largest
|
||||
// first, and converges to the small genuine-dead residue instead of
|
||||
// spending 3 handlers × 60 s grinding every dead fragment. Order of
|
||||
// recovery is exactly big-wins → smaller → smallest, then grind.
|
||||
vec![Box::new(Jump)]
|
||||
vec![
|
||||
Box::new(Bisect),
|
||||
Box::new(Jump),
|
||||
Box::new(Linear {
|
||||
reverse: true,
|
||||
fast: true,
|
||||
}),
|
||||
Box::new(Linear {
|
||||
reverse: false,
|
||||
fast: true,
|
||||
}),
|
||||
]
|
||||
} else {
|
||||
// Tier 1 = deep recovery on the (now small) residue: fast full-batch
|
||||
// mop-up of anything Jump stepped over, then slow deep-recovery reads,
|
||||
// then Bisect for readable islands inside a mostly-dead chunk.
|
||||
vec![
|
||||
Box::new(Linear {
|
||||
reverse: true,
|
||||
fast: true,
|
||||
}),
|
||||
Box::new(Linear {
|
||||
reverse: false,
|
||||
fast: true,
|
||||
}),
|
||||
Box::new(Linear {
|
||||
reverse: true,
|
||||
fast: false,
|
||||
@@ -866,7 +869,6 @@ impl PatchCtx<'_, '_> {
|
||||
reverse: false,
|
||||
fast: false,
|
||||
}),
|
||||
Box::new(Bisect),
|
||||
]
|
||||
};
|
||||
|
||||
@@ -911,7 +913,7 @@ impl PatchCtx<'_, '_> {
|
||||
decrypt_is_aacs: self.decrypt_is_aacs,
|
||||
tick: Some(&mut tick),
|
||||
};
|
||||
run_handlers(&mut ctx, &mut handlers, bad, |_bad| {
|
||||
run_handlers(&mut ctx, &mut handlers, bad, &mut self.scoreboard, |_bad| {
|
||||
now_ptr() + std::time::Duration::from_secs(PER_HANDLER_BUDGET_SECS)
|
||||
})
|
||||
};
|
||||
@@ -1231,8 +1233,10 @@ impl Disc {
|
||||
total_bytes,
|
||||
decrypt_is_aacs,
|
||||
state: PatchLoopState::new(bytes_good_before, total_bytes, initial_batch, work_total),
|
||||
scoreboard: HandlerScoreboard::default(),
|
||||
};
|
||||
ctx.run(&bad_ranges)?;
|
||||
ctx.scoreboard.log();
|
||||
let PatchCtx { state, .. } = ctx;
|
||||
|
||||
// Drain the consumer thread: drop tx, wait for `close` to run
|
||||
|
||||
+93
-18
@@ -420,7 +420,6 @@ impl SectionHandler for Jump {
|
||||
for (rp, rl) in snapshot {
|
||||
let mut off = 0u64;
|
||||
let mut consec_fail = 0u32;
|
||||
let mut jump = JUMP_BASE_BYTES;
|
||||
while off < rl {
|
||||
if ctx.halted() {
|
||||
return HandlerOutcome::Halted;
|
||||
@@ -435,19 +434,22 @@ impl SectionHandler for Jump {
|
||||
ReadHit::Good => {
|
||||
bad.remove(pos, span);
|
||||
consec_fail = 0;
|
||||
jump = JUMP_BASE_BYTES;
|
||||
off += span;
|
||||
}
|
||||
ReadHit::Bad => {
|
||||
consec_fail += 1;
|
||||
if consec_fail >= JUMP_AFTER_FAILS {
|
||||
// Sustained dead run — skip ahead (sector-aligned so
|
||||
// the walk stays batch-aligned) and escalate the next
|
||||
// jump. The skipped span stays bad for Bisect / a
|
||||
// later handler to pin the boundary.
|
||||
let step = (jump / SECTOR).max(1) * SECTOR;
|
||||
off = (off + step).min(rl);
|
||||
jump = jump.saturating_mul(2).min(JUMP_CAP_BYTES);
|
||||
// Sustained dead run — jump to the MIDDLE of the
|
||||
// remaining span (never overshoot the range). Halving
|
||||
// adapts to any size: a big dead run is crossed in
|
||||
// ~log2 jumps, and a small range lands mid-range
|
||||
// instead of being skipped past entirely (the 8 MiB
|
||||
// fixed jump used to leap clean over a <8 MiB range and
|
||||
// miss readable data in its middle). The skipped span
|
||||
// stays bad for Bisect to reclaim.
|
||||
let remaining = rl - off;
|
||||
let step = ((remaining / 2) / SECTOR).max(1) * SECTOR;
|
||||
off += step;
|
||||
consec_fail = 0;
|
||||
} else {
|
||||
off += span;
|
||||
@@ -465,31 +467,102 @@ impl SectionHandler for Jump {
|
||||
}
|
||||
}
|
||||
|
||||
/// Run the handler chain over one section's still-bad set. This is the
|
||||
/// never-hang guarantee: each handler is bounded by the deadline
|
||||
/// `section_deadline_for(bad)` returns, and the loop always drains to
|
||||
/// `Complete`/`Remaining` (whatever is still bad is the caller's residue to
|
||||
/// record as loss). `Halted` / `TransportFault` short-circuit so the caller can
|
||||
/// abort or un-wedge.
|
||||
/// Per-rip handler scorecard. Grades each handler by the recovery RATE it has
|
||||
/// achieved so far (bytes recovered per second of wall time) so the coordinator
|
||||
/// runs the best-performing handler FIRST on later sections and lets a proven
|
||||
/// dud fall to the back. Ephemeral — reset each rip, no persistence. A handler
|
||||
/// not yet tried ranks top (`u64::MAX`) so every handler is calibrated once
|
||||
/// before the ranking narrows to the winners ("try each quick, then prioritise").
|
||||
#[derive(Default)]
|
||||
pub(super) struct HandlerScoreboard {
|
||||
stats: std::collections::HashMap<&'static str, ScoreStat>,
|
||||
}
|
||||
|
||||
#[derive(Default, Clone, Copy)]
|
||||
struct ScoreStat {
|
||||
recovered: u64,
|
||||
nanos: u128,
|
||||
attempts: u64,
|
||||
}
|
||||
|
||||
impl HandlerScoreboard {
|
||||
fn rate(s: &ScoreStat) -> u64 {
|
||||
if s.nanos == 0 {
|
||||
0
|
||||
} else {
|
||||
((s.recovered as u128 * 1_000_000_000) / s.nanos).min(u64::MAX as u128) as u64
|
||||
}
|
||||
}
|
||||
|
||||
/// Record one attempt: bytes recovered over `elapsed`.
|
||||
fn record(&mut self, name: &'static str, recovered: u64, elapsed: std::time::Duration) {
|
||||
let e = self.stats.entry(name).or_default();
|
||||
e.recovered = e.recovered.saturating_add(recovered);
|
||||
e.nanos = e.nanos.saturating_add(elapsed.as_nanos());
|
||||
e.attempts += 1;
|
||||
}
|
||||
|
||||
/// Ranking key (higher runs earlier). Untried → top, so it gets calibrated.
|
||||
fn rank(&self, name: &str) -> u64 {
|
||||
match self.stats.get(name) {
|
||||
None => u64::MAX,
|
||||
Some(s) if s.nanos == 0 => u64::MAX,
|
||||
Some(s) => Self::rate(s),
|
||||
}
|
||||
}
|
||||
|
||||
/// Emit the scorecard to the log so the operator can see, per rip, which
|
||||
/// handler is pulling the weight and which is a dud on this drive/disc.
|
||||
pub(super) fn log(&self) {
|
||||
let mut rows: Vec<_> = self.stats.iter().collect();
|
||||
rows.sort_by_key(|(_, s)| std::cmp::Reverse(Self::rate(s)));
|
||||
for (name, s) in rows {
|
||||
let mbps = s.recovered as f64 / (s.nanos as f64 / 1e9).max(1e-9) / 1_048_576.0;
|
||||
tracing::info!(
|
||||
target: "freemkv::disc",
|
||||
phase = "scorecard",
|
||||
handler = *name,
|
||||
recovered_mb = s.recovered as f64 / 1_048_576.0,
|
||||
attempts = s.attempts,
|
||||
mb_per_s = mbps,
|
||||
"handler scorecard (this rip)"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Run the handler chain over one section's still-bad set, ordered best-first by
|
||||
/// the rip scorecard. Never-hang guarantee: each handler is deadline-bounded and
|
||||
/// the loop always drains to `Complete`/`Remaining`. `Halted` / `TransportFault`
|
||||
/// short-circuit so the caller can abort or un-wedge. Each attempt is scored so
|
||||
/// later sections run the winners first.
|
||||
pub(super) fn run_handlers(
|
||||
ctx: &mut HandlerCtx,
|
||||
handlers: &mut [Box<dyn SectionHandler>],
|
||||
bad: &mut SubRanges,
|
||||
scoreboard: &mut HandlerScoreboard,
|
||||
section_deadline_for: impl Fn(&SubRanges) -> Instant,
|
||||
) -> HandlerOutcome {
|
||||
// Best-first by recovery rate so far; untried handlers rank top (calibrate).
|
||||
handlers.sort_by_key(|h| std::cmp::Reverse(scoreboard.rank(h.name())));
|
||||
for handler in handlers.iter_mut() {
|
||||
if bad.is_empty() {
|
||||
return HandlerOutcome::Complete;
|
||||
}
|
||||
let before = bad.total_len();
|
||||
let deadline = section_deadline_for(bad);
|
||||
let started = (ctx.now)();
|
||||
let outcome = handler.recover(ctx, bad, deadline);
|
||||
let elapsed = (ctx.now)().duration_since(started);
|
||||
let after = bad.total_len();
|
||||
scoreboard.record(handler.name(), before.saturating_sub(after), elapsed);
|
||||
tracing::info!(
|
||||
target: "freemkv::disc",
|
||||
phase = "section_recover.handler",
|
||||
handler = handler.name(),
|
||||
bad_bytes_before = before,
|
||||
bad_bytes_after = bad.total_len(),
|
||||
bad_bytes_after = after,
|
||||
recovered = before.saturating_sub(after),
|
||||
outcome = ?outcome,
|
||||
"handler finished; remaining bad bytes carry to the next handler"
|
||||
);
|
||||
@@ -799,7 +872,8 @@ mod tests {
|
||||
Box::new(Bisect),
|
||||
];
|
||||
let deadline_base = (ctx.now)();
|
||||
let out = run_handlers(&mut ctx, &mut handlers, &mut bad, |_| {
|
||||
let mut scoreboard = HandlerScoreboard::default();
|
||||
let out = run_handlers(&mut ctx, &mut handlers, &mut bad, &mut scoreboard, |_| {
|
||||
deadline_base + Duration::from_secs(30)
|
||||
});
|
||||
assert_eq!(out, HandlerOutcome::Remaining);
|
||||
@@ -831,7 +905,8 @@ mod tests {
|
||||
fast: true,
|
||||
})];
|
||||
let base = (ctx.now)();
|
||||
let out = run_handlers(&mut ctx, &mut handlers, &mut bad, |_| {
|
||||
let mut scoreboard = HandlerScoreboard::default();
|
||||
let out = run_handlers(&mut ctx, &mut handlers, &mut bad, &mut scoreboard, |_| {
|
||||
base + Duration::from_secs(30)
|
||||
});
|
||||
assert_eq!(out, HandlerOutcome::Complete);
|
||||
|
||||
Reference in New Issue
Block a user