|
| 1 | +//! A tail thread's self-reported liveness, packed into one atomic word. |
| 2 | +//! |
| 3 | +//! The old watchdog inferred the tail's state from a bare heartbeat timestamp, |
| 4 | +//! so "silent" had two indistinguishable causes: wedged in a syscall, or |
| 5 | +//! returned for good. A tail that gave up (journal permanently lost, rescan |
| 6 | +//! budget exhausted) then looked exactly like the stuck-IRP case the watchdog |
| 7 | +//! exists for, and it error-logged and cancelled I/O on the dead thread every |
| 8 | +//! scan, forever. The pulse makes the tail say which it is, and makes the |
| 9 | +//! "returned for good" report impossible to forget: it is published by a |
| 10 | +//! [`Drop`] guard, so every exit path, including a panic unwind and any |
| 11 | +//! `return` added later, retires the pulse. |
| 12 | +//! |
| 13 | +//! State and timestamp share one `u64` (tag in the top two bits, unix seconds |
| 14 | +//! below) so a reader always sees a consistent pair; two separate atomics |
| 15 | +//! could tear between a supervisor's loads. |
| 16 | +
|
| 17 | +use std::sync::atomic::{AtomicU64, Ordering}; |
| 18 | + |
| 19 | +const TAG_SHIFT: u32 = 62; |
| 20 | +const SECS_MASK: u64 = (1 << TAG_SHIFT) - 1; |
| 21 | +const TAG_BEAT: u64 = 0; |
| 22 | +const TAG_BUSY: u64 = 1; |
| 23 | +const TAG_RETIRED: u64 = 2; |
| 24 | + |
| 25 | +/// One decoded pulse observation. |
| 26 | +#[derive(Clone, Copy, Debug, PartialEq, Eq)] |
| 27 | +pub(crate) enum Pulse { |
| 28 | + /// The tail loop is turning; unix seconds of its last stamp. |
| 29 | + Beat(u64), |
| 30 | + /// The tail is inside a legitimately long, non-stamping operation (a full |
| 31 | + /// MFT rebuild); unix seconds of when it entered. |
| 32 | + Busy(u64), |
| 33 | + /// The tail thread returned. Not stuck, and not coming back on its own. |
| 34 | + Retired, |
| 35 | +} |
| 36 | + |
| 37 | +/// The shared word a tail writes and its supervisor reads. |
| 38 | +/// |
| 39 | +/// Ordering invariant: `Retired` must be the last value ever stored, and that |
| 40 | +/// is guaranteed structurally rather than atomically. Busy scopes live inside |
| 41 | +/// `tail_loop` while the retire guard wraps the whole call, so guard drop |
| 42 | +/// order publishes any pending `Beat` restore before `Retired`. This keeps the |
| 43 | +/// hot-path stamp a plain relaxed store instead of a compare-exchange. |
| 44 | +pub(crate) struct TailPulse(AtomicU64); |
| 45 | + |
| 46 | +impl TailPulse { |
| 47 | + /// A fresh pulse, beating as of now (the tail is about to start). |
| 48 | + pub(crate) fn new() -> Self { |
| 49 | + Self(AtomicU64::new(pack(TAG_BEAT, now_unix()))) |
| 50 | + } |
| 51 | + |
| 52 | + /// Stamps liveness. Called once per tail-loop turn. |
| 53 | + pub(crate) fn beat(&self) { |
| 54 | + self.0.store(pack(TAG_BEAT, now_unix()), Ordering::Relaxed); |
| 55 | + } |
| 56 | + |
| 57 | + /// The current observation. |
| 58 | + pub(crate) fn load(&self) -> Pulse { |
| 59 | + let word = self.0.load(Ordering::Relaxed); |
| 60 | + let secs = word & SECS_MASK; |
| 61 | + match word >> TAG_SHIFT { |
| 62 | + TAG_BUSY => Pulse::Busy(secs), |
| 63 | + TAG_RETIRED => Pulse::Retired, |
| 64 | + _ => Pulse::Beat(secs), |
| 65 | + } |
| 66 | + } |
| 67 | + |
| 68 | + /// Marks the tail busy until the returned guard drops, which restores a |
| 69 | + /// fresh beat so a long rebuild does not read as an instant stall the |
| 70 | + /// moment it finishes. Scopes must not nest (the inner drop would end the |
| 71 | + /// outer scope early); the tail has exactly one, around `rescan`. |
| 72 | + pub(crate) fn busy_scope(&self) -> BusyGuard<'_> { |
| 73 | + self.0.store(pack(TAG_BUSY, now_unix()), Ordering::Relaxed); |
| 74 | + BusyGuard(self) |
| 75 | + } |
| 76 | + |
| 77 | + /// Publishes `Retired` when the returned guard drops. Held across the |
| 78 | + /// whole `tail_loop` call so every exit, panic unwinds included, reports |
| 79 | + /// the thread as gone rather than leaving a heartbeat that merely went |
| 80 | + /// quiet. |
| 81 | + pub(crate) fn retire_on_drop(&self) -> RetireGuard<'_> { |
| 82 | + RetireGuard(self) |
| 83 | + } |
| 84 | +} |
| 85 | + |
| 86 | +/// Restores a fresh [`Pulse::Beat`] on drop. See [`TailPulse::busy_scope`]. |
| 87 | +pub(crate) struct BusyGuard<'a>(&'a TailPulse); |
| 88 | + |
| 89 | +impl Drop for BusyGuard<'_> { |
| 90 | + fn drop(&mut self) { |
| 91 | + self.0.beat(); |
| 92 | + } |
| 93 | +} |
| 94 | + |
| 95 | +/// Publishes [`Pulse::Retired`] on drop. See [`TailPulse::retire_on_drop`]. |
| 96 | +pub(crate) struct RetireGuard<'a>(&'a TailPulse); |
| 97 | + |
| 98 | +impl Drop for RetireGuard<'_> { |
| 99 | + fn drop(&mut self) { |
| 100 | + self.0.0.store(pack(TAG_RETIRED, 0), Ordering::Relaxed); |
| 101 | + } |
| 102 | +} |
| 103 | + |
| 104 | +fn pack(tag: u64, secs: u64) -> u64 { |
| 105 | + (tag << TAG_SHIFT) | (secs & SECS_MASK) |
| 106 | +} |
| 107 | + |
| 108 | +pub(crate) fn now_unix() -> u64 { |
| 109 | + std::time::SystemTime::now() |
| 110 | + .duration_since(std::time::UNIX_EPOCH) |
| 111 | + .map(|d| d.as_secs()) |
| 112 | + .unwrap_or(0) |
| 113 | +} |
| 114 | + |
| 115 | +#[cfg(test)] |
| 116 | +mod tests { |
| 117 | + use super::*; |
| 118 | + |
| 119 | + #[test] |
| 120 | + fn new_pulse_beats_now() { |
| 121 | + let p = TailPulse::new(); |
| 122 | + let before = now_unix(); |
| 123 | + match p.load() { |
| 124 | + Pulse::Beat(at) => assert!(at <= before && before - at <= 1), |
| 125 | + other => panic!("fresh pulse must beat, got {other:?}"), |
| 126 | + } |
| 127 | + } |
| 128 | + |
| 129 | + #[test] |
| 130 | + fn busy_scope_restores_a_fresh_beat() { |
| 131 | + let p = TailPulse::new(); |
| 132 | + { |
| 133 | + let _busy = p.busy_scope(); |
| 134 | + assert!(matches!(p.load(), Pulse::Busy(_))); |
| 135 | + } |
| 136 | + assert!(matches!(p.load(), Pulse::Beat(_))); |
| 137 | + } |
| 138 | + |
| 139 | + /// The design pillar: retirement is published by unwinding, not by |
| 140 | + /// remembering to call something, so a panicking tail still reports |
| 141 | + /// itself gone instead of leaving a heartbeat that merely went quiet. |
| 142 | + #[test] |
| 143 | + fn a_panicking_thread_still_retires() { |
| 144 | + let p = std::sync::Arc::new(TailPulse::new()); |
| 145 | + let thread_p = p.clone(); |
| 146 | + let joined = std::thread::spawn(move || { |
| 147 | + let _retired = thread_p.retire_on_drop(); |
| 148 | + panic!("tail died unexpectedly"); |
| 149 | + }) |
| 150 | + .join(); |
| 151 | + assert!(joined.is_err(), "the thread must have panicked"); |
| 152 | + assert_eq!(p.load(), Pulse::Retired); |
| 153 | + } |
| 154 | + |
| 155 | + /// Guard drop order is the whole ordering story: a busy scope inside the |
| 156 | + /// retire guard must end first, so `Retired` is the final word even |
| 157 | + /// though nothing enforces it atomically. |
| 158 | + #[test] |
| 159 | + fn retire_wins_over_an_unwinding_busy_scope() { |
| 160 | + let p = TailPulse::new(); |
| 161 | + { |
| 162 | + let _retired = p.retire_on_drop(); |
| 163 | + let _busy = p.busy_scope(); |
| 164 | + // _busy drops first (restores Beat), then _retired publishes. |
| 165 | + } |
| 166 | + assert_eq!(p.load(), Pulse::Retired); |
| 167 | + } |
| 168 | + |
| 169 | + /// Timestamps survive the tag bits for any plausible clock value, |
| 170 | + /// including far-future ones. |
| 171 | + #[test] |
| 172 | + fn tag_bits_never_corrupt_the_seconds() { |
| 173 | + for secs in [0u64, 1, 1_752_800_000, SECS_MASK] { |
| 174 | + assert_eq!(pack(TAG_BEAT, secs) & SECS_MASK, secs); |
| 175 | + assert_eq!(pack(TAG_BUSY, secs) >> TAG_SHIFT, TAG_BUSY); |
| 176 | + } |
| 177 | + } |
| 178 | +} |
0 commit comments