diff --git a/crates/pdfs-cli/src/main.rs b/crates/pdfs-cli/src/main.rs index d4b52ba..7f70509 100644 --- a/crates/pdfs-cli/src/main.rs +++ b/crates/pdfs-cli/src/main.rs @@ -281,6 +281,10 @@ enum Command { }, /// Show the daemon's in-flight transfers (active uploads/downloads). Transfers, + /// Show what the daemon is doing right now: worker threads, queue depths, + /// in-flight control requests and memory. For a daemon that has stopped + /// answering rather than one that is merely busy. + Diagnostics, /// List what is in the account's trash, with the uids the restore and /// delete-forever commands take. Trash, @@ -674,6 +678,7 @@ fn main() -> Result<()> { Command::Mkdir { parent, name } => cmd_mkdir(parent, name), Command::Upload { sources, parent } => cmd_upload(sources, parent), Command::Transfers => cmd_transfers(), + Command::Diagnostics => cmd_diagnostics(), Command::Trash => cmd_trash(), Command::Restore { uids } => cmd_restore(uids), Command::DeleteForever { uids } => cmd_delete_forever(uids), @@ -2153,6 +2158,67 @@ fn cmd_rm(path: PathBuf) -> Result<()> { Ok(()) } +/// Seconds as something a human reads at a glance: the ages here run from +/// "still starting" to "wedged since yesterday". +fn human_secs(secs: u64) -> String { + match secs { + s if s < 60 => format!("{s}s"), + s if s < 3600 => format!("{}m{}s", s / 60, s % 60), + s if s < 86400 => format!("{}h{}m", s / 3600, (s % 3600) / 60), + s => format!("{}d{}h", s / 86400, (s % 86400) / 3600), + } +} + +fn cmd_diagnostics() -> Result<()> { + let response = control_request(CtlRequest::Diagnostics)?; + if emit_json(&response)? { + return Ok(()); + } + match response { + CtlResponse::Diagnostics { diagnostics: d } => { + println!( + "uptime {} rss {} pending ops {}", + human_secs(d.uptime_secs), + human_bytes(d.rss_bytes), + d.pending_ops, + ); + println!( + "control handlers {}/{}", + d.control_handlers, d.control_handler_limit, + ); + println!( + "queued: meta {} (done {}), transfer {} (done {})", + d.meta_queued, d.meta_completed, d.transfer_queued, d.transfer_completed, + ); + println!("workers:"); + for worker in &d.workers { + let what = if worker.busy { + worker.label.as_str() + } else { + "idle" + }; + println!( + " {:<20} {:<12} {}", + worker.name, + what, + human_secs(worker.age_secs) + ); + } + if d.inflight.is_empty() { + println!("in-flight requests: none"); + } else { + println!("in-flight requests (oldest first):"); + for request in &d.inflight { + println!(" {:<20} {}", request.kind, human_secs(request.age_secs)); + } + } + } + CtlResponse::Error { message, kind } => bail!("{}", cli_error(kind, &message)), + other => bail!("unexpected response: {other:?}"), + } + Ok(()) +} + fn cmd_trash() -> Result<()> { match control_request(CtlRequest::ListTrash)? { CtlResponse::Entries { entries } if entries.is_empty() => println!("(trash is empty)"), diff --git a/crates/pdfs-core/src/control.rs b/crates/pdfs-core/src/control.rs index d1ad1d1..bf48e7b 100644 --- a/crates/pdfs-core/src/control.rs +++ b/crates/pdfs-core/src/control.rs @@ -299,6 +299,13 @@ pub enum Request { /// [`Response::Transfers`]. Cheap to poll: the daemon keeps the registry in /// memory, so a front-end can render a live progress widget. GetQueueStatus, + /// Report what the daemon is doing *right now*, for debugging a daemon that + /// stopped answering rather than one that is merely busy. Replies + /// [`Response::Diagnostics`]. + /// + /// Answered from lock-free state on purpose: it has to come back while the + /// daemon is wedged, which is the only moment anyone asks for it. + Diagnostics, /// List what is in the account's trash. Replies with [`Response::Entries`]; /// a trashed node has no path inside the mount, so each entry carries only /// its `uid` — the handle for [`Request::Restore`] and @@ -1400,6 +1407,55 @@ pub fn pending_summary(uploads: u64, changes: u64) -> Option { } } +/// One worker thread in a [`Diagnostics`] report. +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct WorkerState { + /// Thread name (`pdfs-fuse-meta-0`, `pdfs-fuse-3`). + pub name: String, + pub busy: bool, + /// The job in hand (`read`, `lookup`), empty when idle. + pub label: String, + /// How long it has been in that state. A busy worker with a large age is + /// the shape of a hang. + pub age_secs: u64, +} + +/// One control request the daemon is still serving, in a [`Diagnostics`] report. +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct InflightRequest { + /// The request's variant name (`OpenFile`, `PhotosTimeline`). + pub kind: String, + pub age_secs: u64, +} + +/// What the daemon is doing right now (reply to [`Request::Diagnostics`]). +#[derive(Serialize, Deserialize, Debug, Clone, Default)] +pub struct Diagnostics { + /// How long this daemon process has been running. + pub uptime_secs: u64, + /// Resident set size, or `0` where it could not be read. The 2026-09-18 hang + /// peaked at 5.2 GiB, so this is the number that says whether the daemon is + /// growing again. + pub rss_bytes: u64, + pub workers: Vec, + /// Jobs waiting for a worker, per lane. + pub meta_queued: usize, + pub transfer_queued: usize, + /// Jobs finished per lane since start. Queue depth that does not move while + /// these do not either is a stalled lane, not a loaded one. + pub meta_completed: u64, + pub transfer_completed: u64, + /// Control requests still being served, oldest first. + pub inflight: Vec, + /// Control handler threads in use, and the ceiling. At the ceiling the + /// daemon refuses new connections, which looks exactly like a dead daemon + /// from the CLI. + pub control_handlers: usize, + pub control_handler_limit: usize, + /// Queued mutations not yet drained. + pub pending_ops: u64, +} + /// The daemon's reply to a [`Request`]. #[derive(Serialize, Deserialize, Debug)] pub enum Response { @@ -1508,6 +1564,8 @@ pub enum Response { available: bool, items: Vec, }, + /// What the daemon is doing right now (reply to [`Request::Diagnostics`]). + Diagnostics { diagnostics: Diagnostics }, /// How the Takeout import is doing (reply to [`Request::ImportStatus`]). /// `running` is false once it has finished, when `summary` is the final /// report; `summary` is `None` when no import has run this session. diff --git a/crates/pdfs-fuse/src/control.rs b/crates/pdfs-fuse/src/control.rs index 5ec8f1a..16006ac 100644 --- a/crates/pdfs-fuse/src/control.rs +++ b/crates/pdfs-fuse/src/control.rs @@ -67,6 +67,16 @@ impl Drop for ControlHandlerPermit { } } +/// Handler threads in use, for the diagnostics. At the limit the daemon refuses +/// connections, which a front-end cannot tell apart from a dead daemon. +pub(crate) fn active_handlers() -> usize { + ACTIVE_CONTROL_HANDLERS.load(Ordering::Acquire) +} + +pub(crate) fn handler_limit() -> usize { + MAX_CONTROL_HANDLERS +} + fn read_request_line(reader: &mut impl BufRead) -> std::io::Result> { let mut bytes = Vec::new(); loop { @@ -178,6 +188,9 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U return; } }; + // Registered before the request is even parsed, so a handler that wedges + // inside the daemon shows up by name in `pdfs diagnostics`. + let _inflight = super::diagnostics::InflightGuard::new(super::diagnostics::request_kind(&line)); let response = match serde_json::from_str::(line.trim()) { Ok(CtlRequest::Status) => { let pins = core.cache.list_pins(); @@ -657,6 +670,9 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U Err(e) => CtlResponse::error(e), } } + Ok(CtlRequest::Diagnostics) => CtlResponse::Diagnostics { + diagnostics: core.diagnostics(), + }, Ok(CtlRequest::ListTrash) => match core.list_trash() { Ok(entries) => CtlResponse::Entries { entries }, Err(e) => CtlResponse::error(e), diff --git a/crates/pdfs-fuse/src/diagnostics.rs b/crates/pdfs-fuse/src/diagnostics.rs new file mode 100644 index 0000000..fa9c6a2 --- /dev/null +++ b/crates/pdfs-fuse/src/diagnostics.rs @@ -0,0 +1,212 @@ +//! What the daemon is doing right now, for the case where it has stopped doing +//! anything at all. +//! +//! On 2026-09-18 the daemon hung: it consumed 43 minutes of CPU over 20 hours, +//! peaked at 5.2 GiB resident, went silent at 17:10, and only came back after a +//! manual `systemctl --user restart`. There was no panic, so nothing in the +//! journal said what it had been waiting on — and nothing in the daemon could be +//! asked, either. This module is the answer to "what was it stuck on?". +//! +//! Everything here is built to work *while* the daemon is wedged, which rules +//! out the obvious implementations: no lock that a stuck job might hold, no +//! database query on the calling thread, no waiting on the worker pool. A +//! diagnostic that blocks on the hang it is diagnosing reports nothing at +//! exactly the moment it matters. + +use std::sync::LazyLock; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Instant; + +use parking_lot::Mutex; +use pdfs_core::control::{Diagnostics, InflightRequest, WorkerState}; + +use super::Core; + +/// When this process started. Read at the first touch, which +/// [`super::mount::mount`] makes happen during startup so the uptime is the +/// daemon's, not the first diagnostics request's. +static STARTED: LazyLock = LazyLock::new(Instant::now); + +/// Control requests currently being served, keyed by a monotonic ticket. +/// +/// A `Vec` rather than a map because it holds at most `MAX_CONTROL_HANDLERS` +/// entries, and the only operations are push, remove-by-ticket and read-all. +static INFLIGHT: Mutex> = Mutex::new(Vec::new()); +static NEXT_TICKET: AtomicU64 = AtomicU64::new(0); + +/// Start the daemon's uptime clock. Called once, from the mount path. +pub(crate) fn start_clock() { + LazyLock::force(&STARTED); +} + +/// Registers one in-flight control request for as long as it is held. +pub(crate) struct InflightGuard(u64); + +impl InflightGuard { + pub(crate) fn new(kind: String) -> Self { + let ticket = NEXT_TICKET.fetch_add(1, Ordering::Relaxed); + INFLIGHT.lock().push((ticket, kind, Instant::now())); + Self(ticket) + } +} + +impl Drop for InflightGuard { + fn drop(&mut self) { + INFLIGHT.lock().retain(|(ticket, _, _)| *ticket != self.0); + } +} + +/// The variant name of a serialised [`pdfs_core::control::Request`], without +/// parsing it into one. +/// +/// serde's external tagging makes this a property of the text: a unit variant is +/// the bare string `"Status"`, anything else is an object whose single key is +/// the variant name. Reading the name off the line rather than matching on 80 +/// variants keeps this cheap, and keeps it correct when a variant is added. +pub(crate) fn request_kind(line: &str) -> String { + let line = line.trim(); + let name = if let Some(rest) = line.strip_prefix('{') { + rest.trim_start() + .strip_prefix('"') + .and_then(|rest| rest.split('"').next()) + } else { + line.strip_prefix('"') + .and_then(|rest| rest.split('"').next()) + }; + name.unwrap_or("unknown").to_string() +} + +/// Resident set size in bytes, or `0` when `/proc` did not answer. +/// +/// From `VmRSS` in `/proc/self/status`, which is already in kB — `statm` would +/// need the page size, and this needs no dependency at all. +pub(crate) fn rss_bytes() -> u64 { + let Ok(status) = std::fs::read_to_string("/proc/self/status") else { + return 0; + }; + parse_vm_rss(&status) +} + +/// The `VmRSS` line of a `/proc//status`, in bytes. +fn parse_vm_rss(status: &str) -> u64 { + status + .lines() + .find_map(|line| line.strip_prefix("VmRSS:")) + .and_then(|rest| rest.split_whitespace().next()) + .and_then(|kb| kb.parse::().ok()) + .map_or(0, |kb| kb.saturating_mul(1024)) +} + +impl Core { + /// Assemble a [`Diagnostics`] report. + /// + /// The pending-op count is the one field that needs the database, so it is + /// read on a throwaway thread with a short deadline: a daemon wedged *on* + /// the database must still be able to report everything else about itself. + pub(crate) fn diagnostics(&self) -> Diagnostics { + let pool = self.workers.snapshot(); + let now = Instant::now(); + let mut inflight: Vec = INFLIGHT + .try_lock() + .map(|entries| { + entries + .iter() + .map(|(_, kind, started)| InflightRequest { + kind: kind.clone(), + age_secs: now.duration_since(*started).as_secs(), + }) + .collect() + }) + .unwrap_or_default(); + inflight.sort_by_key(|request| std::cmp::Reverse(request.age_secs)); + + Diagnostics { + uptime_secs: STARTED.elapsed().as_secs(), + rss_bytes: rss_bytes(), + workers: pool + .workers + .into_iter() + .map(|worker| WorkerState { + name: worker.name, + busy: worker.busy, + label: worker.label, + age_secs: worker.age_ms / 1000, + }) + .collect(), + meta_queued: pool.meta_queued, + transfer_queued: pool.transfer_queued, + meta_completed: pool.meta_completed, + transfer_completed: pool.transfer_completed, + inflight, + control_handlers: super::control::active_handlers(), + control_handler_limit: super::control::handler_limit(), + pending_ops: self.pending_ops_quickly(), + } + } + + /// Queued mutations, or `0` if the database did not answer within + /// [`PENDING_COUNT_DEADLINE`]. + fn pending_ops_quickly(&self) -> u64 { + let db = self.db.clone(); + let (tx, rx) = std::sync::mpsc::channel(); + if std::thread::Builder::new() + .name("pdfs-diagnostics".into()) + .spawn(move || { + let counts = db.pending_op_counts().unwrap_or_default(); + let total = counts.uploads.max(0) as u64 + counts.changes.max(0) as u64; + let _ = tx.send(total); + }) + .is_err() + { + return 0; + } + rx.recv_timeout(PENDING_COUNT_DEADLINE).unwrap_or(0) + } +} + +/// How long the report waits for the database before reporting without it. +const PENDING_COUNT_DEADLINE: std::time::Duration = std::time::Duration::from_millis(500); + +#[cfg(test)] +mod tests { + use super::*; + + /// The request name is read off the wire format, so it must match both + /// shapes serde produces. + #[test] + fn the_request_kind_comes_off_both_wire_shapes() { + assert_eq!(request_kind("\"Status\""), "Status"); + assert_eq!(request_kind("{\"Pin\":{\"path\":\"a\"}}"), "Pin"); + assert_eq!(request_kind(" { \"Restore\" : {} } "), "Restore"); + assert_eq!(request_kind("garbage"), "unknown"); + } + + /// A guard registers for its lifetime and nothing longer: a handler that + /// returns must not leave a permanent "in flight" entry behind. + #[test] + fn a_guard_registers_only_while_it_lives() { + let before = INFLIGHT.lock().len(); + { + let _guard = InflightGuard::new("OpenFile".into()); + assert_eq!(INFLIGHT.lock().len(), before + 1); + } + assert_eq!(INFLIGHT.lock().len(), before); + } + + /// Whatever else it reports, the resident size has to be a real reading. + #[test] + fn rss_is_readable() { + assert!(rss_bytes() > 0); + } + + /// `VmRSS` is in kB, and a status without one must read as unknown rather + /// than as a daemon using no memory. + #[test] + fn vm_rss_is_kilobytes_and_absence_is_zero() { + assert_eq!( + parse_vm_rss("VmPeak:\t 100 kB\nVmRSS:\t 2048 kB\n"), + 2048 * 1024 + ); + assert_eq!(parse_vm_rss("VmPeak:\t 100 kB\n"), 0); + } +} diff --git a/crates/pdfs-fuse/src/filesystem.rs b/crates/pdfs-fuse/src/filesystem.rs index 05dd0d6..15ac739 100644 --- a/crates/pdfs-fuse/src/filesystem.rs +++ b/crates/pdfs-fuse/src/filesystem.rs @@ -49,7 +49,7 @@ impl Filesystem for ProtonFs { /// ([`Core::account_quota_cached`] serves a warm one from memory). fn statfs(&self, _req: &Request, _ino: INodeNo, reply: ReplyStatfs) { let core = self.core.clone(); - self.core.workers.run(Lane::Meta, move || { + self.core.workers.run(Lane::Meta, "statfs", move || { let (max_space, used_space) = core.account_quota_cached().unwrap_or((0, 0)); let bsize = u64::from(REPORTED_BLKSIZE); let blocks = (max_space.max(0) as u64).div_ceil(bsize); @@ -89,7 +89,7 @@ impl Filesystem for ProtonFs { return; } let fs = self.clone(); - self.core.workers.run(Lane::Meta, move || { + self.core.workers.run(Lane::Meta, "lookup", move || { fs.serve_lookup(parent, &name, reply, true) }); } @@ -183,7 +183,7 @@ impl Filesystem for ProtonFs { let fs = self.clone(); self.core .workers - .run(Lane::Meta, move || fs.serve_opendir(ino, reply)); + .run(Lane::Meta, "opendir", move || fs.serve_opendir(ino, reply)); } fn readdir( @@ -210,9 +210,9 @@ impl Filesystem for ProtonFs { return; } let fs = self.clone(); - self.core - .workers - .run(Lane::Meta, move || fs.serve_readdir(ino, offset, reply)); + self.core.workers.run(Lane::Meta, "readdir", move || { + fs.serve_readdir(ino, offset, reply) + }); } fn releasedir( @@ -237,9 +237,9 @@ impl Filesystem for ProtonFs { return; } let fs = self.clone(); - self.core - .workers - .run(Lane::Transfer, move || fs.serve_open(ino, flags, reply)); + self.core.workers.run(Lane::Transfer, "open", move || { + fs.serve_open(ino, flags, reply) + }); } fn read( @@ -329,7 +329,7 @@ impl Filesystem for ProtonFs { // even read off the FUSE device. It only reads `state`, so moving it // races with nothing. FUSE does not require replies in request order. let core = self.core.clone(); - self.core.workers.run(Lane::Transfer, move || { + self.core.workers.run(Lane::Transfer, "read", move || { match core.read_range(&uid, mtime, fsize, offset, size as u64, cache_blocks) { Ok(bytes) => reply.data(&bytes), Err(e) => reply.error(e), @@ -366,9 +366,9 @@ impl Filesystem for ProtonFs { // one create at a time, with `ls` on the same mount timing out for as // long as the burst lasted. See [`Workers`]. let fs = self.clone(); - self.core - .workers - .run(Lane::Meta, move || fs.serve_create(parent, &name, reply)); + self.core.workers.run(Lane::Meta, "create", move || { + fs.serve_create(parent, &name, reply) + }); } fn write( @@ -506,7 +506,7 @@ impl Filesystem for ProtonFs { // the scratch file). let ino = ino.0; let fs = self.clone(); - self.core.workers.run(Lane::Transfer, move || { + self.core.workers.run(Lane::Transfer, "truncate", move || { if let Err(e) = fs.core.queue_truncate(ino, size) { reply.error(e); return; @@ -620,7 +620,7 @@ impl Filesystem for ProtonFs { let fs = self.clone(); self.core .workers - .run(Lane::Transfer, move || fs.serve_fsync(fh, reply)); + .run(Lane::Transfer, "fsync", move || fs.serve_fsync(fh, reply)); } fn release( @@ -680,7 +680,7 @@ impl Filesystem for ProtonFs { }; // `trash_child` is a remote call; see `serve_create`. let fs = self.clone(); - self.core.workers.run(Lane::Meta, move || { + self.core.workers.run(Lane::Meta, "unlink", move || { fs.serve_unlink(parent, &name_str, reply) }); } @@ -701,9 +701,9 @@ impl Filesystem for ProtonFs { // Enumerates the folder to apply rmdir's emptiness rule, then trashes // it: two remote calls. See `serve_create`. let fs = self.clone(); - self.core - .workers - .run(Lane::Meta, move || fs.serve_rmdir(parent, &name_str, reply)); + self.core.workers.run(Lane::Meta, "rmdir", move || { + fs.serve_rmdir(parent, &name_str, reply) + }); } fn mkdir( @@ -731,9 +731,9 @@ impl Filesystem for ProtonFs { // remote round trip, and inline it blocks every other request on the // mount until it answers. let fs = self.clone(); - self.core - .workers - .run(Lane::Meta, move || fs.serve_mkdir(parent, &name, reply)); + self.core.workers.run(Lane::Meta, "mkdir", move || { + fs.serve_mkdir(parent, &name, reply) + }); } fn rename( @@ -774,7 +774,7 @@ impl Filesystem for ProtonFs { // move), and the replaced-destination path adds a trash on top. See // `serve_create` for why none of that may run on the dispatch loop. let fs = self.clone(); - self.core.workers.run(Lane::Meta, move || { + self.core.workers.run(Lane::Meta, "rename", move || { fs.serve_rename(parent, &name, newparent, &newname, flags, reply) }); } @@ -799,7 +799,7 @@ impl Filesystem for ProtonFs { // network per call (B5). let fs = self.clone(); let name = name.to_os_string(); - self.core.workers.run(Lane::Meta, move || { + self.core.workers.run(Lane::Meta, "getxattr", move || { fs.serve_getxattr(ino, &name, size, reply) }); } @@ -1099,7 +1099,7 @@ impl ProtonFs { if !off_loop { let fs = self.clone(); let name = name.to_string(); - self.core.workers.run(Lane::Meta, move || { + self.core.workers.run(Lane::Meta, "lookup", move || { fs.serve_lookup(parent, &name, reply, true) }); return; diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index 1c04cc6..c99ede4 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -83,6 +83,7 @@ mod background; mod control; use control::run_control_socket; mod devices; +mod diagnostics; mod drain; mod filesystem; pub use filesystem::ProtonFs; diff --git a/crates/pdfs-fuse/src/mount.rs b/crates/pdfs-fuse/src/mount.rs index bea904f..53a125d 100644 --- a/crates/pdfs-fuse/src/mount.rs +++ b/crates/pdfs-fuse/src/mount.rs @@ -341,6 +341,9 @@ pub fn mount( username, sweep_mode, } = options; + // Start the uptime clock here rather than at the first `pdfs diagnostics`, + // so the age it reports is the daemon's own. + super::diagnostics::start_clock(); let (root, online) = fetch_or_recall_root(&client, &rt, &db)?; let scope = root.tree_event_scope_id(); db.mount_upsert_my_files(&mountpoint.to_string_lossy(), &root.uid.to_string(), None) diff --git a/crates/pdfs-fuse/src/workers.rs b/crates/pdfs-fuse/src/workers.rs index 5b93717..743d534 100644 --- a/crates/pdfs-fuse/src/workers.rs +++ b/crates/pdfs-fuse/src/workers.rs @@ -12,6 +12,8 @@ use std::collections::VecDeque; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; +use std::time::Instant; use parking_lot::{Condvar, Mutex}; use tracing::warn; @@ -53,6 +55,13 @@ const META_WORKERS: usize = 3; type Job = Box; +/// A queued job and the name its caller gave it, so a job that never finishes +/// can be named in the diagnostics instead of showing up as "a worker is busy". +struct Queued { + label: &'static str, + job: Job, +} + /// Which lane a job belongs in. #[derive(Clone, Copy, PartialEq, Eq)] pub(crate) enum Lane { @@ -65,12 +74,63 @@ pub(crate) enum Lane { #[derive(Default)] struct Queues { - meta: VecDeque, - transfer: VecDeque, + meta: VecDeque, + transfer: VecDeque, /// Set when the pool is dropped, to wake and retire every worker. closed: bool, } +/// What one worker thread is doing, published for [`Workers::snapshot`]. +/// +/// Every field is an atomic (or a lock the worker only ever holds for the length +/// of a pointer write) because the whole point of the snapshot is to answer +/// *while the daemon is wedged*: a diagnostic that waits on the same lock the +/// stuck job holds reports nothing at exactly the moment it is needed. +struct Slot { + name: String, + /// True between picking a job up and finishing it. + busy: AtomicBool, + /// When the current state (busy or idle) began, as milliseconds since the + /// pool started. Monotonic, so an age is always a real duration. + since_ms: AtomicU64, + /// The label of the job in hand. Behind a mutex because a `&'static str` is + /// two words and there is no atomic for that; the worker holds it only to + /// swap the pointer, and the snapshot never blocks on it (see + /// [`Workers::snapshot`]). + label: Mutex<&'static str>, +} + +impl Slot { + fn new(name: String, now_ms: u64) -> Self { + Self { + name, + busy: AtomicBool::new(false), + since_ms: AtomicU64::new(now_ms), + label: Mutex::new(""), + } + } +} + +/// One worker's state at the moment [`Workers::snapshot`] read it. +pub(crate) struct WorkerSnapshot { + pub(crate) name: String, + pub(crate) busy: bool, + /// How long the worker has been in that state. + pub(crate) age_ms: u64, + /// The job in hand, or `""` when idle. `"?"` when the label could not be + /// read without blocking. + pub(crate) label: String, +} + +/// The pool's state at the moment [`Workers::snapshot`] read it. +pub(crate) struct PoolSnapshot { + pub(crate) workers: Vec, + pub(crate) meta_queued: usize, + pub(crate) transfer_queued: usize, + pub(crate) meta_completed: u64, + pub(crate) transfer_completed: u64, +} + /// The queues and the two wait sets over them. /// /// Two condvars, not one, because the workers are not interchangeable: a @@ -86,6 +146,22 @@ struct Pool { meta_cv: Condvar, /// Waited on by the general workers. general_cv: Condvar, + /// Queue depths, mirrored outside the lock so a snapshot can read them + /// without waiting on whatever is holding it. + meta_queued: AtomicUsize, + transfer_queued: AtomicUsize, + /// Jobs finished per lane. A depth that stays high while this stays still is + /// a stalled lane rather than a busy one. + meta_completed: AtomicU64, + transfer_completed: AtomicU64, + slots: Vec, + started: Instant, +} + +impl Pool { + fn now_ms(&self) -> u64 { + self.started.elapsed().as_millis() as u64 + } } /// Bounded thread pool behind the network-touching FUSE handlers. @@ -98,21 +174,35 @@ pub(crate) struct Workers { impl Workers { pub(crate) fn new(n: usize) -> std::io::Result { + // Never leave the pool without a general worker, however small `n` is. + let meta_workers = META_WORKERS.min(n.saturating_sub(1)); + let names: Vec = (0..n) + .map(|i| { + format!( + "pdfs-fuse-{}{i}", + if i < meta_workers { "meta-" } else { "" } + ) + }) + .collect(); let inner = Arc::new(Pool { queues: Mutex::new(Queues::default()), meta_cv: Condvar::new(), general_cv: Condvar::new(), + meta_queued: AtomicUsize::new(0), + transfer_queued: AtomicUsize::new(0), + meta_completed: AtomicU64::new(0), + transfer_completed: AtomicU64::new(0), + slots: names + .iter() + .map(|name| Slot::new(name.clone(), 0)) + .collect(), + started: Instant::now(), }); - // Never leave the pool without a general worker, however small `n` is. - let meta_workers = META_WORKERS.min(n.saturating_sub(1)); - for i in 0..n { + for (i, name) in names.iter().enumerate() { let inner = inner.clone(); let meta_only = i < meta_workers; std::thread::Builder::new() - .name(format!( - "pdfs-fuse-{}{i}", - if meta_only { "meta-" } else { "" } - )) + .name(name.clone()) .spawn(move || { let cv = if meta_only { &inner.meta_cv @@ -121,17 +211,20 @@ impl Workers { }; loop { let mut q = inner.queues.lock(); - let job = loop { + let picked = loop { // A general worker takes transfers first: metadata // already has threads of its own, so draining the // bulk queue is the useful thing for it to do. - let picked = if meta_only { - q.meta.pop_front() + let taken = if meta_only { + q.meta.pop_front().map(|job| (Lane::Meta, job)) } else { - q.transfer.pop_front().or_else(|| q.meta.pop_front()) + q.transfer + .pop_front() + .map(|job| (Lane::Transfer, job)) + .or_else(|| q.meta.pop_front().map(|job| (Lane::Meta, job))) }; - if let Some(job) = picked { - break Some(job); + if let Some(taken) = taken { + break Some(taken); } if q.closed { break None; @@ -139,7 +232,15 @@ impl Workers { cv.wait(&mut q); }; drop(q); - let Some(job) = job else { break }; + let Some((lane, queued)) = picked else { break }; + match lane { + Lane::Meta => inner.meta_queued.fetch_sub(1, Ordering::Relaxed), + Lane::Transfer => inner.transfer_queued.fetch_sub(1, Ordering::Relaxed), + }; + let slot = &inner.slots[i]; + *slot.label.lock() = queued.label; + slot.since_ms.store(inner.now_ms(), Ordering::Relaxed); + slot.busy.store(true, Ordering::Relaxed); // A panicking handler must not cost the pool a worker // for the rest of the run. The dropped `Reply` answers // EIO on its own, so the caller of the failed op is @@ -164,7 +265,16 @@ impl Workers { // poisoning at all. Anything new reached from a job owes // the same check — an SDK-side `.lock().unwrap()` // silently invalidates this comment. - let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(job)); + let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(queued.job)); + slot.busy.store(false, Ordering::Relaxed); + slot.since_ms.store(inner.now_ms(), Ordering::Relaxed); + *slot.label.lock() = ""; + match lane { + Lane::Meta => inner.meta_completed.fetch_add(1, Ordering::Relaxed), + Lane::Transfer => { + inner.transfer_completed.fetch_add(1, Ordering::Relaxed) + } + }; } })?; } @@ -179,7 +289,9 @@ impl Workers { /// a pending job holds a `Reply` and a few fields, while the 4 MiB block /// buffer is allocated inside the job once it runs, and the SDK's in-flight /// semaphore is what bounds how many of those exist at once. - pub(crate) fn run(&self, lane: Lane, job: impl FnOnce() + Send + 'static) { + /// `label` names the work for the diagnostics (`pdfs diagnostics`) and for + /// the stall warnings: it is what a hung daemon reports instead of "busy". + pub(crate) fn run(&self, lane: Lane, label: &'static str, job: impl FnOnce() + Send + 'static) { let mut q = self.inner.queues.lock(); if q.closed { // Pre-pool behaviour: a shut-down pool degrades to a slow mount @@ -189,9 +301,19 @@ impl Workers { job(); return; } + let queued = Queued { + label, + job: Box::new(job), + }; match lane { - Lane::Meta => q.meta.push_back(Box::new(job)), - Lane::Transfer => q.transfer.push_back(Box::new(job)), + Lane::Meta => { + q.meta.push_back(queued); + self.inner.meta_queued.fetch_add(1, Ordering::Relaxed); + } + Lane::Transfer => { + q.transfer.push_back(queued); + self.inner.transfer_queued.fetch_add(1, Ordering::Relaxed); + } } drop(q); match lane { @@ -209,6 +331,41 @@ impl Workers { } } } + + /// What every worker is doing right now, without taking the queue lock. + /// + /// A diagnostic is worth having exactly when the daemon is stuck, so this + /// reads atomics only, and `try_lock`s each label rather than waiting for + /// it: a label is held for one pointer write, so failing to get it means a + /// racing worker, not a wedged one, and `"?"` is a better answer than a + /// diagnostics request that hangs too. + pub(crate) fn snapshot(&self) -> PoolSnapshot { + let now = self.inner.now_ms(); + let workers = self + .inner + .slots + .iter() + .map(|slot| { + let label = slot + .label + .try_lock() + .map_or_else(|| "?".to_string(), |label| (*label).to_string()); + WorkerSnapshot { + name: slot.name.clone(), + busy: slot.busy.load(Ordering::Relaxed), + age_ms: now.saturating_sub(slot.since_ms.load(Ordering::Relaxed)), + label, + } + }) + .collect(); + PoolSnapshot { + workers, + meta_queued: self.inner.meta_queued.load(Ordering::Relaxed), + transfer_queued: self.inner.transfer_queued.load(Ordering::Relaxed), + meta_completed: self.inner.meta_completed.load(Ordering::Relaxed), + transfer_completed: self.inner.transfer_completed.load(Ordering::Relaxed), + } + } } impl Drop for Workers { @@ -225,6 +382,35 @@ mod tests { use std::sync::mpsc; use std::time::Duration; + /// A diagnostic is only useful if it names the work. A running job must show + /// up under its label, and the queue behind it must show up as depth. + #[test] + fn a_snapshot_names_the_running_job_and_the_queue_behind_it() { + let pool = Workers::new(2).unwrap(); + let (release_tx, release_rx) = mpsc::channel::<()>(); + let release_rx = Arc::new(Mutex::new(release_rx)); + let (started_tx, started_rx) = mpsc::channel(); + { + let release_rx = release_rx.clone(); + pool.run(Lane::Transfer, "read", move || { + started_tx.send(()).unwrap(); + let _ = release_rx.lock().recv(); + }); + } + started_rx + .recv_timeout(Duration::from_secs(5)) + .expect("the job starts"); + + let snapshot = pool.snapshot(); + assert!( + snapshot.workers.iter().any(|w| w.busy && w.label == "read"), + "the running job is named" + ); + assert_eq!(snapshot.transfer_completed, 0); + + let _ = release_tx.send(()); + } + /// Audit A6. Transfers filling every general worker must not delay a /// metadata job: that is what the reserved threads are for. #[test] @@ -240,7 +426,7 @@ mod tests { for _ in 0..general { let release_rx = release_rx.clone(); let started_tx = started_tx.clone(); - pool.run(Lane::Transfer, move || { + pool.run(Lane::Transfer, "test", move || { started_tx.send(()).unwrap(); let _ = release_rx.lock().recv(); }); @@ -255,13 +441,13 @@ mod tests { // in the way that used to starve metadata. for _ in 0..16 { let release_rx = release_rx.clone(); - pool.run(Lane::Transfer, move || { + pool.run(Lane::Transfer, "test", move || { let _ = release_rx.lock().recv(); }); } let (meta_tx, meta_rx) = mpsc::channel(); - pool.run(Lane::Meta, move || meta_tx.send(()).unwrap()); + pool.run(Lane::Meta, "test", move || meta_tx.send(()).unwrap()); meta_rx .recv_timeout(Duration::from_secs(5)) .expect("a metadata job runs while every transfer thread is stuck"); @@ -279,7 +465,7 @@ mod tests { // must have been served by the general worker. let pool = Workers::new(1).unwrap(); let (tx, rx) = mpsc::channel(); - pool.run(Lane::Meta, move || tx.send(()).unwrap()); + pool.run(Lane::Meta, "test", move || tx.send(()).unwrap()); rx.recv_timeout(Duration::from_secs(5)) .expect("a general worker takes metadata when no transfer is waiting"); } @@ -288,9 +474,9 @@ mod tests { #[test] fn a_panicking_job_does_not_retire_its_worker() { let pool = Workers::new(1).unwrap(); - pool.run(Lane::Meta, || panic!("boom")); + pool.run(Lane::Meta, "test", || panic!("boom")); let (tx, rx) = mpsc::channel(); - pool.run(Lane::Meta, move || tx.send(()).unwrap()); + pool.run(Lane::Meta, "test", move || tx.send(()).unwrap()); rx.recv_timeout(Duration::from_secs(5)) .expect("the pool still serves work after a panic"); }