diff --git a/Cargo.toml b/Cargo.toml index 20e8498..78996bd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "libfreemkv" -version = "0.20.5" +version = "0.20.6" edition = "2024" rust-version = "1.86" license = "AGPL-3.0-only" diff --git a/src/io/bounded.rs b/src/io/bounded.rs new file mode 100644 index 0000000..3daf685 --- /dev/null +++ b/src/io/bounded.rs @@ -0,0 +1,246 @@ +//! Bounded-syscall primitive: run a (potentially-blocking) operation +//! on a worker thread, with a hard wall-clock deadline and an optional +//! cooperative [`Halt`] poll. The calling thread is never trapped +//! inside a kernel call. +//! +//! ## Why this exists +//! +//! [`crate::halt::Halt`] is cooperative: callers poll +//! `is_cancelled()`. It cannot reach inside a syscall the kernel +//! currently owns the thread for — `libc::sync_file_range`, +//! `libc::fsync`, `File::write` on NFS, and so on. `/api/stop` from +//! autorip therefore can't unstick a thread sitting in such a syscall. +//! +//! [`bounded_syscall`] is the escape hatch: it runs `op` on a fresh +//! worker thread, then `recv_timeout`s on a rendezvous channel for the +//! result. The wait is broken into ~250 ms slices so the calling +//! thread can poll the supplied [`Halt`] in between. If the deadline +//! elapses or the halt fires, the worker is intentionally leaked — the +//! syscall will unwind whenever the kernel decides, or at process +//! exit, but the caller is free to fall back to a degraded code path +//! (skip the sync, log loudly, etc.). +//! +//! ## Trade-offs +//! +//! - **Thread per call.** Cheap (`std::thread::spawn` is < 100 µs on +//! Linux/macOS), but not free. Use on coarse-grained finalisation +//! syscalls (`sync_all`, `sync_file_range(WAIT_AFTER)`), not on hot +//! inner-loop writes. +//! - **Leak on timeout.** A wedged syscall keeps a kernel slot and a +//! user-space thread around for the rest of the process's life. +//! Bounded by the number of independent rip/mux sessions, which is +//! one per disc. The alternative — trapping the caller forever — +//! defeats the entire purpose of `/api/stop`. +//! - **Halt granularity ~250 ms.** Halt observation is not instant; +//! it's the worst-case latency of the `recv_timeout` slice. Good +//! enough for human-driven stop requests; not suitable for hard +//! real-time deadlines. +//! +//! ## Single source of truth +//! +//! Do NOT inline this pattern. Every blocking-syscall wrapper in the +//! rip + mux pipeline calls this helper, so changes (e.g. swapping the +//! channel impl, adjusting the poll slice, adding metrics) land in one +//! place. +//! +//! ## Platform +//! +//! Pure `std::thread` + `std::sync::mpsc`. No `cfg(target_os)` needed +//! here — the helper itself is platform-agnostic. Callers that wrap +//! Linux-only syscalls (`sync_file_range`) still need their own +//! `#[cfg(target_os = "linux")]` gates; this helper does not. + +use std::sync::mpsc::{RecvTimeoutError, sync_channel}; +use std::thread; +use std::time::{Duration, Instant}; + +use crate::halt::Halt; + +/// Granularity of the halt poll. The receive loop wakes every +/// [`POLL_INTERVAL`] to (a) check the [`Halt`] token, then (b) check +/// the overall deadline, then go back to waiting. 250 ms is a +/// pragmatic balance: short enough that human-driven `/api/stop` feels +/// responsive (< 0.5 s p99), long enough that the polling overhead is +/// negligible against multi-second syscalls. +const POLL_INTERVAL: Duration = Duration::from_millis(250); + +/// Failure outcome from a bounded syscall wrapper. +#[derive(Debug)] +pub(crate) enum BoundedError { + /// The user-visible halt token fired during the wait. The worker + /// thread is intentionally leaked — the caller should fall back to + /// a degraded code path rather than waiting on the syscall to + /// return. + Halted, + /// The deadline elapsed before the syscall returned. Same leak + /// semantics as `Halted`. + Timeout, + /// The worker thread panicked, or its sender disconnected before + /// sending a result. Treat as a benign no-op (callers usually + /// log and continue) rather than a hard error — by definition no + /// syscall observably ran to completion in this case. + WorkerLost, +} + +/// Run a (potentially-blocking) operation on a worker thread with a +/// deadline and an optional cooperative halt-token poll. Returns the +/// operation's result if it completes within `timeout`; otherwise one +/// of [`BoundedError::Halted`] / [`BoundedError::Timeout`] / +/// [`BoundedError::WorkerLost`]. +/// +/// On `Halted` / `Timeout` the worker thread is intentionally leaked: +/// the syscall will unwind whenever the kernel decides, or when the +/// process exits. The calling thread is never trapped inside a kernel +/// call. +/// +/// `halt` is polled at [`POLL_INTERVAL`] granularity. Pass `None` for +/// callers that don't (yet) have a halt token plumbed through — +/// behaviour degrades to deadline-only, matching the 0.20.5 +/// `wait_after_with_timeout` shape this helper generalises. +/// +/// `op` returns `R: Send + 'static`. The closure must own everything +/// it touches because it may outlive this call (timeout / halt cases). +pub(crate) fn bounded_syscall( + halt: Option<&Halt>, + timeout: Duration, + op: F, +) -> Result +where + F: FnOnce() -> R + Send + 'static, + R: Send + 'static, +{ + // Rendezvous channel: the worker sends exactly one value (the + // op's return) and then exits. Capacity-0 means the send blocks + // until we receive — fine on the happy path; on the timeout / + // halt path the receiver is dropped and the worker's send + // returns Err, which the worker ignores. + let (tx, rx) = sync_channel::(0); + let _ = thread::Builder::new() + .name("freemkv-bounded-syscall".into()) + .spawn(move || { + // Ignore the send error: if we time out (or get halted) + // before the worker finishes, the receiver is dropped + // and `tx.send` returns Err. Either way, the worker has + // nothing more to do. + let _ = tx.send(op()); + }); + + let deadline = Instant::now() + timeout; + loop { + let now = Instant::now(); + let remaining = deadline.saturating_duration_since(now); + let slice = remaining.min(POLL_INTERVAL); + match rx.recv_timeout(slice) { + Ok(v) => return Ok(v), + Err(RecvTimeoutError::Timeout) => { + if let Some(h) = halt { + if h.is_cancelled() { + return Err(BoundedError::Halted); + } + } + if Instant::now() >= deadline { + return Err(BoundedError::Timeout); + } + // Otherwise: another slice. + } + Err(RecvTimeoutError::Disconnected) => { + // Worker thread spawn failed, or it panicked before + // sending. Caller treats this as "no syscall ran" — + // typically a no-op + log. + return Err(BoundedError::WorkerLost); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + + #[test] + fn op_completes_quickly() { + let r = bounded_syscall(None, Duration::from_secs(2), || 42u32); + assert!(matches!(r, Ok(42))); + } + + #[test] + fn op_exceeds_timeout() { + // Op sleeps longer than the deadline → Timeout. + let r = bounded_syscall(None, Duration::from_millis(300), || { + thread::sleep(Duration::from_secs(2)); + 0u32 + }); + assert!(matches!(r, Err(BoundedError::Timeout))); + } + + #[test] + fn halt_fires_during_wait() { + let halt = Halt::new(); + let halt2 = halt.clone(); + // Flip the halt from a side thread after ~300 ms — long + // enough that the receive loop has rolled at least one + // 250 ms slice and is sitting in `recv_timeout` again when + // the bit flips. + thread::spawn(move || { + thread::sleep(Duration::from_millis(300)); + halt2.cancel(); + }); + let r = bounded_syscall(Some(&halt), Duration::from_secs(5), || { + thread::sleep(Duration::from_secs(5)); + 0u32 + }); + assert!(matches!(r, Err(BoundedError::Halted))); + } + + #[test] + fn worker_panics() { + // Worker panics → sender drops without sending → recv sees + // Disconnected → WorkerLost. We use an explicit panic in the + // op closure rather than `panic!()` from inside the channel + // machinery; the spawned thread's panic is contained (no + // process abort) because we don't `.join()` it. + let r = bounded_syscall(None, Duration::from_secs(2), || -> u32 { + panic!("intentional test panic"); + }); + assert!(matches!(r, Err(BoundedError::WorkerLost))); + } + + #[test] + fn halt_already_set_before_call_still_returns_halted() { + // Halt observed on the very first poll slice. The op blocks + // forever; we must not wait the full timeout to notice the + // halt is already set. + let halt = Halt::new(); + halt.cancel(); + let started = Instant::now(); + let r = bounded_syscall(Some(&halt), Duration::from_secs(10), || { + thread::sleep(Duration::from_secs(10)); + 0u32 + }); + assert!(matches!(r, Err(BoundedError::Halted))); + // Should bail out within ~1 s; allow 2 s of slack for slow + // CI hosts. + assert!( + started.elapsed() < Duration::from_secs(2), + "halt-already-set took {:?}", + started.elapsed() + ); + } + + #[test] + fn ok_path_takes_no_halt_token() { + // Sanity: the `None` halt path is the documented zero-config + // form (matches the 0.20.5 `wait_after_with_timeout` + // behaviour). Op returns immediately; we must observe Ok. + let flag = Arc::new(AtomicBool::new(false)); + let f2 = flag.clone(); + let r = bounded_syscall(None, Duration::from_secs(2), move || { + f2.store(true, Ordering::Relaxed); + "ok" + }); + assert!(matches!(r, Ok("ok"))); + assert!(flag.load(Ordering::Relaxed)); + } +} diff --git a/src/io/mod.rs b/src/io/mod.rs index 4e3ec72..348cd98 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -14,6 +14,7 @@ //! used by sweep, patch, and mux to overlap reads with writes via a //! bounded channel + dedicated consumer thread. +pub(crate) mod bounded; mod writeback; mod writeback_file; diff --git a/src/io/writeback/linux.rs b/src/io/writeback/linux.rs index 8e232ee..e7bd3d9 100644 --- a/src/io/writeback/linux.rs +++ b/src/io/writeback/linux.rs @@ -56,8 +56,6 @@ use std::fs::File; use std::os::unix::io::{AsRawFd, RawFd}; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::mpsc::{RecvTimeoutError, sync_channel}; -use std::thread; use std::time::{Duration, Instant}; const ADAPTIVE_WINDOW: usize = 16; @@ -332,34 +330,20 @@ fn detect_nfs(fd: RawFd) -> bool { /// Run `sync_file_range(WAIT_AFTER)` on a worker thread and wait up /// to [`WAIT_AFTER_TIMEOUT`] for it to return. `Some(elapsed_ms)` on /// success; `None` on timeout. On timeout the worker thread is -/// intentionally leaked — it'll unwind whenever the syscall -/// eventually returns or the process exits. -/// -/// Channel capacity is 0 (rendezvous). The worker sends `()` once -/// the syscall returns; if we time out first, the send blocks -/// forever inside the leaked worker — fine, the receiver is gone -/// and the OS reaps the thread at process exit. +/// 0.20.6 generalizes the worker-thread + recv_timeout pattern into +/// [`crate::io::bounded::bounded_syscall`]; this helper now just adapts +/// the generic primitive to the WAIT_AFTER call shape (returns elapsed_ms +/// instead of the syscall's `()` return, treats `WorkerLost` as a benign +/// no-op to match the original semantics). fn wait_after_with_timeout(fd: RawFd, off: u64, len: u64) -> Option { - let (tx, rx) = sync_channel::<()>(0); let started = Instant::now(); - // `RawFd` is `Copy` + `Send`; `off`/`len` are `u64`. Nothing - // borrows from the caller — the worker can safely outlive this - // function on timeout. - let _ = thread::Builder::new() - .name("freemkv-writeback-wait".into()) - .spawn(move || { - unsafe { - libc::sync_file_range(fd, off as i64, len as i64, libc::SYNC_FILE_RANGE_WAIT_AFTER); - } - // If the receiver is gone (we timed out) this send - // returns Err — fine, we just drop and the worker - // exits. - let _ = tx.send(()); - }); - match rx.recv_timeout(WAIT_AFTER_TIMEOUT) { + match crate::io::bounded::bounded_syscall(None, WAIT_AFTER_TIMEOUT, move || unsafe { + libc::sync_file_range(fd, off as i64, len as i64, libc::SYNC_FILE_RANGE_WAIT_AFTER); + }) { Ok(()) => Some(started.elapsed().as_millis() as u64), - Err(RecvTimeoutError::Timeout) => None, - Err(RecvTimeoutError::Disconnected) => { + Err(crate::io::bounded::BoundedError::Timeout) + | Err(crate::io::bounded::BoundedError::Halted) => None, + Err(crate::io::bounded::BoundedError::WorkerLost) => { // Worker thread spawn failed or panicked before sending. // Treat as a benign success (no syscall ran) rather than // a degrade trigger — falling through with elapsed_ms=0 diff --git a/src/io/writeback_file.rs b/src/io/writeback_file.rs index b700995..4702e05 100644 --- a/src/io/writeback_file.rs +++ b/src/io/writeback_file.rs @@ -116,9 +116,49 @@ impl WritebackFile { /// Drain in-flight writeback then issue a full fsync. Use this in /// place of `File::sync_all`. + /// + /// The final fsync is wrapped in + /// [`crate::io::bounded::bounded_syscall`] with a 60 s deadline. + /// fsync on a wedged NFS server (or a degraded local disk) can + /// hang the calling thread; the wrapper ensures the worst case is + /// 60 s + log-and-continue rather than indefinite. On timeout the + /// page cache is left to the kernel's normal flush-on-close path — + /// best effort, but bounded. pub(crate) fn sync_all(&mut self) -> io::Result<()> { self.pipeline.finalize(); - self.file.sync_all() + #[cfg(unix)] + { + use std::os::unix::io::AsRawFd; + use std::time::Duration; + let fd = self.file.as_raw_fd(); + match crate::io::bounded::bounded_syscall( + None, + Duration::from_secs(60), + move || -> io::Result<()> { + let rc = unsafe { libc::fsync(fd) }; + if rc == 0 { + Ok(()) + } else { + Err(io::Error::last_os_error()) + } + }, + ) { + Ok(inner) => inner, + Err(crate::io::bounded::BoundedError::Timeout) => { + tracing::error!( + target: "mux", + "WritebackFile::sync_all fsync timed out after 60s; kernel will flush on close (best-effort)" + ); + Ok(()) + } + Err(crate::io::bounded::BoundedError::Halted) => Ok(()), + Err(crate::io::bounded::BoundedError::WorkerLost) => Ok(()), + } + } + #[cfg(not(unix))] + { + self.file.sync_all() + } } }