From 9d092360adfb048f3b24f98c4ccb803222432985 Mon Sep 17 00:00:00 2001 From: Matthew Jackson Date: Tue, 12 May 2026 01:02:38 -0700 Subject: [PATCH] v0.18.21: verbose debug logging for pipeline stalls (fixed) --- Cargo.toml | 2 +- src/io/pipeline.rs | 111 +++++++++++++++++++++++++++++++++++---------- 2 files changed, 89 insertions(+), 24 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 42de321..4354115 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "libfreemkv" -version = "0.18.20" +version = "0.18.21" edition = "2024" rust-version = "1.86" license = "AGPL-3.0-only" diff --git a/src/io/pipeline.rs b/src/io/pipeline.rs index f028ee0..1beade6 100644 --- a/src/io/pipeline.rs +++ b/src/io/pipeline.rs @@ -27,15 +27,11 @@ //! - Consumer panic is converted into //! `Error::IoError { source: io::Error::other(...) }`. //! -//! ## Dead-code suppression +//! ## Debug logging //! -//! Both `Disc::sweep` (`DEFAULT_PIPELINE_DEPTH`, `spawn_named`, -//! `try_send`) and `Disc::patch` (`WRITE_THROUGH_DEPTH`, `Flow::Stop`) -//! are now in-tree callers (0.18 round 2). The targeted `#[allow]` -//! markers below cover the corners of the API not exercised by either -//! call site (e.g. `Pipeline::spawn` without a name; `Flow::Stop` -//! return path from sweep's sink — sweep always consumes the producer's -//! full work-list). Drop the allows once mux migrates. +//! Set `FREEMKV_DEBUG=1` environment variable to enable verbose debug +//! logging throughout the pipeline (channel sends/receives, backpressure, +//! consumer lag detection). This is critical for diagnosing stalls. use std::io; use std::sync::mpsc::{SyncSender, sync_channel}; @@ -43,6 +39,14 @@ use std::thread::{self, JoinHandle}; use crate::error::Error; +/// Check if verbose debug logging is enabled via FREEMKV_DEBUG env var. +fn debug_enabled() -> bool { + std::env::var("FREEMKV_DEBUG") + .ok() + .map(|v| v == "1" || v == "true" || v == "yes") + .unwrap_or(false) +} + /// Default channel depth for callers without a specific reason to /// pick another value. Kept conservative (4) — most callers should /// use READ_PIPELINE_DEPTH or WRITE_PIPELINE_DEPTH instead. @@ -149,23 +153,56 @@ impl Pipeline { let mut first_err: Option = None; let mut stopped = false; - while let Ok(item) = rx.recv() { - if first_err.is_some() || stopped { - // Drain remaining items so the producer never - // blocks on a dead receiver. `apply` is not - // called once we've decided to stop. - continue; - } - match sink.apply(item) { - Ok(Flow::Continue) => {} - Ok(Flow::Stop) => { - stopped = true; + while let Ok(item) = rx.recv() { + if debug_enabled() { + tracing::debug!( + "Pipeline receive: item={}", + std::any::type_name::() + ); } - Err(e) => { - first_err = Some(e); + + let apply_start = std::time::Instant::now(); + + if first_err.is_some() || stopped { + // Drain remaining items so the producer never + // blocks on a dead receiver. `apply` is not + // called once we've decided to stop. + continue; + } + match sink.apply(item) { + Ok(Flow::Continue) => {} + Ok(Flow::Stop) => { + stopped = true; + if debug_enabled() { + tracing::debug!("Pipeline: consumer returned Flow::Stop"); + } + } + Err(e) => { + first_err = Some(e); + if debug_enabled() { + tracing::debug!( + "Pipeline: apply error, stopping, err={:?}", + e + ); + } + } + } + + let apply_elapsed = apply_start.elapsed(); + if debug_enabled() && apply_elapsed > std::time::Duration::from_millis(100) { + tracing::debug!( + "Pipeline apply: took {:.2}s, item={}", + apply_elapsed.as_secs_f64(), + std::any::type_name::() + ); + } else if debug_enabled() { + tracing::debug!( + "Pipeline apply: OK in {:.3}ms, item={}", + apply_elapsed.as_micros(), + std::any::type_name::() + ); } } - } match first_err { Some(e) => Err(e), @@ -189,7 +226,35 @@ impl Pipeline { /// independent signal (e.g. `Halt`) — `send` alone is not the /// notification edge. pub fn send(&self, item: I) -> Result<(), I> { - self.tx.send(item).map_err(|e| e.0) + let start = std::time::Instant::now(); + match self.tx.send(item) { + Ok(()) => { + let elapsed = start.elapsed(); + if debug_enabled() && elapsed > std::time::Duration::from_millis(10) { + tracing::debug!( + "Pipeline send: blocked {:.2}s, item={}", + elapsed.as_secs_f64(), + std::any::type_name::() + ); + } else if debug_enabled() { + tracing::debug!("Pipeline send: OK in {:.3}ms", elapsed.as_micros()); + } + Ok(()) + } + Err(e) => { + let elapsed = start.elapsed(); + if debug_enabled() && elapsed > std::time::Duration::from_millis(10) { + tracing::debug!( + "Pipeline send: blocked {:.2}s before channel closed, item={}", + elapsed.as_secs_f64(), + std::any::type_name::() + ); + } else if debug_enabled() { + tracing::debug!("Pipeline send: failed after {:.3}ms", elapsed.as_micros()); + } + Err(e.0) + } + } } /// Non-blocking variant of [`Pipeline::send`]. If the channel is