diff --git a/Cargo.toml b/Cargo.toml index 6ee6d5a..f58a38d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -30,3 +30,4 @@ tabled = "0.18" tempfile = "3.27.0" tracing = "0.1" tracing-subscriber = "0.3" +zstd = { version = "0.13.3", default-features = false } diff --git a/candumpr/Cargo.toml b/candumpr/Cargo.toml index 0c1f903..bcbe7a8 100644 --- a/candumpr/Cargo.toml +++ b/candumpr/Cargo.toml @@ -21,6 +21,7 @@ libc.workspace = true neli.workspace = true tracing.workspace = true tracing-subscriber.workspace = true +zstd.workspace = true [dev-dependencies] ctor.workspace = true diff --git a/candumpr/src/main.rs b/candumpr/src/main.rs index 4688b6e..389c541 100644 --- a/candumpr/src/main.rs +++ b/candumpr/src/main.rs @@ -109,10 +109,12 @@ enum Format { impl Format { /// Log file extension for each output format - fn ext(&self) -> &'static str { - match self { - Format::CandumpFile => "log", - Format::CandumpConsole => "txt", + fn ext(&self, compress: bool) -> &'static str { + match (self, compress) { + (Format::CandumpFile, false) => "log", + (Format::CandumpFile, true) => "log.zst", + (Format::CandumpConsole, false) => "txt", + (Format::CandumpConsole, true) => "txt.zst", } } } @@ -137,6 +139,10 @@ struct Cli { #[arg(long, value_enum, default_value = "candump-file")] format: Format, + /// Compress output with zstd. Requires --log or --output. + #[arg(long, short = 'c')] + compress: bool, + /// Timestamp rendering mode. Only applies to the candump formats. #[arg(long, value_enum, default_value = "absolute")] timestamp: TimestampMode, @@ -170,6 +176,23 @@ fn main() -> ExitCode { return ExitCode::FAILURE; } + // Compressing stdout would hand a terminal a binary stream, and the pipe case is served just as + // well by `candumpr can0 | zstd`. + if cli.compress && !cli.log && cli.output.is_none() { + tracing::error!("--compress requires --log or --output"); + return ExitCode::FAILURE; + } + if let Some(path) = &cli.output + && cli.compress + && path.extension().is_none_or(|ext| ext != "zst") + { + // File extensions are useful, but not required. Give a QoL warning. + tracing::warn!( + path = %path.display(), + "output is zstd-compressed but the path does not end in .zst" + ); + } + // The sockets vector defines the canonical interface ordering. The orderings of: // // 1. cli.interfaces @@ -253,7 +276,7 @@ fn main() -> ExitCode { .map(|interface| Output::Template { dir: ".".into(), interface: interface.clone(), - ext: cli.format.ext().to_string(), + ext: cli.format.ext(cli.compress).to_string(), next_index: 0, }) .collect() @@ -268,6 +291,7 @@ fn main() -> ExitCode { let formatter = make_formatter(); let mut config = SinkConfig::new(output); config.header = formatter.header().map(|h| h.to_vec()); + config.compress = cli.compress; (formatter, Sink::new(config)) }) .collect(); diff --git a/candumpr/src/sink.rs b/candumpr/src/sink.rs index de5acac..97951c9 100644 --- a/candumpr/src/sink.rs +++ b/candumpr/src/sink.rs @@ -3,7 +3,7 @@ use std::path::PathBuf; use std::time::{Duration, Instant}; use crate::recv::Timestamp; -use crate::writer::{FileWriter, StdoutWriter, Writer}; +use crate::writer::{FileWriter, StdoutWriter, Writer, ZstdWriter}; /// Where a [Sink] writes. pub enum Output { @@ -21,42 +21,6 @@ pub enum Output { }, } -impl Output { - /// Construct the writer for this output type. - fn open( - &self, - timestamp: Timestamp, - flush_threshold_bytes: usize, - ) -> std::io::Result> { - let path = match self { - Output::Stdout => return Ok(Box::new(StdoutWriter::new())), - Output::Path(path) => path.clone(), - Output::Template { - dir, - interface, - ext, - next_index, - } => dir.join(template_filename( - *next_index, - interface, - timestamp.sec, - ext, - )), - }; - if let Some(parent) = path.parent() - && !parent.as_os_str().is_empty() - { - std::fs::create_dir_all(parent)?; - } - let file = std::fs::File::create(&path)?; - tracing::info!(path = %path.display(), "created log file"); - Ok(Box::new(std::io::BufWriter::with_capacity( - flush_threshold_bytes, - FileWriter::new(file), - ))) - } -} - /// Configuration for a [Sink]. pub struct SinkConfig { pub output: Output, @@ -70,6 +34,8 @@ pub struct SinkConfig { pub sync_interval: Option, /// Whether activation failures that waiting could heal are retried, or should be fatal. pub retry_activation_failures: bool, + /// Compress file output with zstd. Ignored for [Output::Stdout]. + pub compress: bool, } impl SinkConfig { @@ -81,6 +47,42 @@ impl SinkConfig { flush_interval: Some(Duration::from_secs(5)), sync_interval: Some(Duration::from_secs(5 * 60)), retry_activation_failures: false, + compress: false, + } + } + + /// Construct the writer stack for this config's output. + fn open_writer(&self, timestamp: Timestamp) -> std::io::Result> { + let path = match &self.output { + // Unreachable with compression: main rejects --compress against stdout. + Output::Stdout => return Ok(Box::new(StdoutWriter::new())), + Output::Path(path) => path.clone(), + Output::Template { + dir, + interface, + ext, + next_index, + } => dir.join(template_filename( + *next_index, + interface, + timestamp.sec, + ext, + )), + }; + if let Some(parent) = path.parent() + && !parent.as_os_str().is_empty() + { + std::fs::create_dir_all(parent)?; + } + let file = FileWriter::new(std::fs::File::create(&path)?); + tracing::info!(path = %path.display(), "created log file"); + if self.compress { + Ok(Box::new(ZstdWriter::new(file)?)) + } else { + Ok(Box::new(std::io::BufWriter::with_capacity( + self.flush_threshold_bytes, + file, + ))) } } } @@ -151,11 +153,7 @@ impl Sink { self.state = SinkState::Pending { last_attempt: Some(Instant::now()), }; - let mut writer = match self - .config - .output - .open(timestamp, self.config.flush_threshold_bytes) - { + let mut writer = match self.config.open_writer(timestamp) { Ok(writer) => writer, Err(e) => { if classify(e.kind(), self.config.retry_activation_failures) diff --git a/candumpr/src/writer/mod.rs b/candumpr/src/writer/mod.rs index 6baccc6..d3c802d 100644 --- a/candumpr/src/writer/mod.rs +++ b/candumpr/src/writer/mod.rs @@ -1,10 +1,12 @@ mod file; mod stdout; +mod zstd; use std::io::Write; pub use file::FileWriter; pub use stdout::StdoutWriter; +pub use zstd::ZstdWriter; /// Writes formatted frame data to an output destination. pub trait Writer: std::io::Write { diff --git a/candumpr/src/writer/zstd.rs b/candumpr/src/writer/zstd.rs new file mode 100644 index 0000000..a22e87f --- /dev/null +++ b/candumpr/src/writer/zstd.rs @@ -0,0 +1,184 @@ +use std::io::Write; + +use zstd::stream::raw::CParameter; + +use super::Writer; + +// TODO: Tune these given real sample data +const LEVEL: i32 = 1; +const WINDOW_LOG: u32 = 15; + +/// Compresses formatted frames with streaming zstd, one zstd frame per file. +pub struct ZstdWriter { + // Write to a Vec instead of the inner Writer so that we can always write chunks of + // compressed data evenly divisible by CanFrames (no partial frames are written). We do this by + // consuming from the Vec only upon flush(), which is only ever called at a Frame boundary. + encoder: zstd::stream::write::Encoder<'static, Vec>, + inner: W, +} + +impl ZstdWriter { + pub fn new(inner: W) -> std::io::Result { + Self::with_params(inner, LEVEL, WINDOW_LOG) + } + + pub fn with_params(inner: W, level: i32, window_log: u32) -> std::io::Result { + let mut encoder = zstd::stream::raw::Encoder::new(level)?; + encoder.set_parameter(CParameter::WindowLog(window_log))?; + Ok(Self { + encoder: zstd::stream::write::Encoder::with_encoder(Vec::new(), encoder), + inner, + }) + } +} + +impl std::io::Write for ZstdWriter { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.encoder.write(buf) + } + + /// Close the current block and flush + /// + /// The [ZstdWriter] does not write to disk outside of [ZstdWriter::flush] in order to guarantee + /// that a CAN frame is never partially written across two compressed chunks. + fn flush(&mut self) -> std::io::Result<()> { + self.encoder.flush()?; + let buf = self.encoder.get_mut(); + // We can't use Write::write_all, because in the case of a partial write it does not + // indicate how much was written. So we mirror the implementation of BufWriter::flush_buf + // which lets us retain unwritten bytes and retry. If we dropped bytes on write failure, + // even if we started writing bytes again afterwards, everything after that point would be + // corrupted. + let mut written = 0; + let result = loop { + if written == buf.len() { + break Ok(()); + } + match self.inner.write(&buf[written..]) { + Ok(0) => { + break Err(std::io::Error::new( + std::io::ErrorKind::WriteZero, + "failed to write compressed data", + )); + } + Ok(n) => written += n, + Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {} + Err(e) => break Err(e), + } + }; + buf.drain(..written); + result?; + self.inner.flush() + } +} + +impl Writer for ZstdWriter { + fn sync(&mut self) -> std::io::Result<()> { + self.flush()?; + self.inner.sync() + } + + /// Finish writing. + /// + /// Flushes internal buffers, and writes the zstd frame epilogue. Do not write additional data + /// after calling finish(). + fn finish(&mut self) -> std::io::Result<()> { + self.encoder.do_finish()?; + self.flush()?; + self.inner.finish() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::recv::receiver::BATCH_CAPACITY; + use crate::writer::FileWriter; + + /// Formatted candump frames for `range`, grouped the way a [Sink](crate::sink::Sink) writes + /// them: one `Vec` per receive batch. + fn batches(range: std::ops::Range) -> Vec> { + range + .map(|i| format!("(1732117385.{:06}) vcan0 123#{i:08X}\n", i % 1_000_000)) + .collect::>() + .chunks(BATCH_CAPACITY) + .map(|batch| batch.concat().into_bytes()) + .collect() + } + + /// The inner [Writer] under a [ZstdWriter] under test. + /// + /// Keeps what it accepts, and counts `write` calls, which is how many times compressed bytes + /// left the [ZstdWriter]. + #[derive(Default)] + struct WriteCountingWriter { + bytes: Vec, + writes: usize, + /// When set, the first `write` accepts only this many bytes and the second fails with EIO + flaky: Option, + } + + impl std::io::Write for WriteCountingWriter { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.writes += 1; + let accepted = match (self.flaky, self.writes) { + (Some(n), 1) => n.min(buf.len()), + (Some(_), 2) => return Err(std::io::Error::from_raw_os_error(libc::EIO)), + _ => buf.len(), + }; + self.bytes.extend_from_slice(&buf[..accepted]); + Ok(accepted) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl Writer for WriteCountingWriter { + fn sync(&mut self) -> std::io::Result<()> { + Ok(()) + } + + fn finish(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + // happy path + #[test] + fn round_trip_decodes_to_input() { + let file = tempfile::NamedTempFile::new().unwrap(); + let batches = batches(0..1000); + + let mut writer = ZstdWriter::new(FileWriter::new(file.reopen().unwrap())).unwrap(); + for batch in &batches { + writer.write_all(batch).unwrap(); + } + writer.finish().unwrap(); + + let raw = std::fs::read(file.path()).unwrap(); + assert_eq!(zstd::decode_all(&raw[..]).unwrap(), batches.concat()); + } + + // recovery after write errors shouldn't result in something that can't decompress + #[test] + fn partial_write_retains_the_remainder() { + let mut writer = ZstdWriter::new(WriteCountingWriter { + flaky: Some(100), + ..WriteCountingWriter::default() + }) + .unwrap(); + let input = batches(0..1000).concat(); + writer.write_all(&input).unwrap(); + + let err = writer.flush().unwrap_err(); + assert_eq!(err.raw_os_error(), Some(libc::EIO)); + assert_eq!(writer.inner.bytes.len(), 100, "kept only what was accepted"); + + // Retrying recovers everything: the bytes the inner writer refused are still buffered. + // Dropping them instead would leave a hole that costs every frame after it. + writer.finish().unwrap(); + assert_eq!(zstd::decode_all(&writer.inner.bytes[..]).unwrap(), input); + } +} diff --git a/candumpr/tests/compress.rs b/candumpr/tests/compress.rs new file mode 100644 index 0000000..0caea61 --- /dev/null +++ b/candumpr/tests/compress.rs @@ -0,0 +1,162 @@ +use std::os::unix::io::AsFd; +use std::path::{Path, PathBuf}; +use std::process::ExitStatus; +use std::time::{Duration, Instant}; + +use candumpr::can::{self, LinuxCanFrame}; +use pretty_assertions::assert_eq; +use vcan_fixture::VcanHarness; +use vcan_fixture::prelude::*; + +#[ctor::ctor] +fn setup() { + tracing_subscriber::fmt() + .with_test_writer() + .with_ansi(true) + .init(); + vcan_fixture::enter_namespace(); +} + +/// Decompress with the zstd CLI, which is the whole point of using a standard format. +/// +/// Returns the status as well as the output so that we can handle the case where a log that was +/// never closed decodes completely but exits nonzero (all windows written, but no frame epilogue). +fn zstd_d(path: &Path) -> (ExitStatus, Vec) { + let out = std::process::Command::new("zstd") + .arg("-dc") + .arg(path) + .output() + .unwrap(); + eprint!("{}", String::from_utf8_lossy(&out.stderr)); + (out.status, out.stdout) +} + +/// The single log file candumpr is writing in `dir`, once it has created one. +fn log_file(dir: &Path) -> Option { + std::fs::read_dir(dir) + .unwrap() + .next() + .map(|entry| entry.unwrap().path()) +} + +#[test] +#[cfg_attr(feature = "ci", ignore = "requires vcan")] +fn logs_compressed_to_a_zst_file() { + let vcans = VcanHarness::new(1).unwrap(); + let iface = &vcans.names()[0]; + let dir = tempfile::TempDir::new().unwrap(); + + let child = tool!("candumpr") + .args(["-l", "--compress", "--timestamp", "zero"]) + .arg(iface) + .current_dir(dir.path()) + .spawn_piped() + .unwrap(); + + // Give enough time for candumpr to create its socket and start receiving + std::thread::sleep(Duration::from_millis(200)); + + let tx = can::open_can_raw_blocking(iface).unwrap(); + can::send_frame(tx.as_fd(), &LinuxCanFrame::new(0x123, &[0xAB])).unwrap(); + + // The file is created by the same write that takes the frame, and the clean shutdown below + // writes everything out, so its mere existence is enough to signal on. + let deadline = Instant::now() + Duration::from_secs(5); + let path = loop { + if let Some(path) = log_file(dir.path()) { + break path; + } + assert!(Instant::now() < deadline, "candumpr created no log file"); + std::thread::sleep(Duration::from_millis(20)); + }; + child.signal(libc::SIGTERM).unwrap(); + let output = child.captured_output().unwrap(); + assert!( + output.status.success(), + "expected a clean exit, got {}", + output.status + ); + + let name = path.file_name().unwrap().to_str().unwrap(); + let prefix = format!("i0000_{iface}_"); + assert!( + name.starts_with(&prefix) && name.ends_with(".log.zst"), + "expected a {prefix}*.log.zst file, got {name}" + ); + + // A cleanly closed log has its zstd epilogue, so the CLI is happy. + let (status, stdout) = zstd_d(&path); + assert!(status.success(), "zstd -d exited {status}"); + assert_eq!( + String::from_utf8(stdout).unwrap(), + format!("(000.000000) {iface} 123#AB\n") + ); +} + +#[test] +#[cfg_attr(feature = "ci", ignore = "requires vcan")] +fn sigkill_leaves_a_decodable_prefix() { + let vcans = VcanHarness::new(1).unwrap(); + let iface = &vcans.names()[0]; + let dir = tempfile::TempDir::new().unwrap(); + + let child = tool!("candumpr") + .args(["-l", "--compress"]) + .arg(iface) + .current_dir(dir.path()) + .spawn_piped() + .unwrap(); + + // Give enough time for candumpr to create its socket and start receiving + std::thread::sleep(Duration::from_millis(200)); + + // Keep traffic flowing and kill the moment the first bytes reach the file, so that the kill + // lands in the middle of a zstd block rather than after one. + let tx = can::open_can_raw_blocking(iface).unwrap(); + let mut sent = 0u32; + let path = loop { + for _ in 0..50 { + can::send_frame(tx.as_fd(), &LinuxCanFrame::new(0x123, &sent.to_be_bytes())).unwrap(); + sent += 1; + } + if let Some(path) = log_file(dir.path()) + && path.metadata().unwrap().len() > 0 + { + break path; + } + assert!( + sent < 50_000, + "no bytes reached the log after {sent} frames" + ); + std::thread::sleep(Duration::from_millis(25)); + }; + child.signal(libc::SIGKILL).unwrap(); + let _ = child.captured_output(); + + let (status, stdout) = zstd_d(&path); + // No epilogue was ever written, so the CLI reports the stream ended early even though it + // handed us everything that was in it. + assert!( + !status.success(), + "a log that was never closed does not decode cleanly; got status: {status}" + ); + + let text = String::from_utf8(stdout).unwrap(); + assert!(!text.is_empty(), "recovered nothing from {path:?}"); + // ZstdWriter only writes bytes to the FileWriter on ZstdWriter::flush(), which only ever + // happens between frames. This is to ensure that partial frames are never written outside of + // power loss. + assert!( + text.ends_with('\n'), + "recovered a partial frame, ending {:?}", + &text[text.len().saturating_sub(60)..] + ); + for (i, line) in text.lines().enumerate() { + let counter = line.rsplit('#').next().unwrap(); + assert_eq!( + u32::from_str_radix(counter, 16).unwrap(), + i as u32, + "line {i} is out of sequence: {line}" + ); + } +}