diff --git a/candumpr/src/debounce.rs b/candumpr/src/debounce.rs new file mode 100644 index 0000000..cdfcfc5 --- /dev/null +++ b/candumpr/src/debounce.rs @@ -0,0 +1,66 @@ +use std::time::{Duration, Instant}; + +/// Coalesces a burst of events into a single action. +/// +/// Each trigger pushes the deadline back, so a burst of events results in one action after the +/// burst settles rather than one action per event. A steady stream of triggers spaced closer than +/// the window postpones the action indefinitely; callers that need an upper bound must enforce it +/// themselves. +pub struct Debounce { + window: Duration, + deadline: Option, +} + +impl Debounce { + /// Create an idle Debounce that fires `window` after the most recent trigger. + pub fn new(window: Duration) -> Self { + Self { + window, + deadline: None, + } + } + + /// Record an event at `now`, (re)scheduling the deadline for `now + window`. + pub fn trigger(&mut self, now: Instant) { + self.deadline = Some(now + self.window); + } + + /// Check if the deadline has passed as of `now`. + pub fn expired(&mut self, now: Instant) -> bool { + if self.deadline.is_some_and(|deadline| now >= deadline) { + self.deadline = None; + true + } else { + false + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn fires_once_after_window() { + let mut d = Debounce::new(Duration::from_millis(500)); + let t0 = Instant::now(); + assert!(!d.expired(t0), "must not fire without a trigger"); + d.trigger(t0); + assert!(!d.expired(t0 + Duration::from_millis(499))); + assert!(d.expired(t0 + Duration::from_millis(500))); + assert!( + !d.expired(t0 + Duration::from_millis(501)), + "must reset to idle after firing" + ); + } + + #[test] + fn retrigger_pushes_deadline_back() { + let mut d = Debounce::new(Duration::from_millis(500)); + let t0 = Instant::now(); + d.trigger(t0); + d.trigger(t0 + Duration::from_millis(400)); + assert!(!d.expired(t0 + Duration::from_millis(500))); + assert!(d.expired(t0 + Duration::from_millis(900))); + } +} diff --git a/candumpr/src/lib.rs b/candumpr/src/lib.rs index 2a97856..ac3b077 100644 --- a/candumpr/src/lib.rs +++ b/candumpr/src/lib.rs @@ -1,4 +1,5 @@ pub mod can; +pub mod debounce; pub mod errframe; pub mod format; pub mod frame; diff --git a/candumpr/src/main.rs b/candumpr/src/main.rs index e19af5b..d83fc4c 100644 --- a/candumpr/src/main.rs +++ b/candumpr/src/main.rs @@ -1,9 +1,10 @@ use std::os::unix::io::AsFd; use std::process::ExitCode; use std::sync::atomic::{AtomicBool, Ordering}; -use std::time::Duration; +use std::time::{Duration, Instant}; use candumpr::can; +use candumpr::debounce::Debounce; use candumpr::errframe::{BusState, ErrorFrame}; use candumpr::format::{CanutilsConsoleFormatter, CanutilsFileFormatter, Formatter, TimestampMode}; use candumpr::frame::CanFrame; @@ -21,14 +22,23 @@ extern "C" fn signal_handler(_sig: libc::c_int) { STOP.store(true, Ordering::Relaxed); } +/// True if any error in the chain is an EPIPE. +fn is_broken_pipe(err: &eyre::Report) -> bool { + err.chain() + .filter_map(|e| e.downcast_ref::()) + .any(|io| io.kind() == std::io::ErrorKind::BrokenPipe) +} + /// Log a link-state edge to stderr, ignoring repeats of the last observed state. -fn handle_link_event(event: LinkEvent, link_up: &mut [Option], names: &[String]) { +/// +/// Returns true if the interface transitioned to down. +fn handle_link_event(event: LinkEvent, link_up: &mut [Option], names: &[String]) -> bool { let (sock_id, up) = match event { LinkEvent::LinkUp { sock_id } => (sock_id, true), LinkEvent::LinkDown { sock_id } => (sock_id, false), }; if link_up[sock_id] == Some(up) { - return; + return false; } link_up[sock_id] = Some(up); let interface = &names[sock_id]; @@ -37,10 +47,14 @@ fn handle_link_event(event: LinkEvent, link_up: &mut [Option], names: &[St } else { tracing::warn!(interface = %interface, "interface link down"); } + !up } /// Log each error frame in `batch` at debug level, and log bus-state transitions (edges only). -fn log_error_frames(batch: &[CanFrame], bus_state: &mut [BusState], names: &[String]) { +/// +/// Returns true if any interface transitioned into bus-off. +fn log_error_frames(batch: &[CanFrame], bus_state: &mut [BusState], names: &[String]) -> bool { + let mut bus_off = false; for frame in batch { let Some(err) = ErrorFrame::parse(&frame.raw) else { continue; @@ -63,9 +77,13 @@ fn log_error_frames(batch: &[CanFrame], bus_state: &mut [BusState], names: &[Str BusState::ErrorWarning | BusState::ErrorPassive => { tracing::warn!(interface = %interface, "bus state {old} -> {new}") } - BusState::BusOff => tracing::error!(interface = %interface, "bus state {old} -> {new}"), + BusState::BusOff => { + tracing::error!(interface = %interface, "bus state {old} -> {new}"); + bus_off = true; + } } } + bus_off } /// Output format for received frames. @@ -195,12 +213,21 @@ fn main() -> ExitCode { // Every error sets `failed` so the process still exits nonzero. let mut failed = false; + // Debounce link state and bus state events. + let mut state_debounce = Debounce::new(Duration::from_millis(200)); + loop { select! { recv(full_rx) -> msg => match msg { Ok(mut batch) => { - log_error_frames(&batch, &mut bus_state, &names); + if log_error_frames(&batch, &mut bus_state, &names) { + state_debounce.trigger(Instant::now()); + } if let Err(e) = pipeline.write_batch(&batch) { + if is_broken_pipe(&e) { + tracing::debug!("output closed; shutting down"); + break; + } tracing::error!(error = ?e, "failed to write batch"); failed = true; } @@ -210,11 +237,22 @@ fn main() -> ExitCode { Err(_) => break, }, recv(event_rx) -> msg => match msg { - Ok(event) => handle_link_event(event, &mut link_up, &names), + Ok(event) => { + if handle_link_event(event, &mut link_up, &names) { + state_debounce.trigger(Instant::now()); + } + } Err(_) => break, }, default(Duration::from_millis(100)) => {} } + if state_debounce.expired(Instant::now()) { + tracing::debug!("syncing after link-down or bus-off"); + if let Err(e) = pipeline.sync() { + tracing::error!(error = ?e, "link-down or bus-off sync failed"); + failed = true; + } + } if let Err(e) = pipeline.tick() { tracing::error!(error = ?e, "periodic flush or sync failed"); failed = true; @@ -255,6 +293,9 @@ fn main() -> ExitCode { // Drain everything the receiver queued before it exited. while let Ok(mut batch) = full_rx.try_recv() { if let Err(e) = pipeline.write_batch(&batch) { + if is_broken_pipe(&e) { + break; + } tracing::error!(error = ?e, "failed to write batch during drain"); failed = true; } diff --git a/candumpr/tests/epipe.rs b/candumpr/tests/epipe.rs new file mode 100644 index 0000000..8331d31 --- /dev/null +++ b/candumpr/tests/epipe.rs @@ -0,0 +1,73 @@ +use std::io::{BufRead, BufReader}; +use std::os::unix::io::AsFd; +use std::process::{Command, Stdio}; +use std::time::{Duration, Instant}; + +use candumpr::can::{self, LinuxCanFrame}; +use vcan_fixture::VcanHarness; + +#[ctor::ctor] +fn setup() { + tracing_subscriber::fmt().with_test_writer().init(); + vcan_fixture::enter_namespace(); +} + +/// When the stdout consumer goes away (`candumpr can0 | head`), candumpr must notice the broken +/// pipe on its next write and exit cleanly on its own, without being signalled. +#[test] +#[cfg_attr(feature = "ci", ignore = "requires vcan")] +fn exits_cleanly_when_stdout_closes() { + let vcans = VcanHarness::new(1).unwrap(); + let iface = &vcans.names()[0]; + + let mut child = Command::new(env!("CARGO_BIN_EXE_candumpr")) + .arg(iface) + .arg("--log-level=INFO") // TRACE level fills up the stderr buf and prevents EPIPE error + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + + // Give the process time to set up io_uring and start receiving. + std::thread::sleep(Duration::from_millis(200)); + + let tx = can::open_can_raw_blocking(iface).unwrap(); + let frame = LinuxCanFrame::new(0x18FECA00 | libc::CAN_EFF_FLAG, &[0xAA, 0xBB]); + + // Prove the pipe is live: send one frame and read the line it produces. + let mut stdout = BufReader::new(child.stdout.take().unwrap()); + can::send_frame(tx.as_fd(), &frame).unwrap(); + let mut line = String::new(); + stdout.read_line(&mut line).unwrap(); + assert!( + line.ends_with('\n'), + "expected a full output line: {line:?}" + ); + + // Close the read end, like `head` exiting. The next batch write hits EPIPE. + drop(stdout); + can::send_frame(tx.as_fd(), &frame).unwrap(); + + let deadline = Instant::now() + Duration::from_secs(5); + let status = loop { + if let Some(status) = child.try_wait().unwrap() { + break status; + } + if Instant::now() >= deadline { + child.kill().unwrap(); + let output = child.wait_with_output().unwrap(); + panic!( + "candumpr did not exit after stdout closed. stderr:\n{}", + String::from_utf8_lossy(&output.stderr) + ); + } + std::thread::sleep(Duration::from_millis(50)); + }; + + let output = child.wait_with_output().unwrap(); + eprint!("{}", String::from_utf8_lossy(&output.stderr)); + assert!( + status.success(), + "expected clean exit after stdout closed, got {status}" + ); +} diff --git a/docs/design/candumpr/03-benchmarks.md b/docs/design/candumpr/03-benchmarks.md index d5b171c..f79c0da 100644 --- a/docs/design/candumpr/03-benchmarks.md +++ b/docs/design/candumpr/03-benchmarks.md @@ -24,7 +24,7 @@ each using three different benchmarks. These benchmarks can be run with ```sh -cargo install gungruan-runner +cargo install gungraun-runner cargo bench ``` diff --git a/docs/developer/quickstart.md b/docs/developer/quickstart.md index b32ffae..4db5aa3 100644 --- a/docs/developer/quickstart.md +++ b/docs/developer/quickstart.md @@ -32,6 +32,35 @@ sudo ip link add dev can0 type vcan sudo ip link set up can0 ``` +## Test dependencies + +Just building the project doesn't have any dependencies other than Cargo and a C toolchain. But +testing this project requires the `vcan` kernel module, and the ability to enter a user namespace. + +### Ubuntu 24.04 + +```sh +sudo apt-get install -y linux-modules-extra-"$(uname -r)" +sudo modprobe vcan +sudo sysctl kernel.apparmor_restrict_unprivileged_userns=0 +``` + +### Ubuntu 26.04 + +The `vcan` kernel module has been merged back into `linux-modules`, so there's no extra packages to +install. But in order to use the [vcan-fixture](/vcan-fixture/src/lib.rs) crate, you still need to + +```sh +sudo modprobe vcan +sudo sysctl kernel.apparmor_restrict_unprivileged_userns=0 +``` + +### Fedora 44 + +```sh +sudo modprobe vcan +``` + ## Tests Tests may be run either with `cargo test` or : @@ -57,9 +86,8 @@ provides several features: * `bench::start_cpu_load(num, percent)` - starts `num` threads doing a PWM-like busyloop to hit `percent` CPU usage -It's assumed that the local developer environment has the necessary vcan kernel module. In CI, we -attempt to install the vcan module, but can skip the vcan-dependent tests with a warning if it's not -available. +In CI, we attempt to install the vcan module, but can skip the vcan-dependent tests with a warning +if it's not available. ### ASAN