Minor cleanups

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<Instant>,
+}
+
+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)));
+    }
+}

candumpr/src/lib.rs

@@ -1,4 +1,5 @@
 pub mod can;
+pub mod debounce;
 pub mod errframe;
 pub mod format;
 pub mod frame;

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::<std::io::Error>())
+        .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<bool>], names: &[String]) {
+///
+/// Returns true if the interface transitioned to down.
+fn handle_link_event(event: LinkEvent, link_up: &mut [Option<bool>], 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<bool>], 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;
         }

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}"
+    );
+}

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
 ```
 

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 <https://nexte.st>:
@@ -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