Add recorder: raw CAN and GNSS sessions with failsafes and a storage bound
Reads SocketCAN and gpsd directly. Sessions start on the first frame or on 'ctl start', end after 15 s of bus idle or on 'ctl stop'; blips under 3 s are discarded. candump -L chunks fdatasync'd every second and zstd'd once closed; oldest sessions deleted to stay under 50 GB / above 20 GB free. Kernel and queue drops, link changes and clock steps are recorded as events. Bench-tested on vcan with the first car capture (byte-identical) and installed on the car Orin as yukart-recorder.service. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DDwfd6Ki8UfG4hK6W5YpJz
This commit is contained in:
@@ -0,0 +1,177 @@
|
||||
//! Raw SocketCAN reception: kernel timestamps, the kernel's own drop counter,
|
||||
//! and error frames. Read-only — the socket is never written to.
|
||||
|
||||
use std::ffi::CString;
|
||||
use std::fmt::Write as _;
|
||||
use std::io;
|
||||
use std::mem::{size_of, zeroed};
|
||||
use std::time::Duration;
|
||||
|
||||
pub struct Frame {
|
||||
/// CLOCK_REALTIME at reception in the driver, from SO_TIMESTAMP.
|
||||
pub sec: i64,
|
||||
pub usec: i64,
|
||||
/// As delivered: carries the EFF / RTR / ERR flag bits.
|
||||
pub id: u32,
|
||||
pub len: u8,
|
||||
pub data: [u8; 8],
|
||||
}
|
||||
|
||||
impl Frame {
|
||||
/// One `candump -L` line, without the newline, so captures stay replayable
|
||||
/// with `canplayer` and readable by anything already written against them.
|
||||
pub fn write_candump(&self, iface: &str, out: &mut String) {
|
||||
let _ = write!(out, "({}.{:06}) {} ", self.sec, self.usec, iface);
|
||||
if self.id & libc::CAN_ERR_FLAG != 0 {
|
||||
let _ = write!(out, "{:08X}", self.id & (libc::CAN_ERR_MASK | libc::CAN_ERR_FLAG));
|
||||
} else if self.id & libc::CAN_EFF_FLAG != 0 {
|
||||
let _ = write!(out, "{:08X}", self.id & libc::CAN_EFF_MASK);
|
||||
} else {
|
||||
let _ = write!(out, "{:03X}", self.id & libc::CAN_SFF_MASK);
|
||||
}
|
||||
out.push('#');
|
||||
if self.id & libc::CAN_RTR_FLAG != 0 {
|
||||
out.push('R');
|
||||
} else {
|
||||
for b in &self.data[..self.len.min(8) as usize] {
|
||||
let _ = write!(out, "{:02X}", b);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub enum Recv {
|
||||
/// A frame, and how many frames the kernel dropped since the previous one.
|
||||
Frame(Frame, u32),
|
||||
Timeout,
|
||||
}
|
||||
|
||||
pub struct CanSocket {
|
||||
fd: i32,
|
||||
kernel_drops: u32,
|
||||
}
|
||||
|
||||
impl Drop for CanSocket {
|
||||
fn drop(&mut self) {
|
||||
unsafe { libc::close(self.fd) };
|
||||
}
|
||||
}
|
||||
|
||||
fn setopt<T>(fd: i32, level: i32, name: i32, val: &T) -> io::Result<()> {
|
||||
let r = unsafe {
|
||||
libc::setsockopt(fd, level, name, val as *const T as *const libc::c_void, size_of::<T>() as u32)
|
||||
};
|
||||
if r < 0 { Err(io::Error::last_os_error()) } else { Ok(()) }
|
||||
}
|
||||
|
||||
impl CanSocket {
|
||||
/// Fails with NotFound-like errors while the adapter is unplugged; callers retry.
|
||||
pub fn open(iface: &str, timeout: Duration) -> io::Result<Self> {
|
||||
let name = CString::new(iface).map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?;
|
||||
let idx = unsafe { libc::if_nametoindex(name.as_ptr()) };
|
||||
if idx == 0 {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
let fd = unsafe { libc::socket(libc::PF_CAN, libc::SOCK_RAW | libc::SOCK_CLOEXEC, libc::CAN_RAW) };
|
||||
if fd < 0 {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
let sock = CanSocket { fd, kernel_drops: 0 };
|
||||
|
||||
// Error frames are evidence (bus-off, wrong bitrate, a dying adapter), so record them all.
|
||||
setopt(fd, libc::SOL_CAN_RAW, libc::CAN_RAW_ERR_FILTER, &libc::CAN_ERR_MASK)?;
|
||||
setopt(fd, libc::SOL_SOCKET, libc::SO_TIMESTAMP, &1i32)?;
|
||||
setopt(fd, libc::SOL_SOCKET, libc::SO_RXQ_OVFL, &1i32)?;
|
||||
// Clamped to net.core.rmem_max; best effort.
|
||||
let _ = setopt(fd, libc::SOL_SOCKET, libc::SO_RCVBUF, &(4i32 << 20));
|
||||
let tv = libc::timeval { tv_sec: timeout.as_secs() as _, tv_usec: timeout.subsec_micros() as _ };
|
||||
setopt(fd, libc::SOL_SOCKET, libc::SO_RCVTIMEO, &tv)?;
|
||||
|
||||
let mut addr: libc::sockaddr_can = unsafe { zeroed() };
|
||||
addr.can_family = libc::AF_CAN as u16;
|
||||
addr.can_ifindex = idx as i32;
|
||||
let r = unsafe {
|
||||
libc::bind(fd, &addr as *const _ as *const libc::sockaddr, size_of::<libc::sockaddr_can>() as u32)
|
||||
};
|
||||
if r < 0 {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
Ok(sock)
|
||||
}
|
||||
|
||||
pub fn recv(&mut self) -> io::Result<Recv> {
|
||||
let mut frame: libc::can_frame = unsafe { zeroed() };
|
||||
let mut iov = libc::iovec {
|
||||
iov_base: &mut frame as *mut _ as *mut libc::c_void,
|
||||
iov_len: size_of::<libc::can_frame>(),
|
||||
};
|
||||
let mut cbuf = [0u64; 16];
|
||||
let mut msg: libc::msghdr = unsafe { zeroed() };
|
||||
msg.msg_iov = &mut iov;
|
||||
msg.msg_iovlen = 1;
|
||||
msg.msg_control = cbuf.as_mut_ptr() as *mut libc::c_void;
|
||||
msg.msg_controllen = size_of::<[u64; 16]>() as _;
|
||||
|
||||
let n = unsafe { libc::recvmsg(self.fd, &mut msg, 0) };
|
||||
if n < 0 {
|
||||
let e = io::Error::last_os_error();
|
||||
return match e.kind() {
|
||||
io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut => Ok(Recv::Timeout),
|
||||
io::ErrorKind::Interrupted => Ok(Recv::Timeout),
|
||||
_ => Err(e),
|
||||
};
|
||||
}
|
||||
|
||||
let (mut sec, mut usec) = (None, 0);
|
||||
let mut drops = self.kernel_drops;
|
||||
unsafe {
|
||||
let mut c = libc::CMSG_FIRSTHDR(&msg);
|
||||
while !c.is_null() {
|
||||
let (level, ty) = ((*c).cmsg_level, (*c).cmsg_type);
|
||||
if level == libc::SOL_SOCKET && ty == libc::SCM_TIMESTAMP {
|
||||
let tv = std::ptr::read_unaligned(libc::CMSG_DATA(c) as *const libc::timeval);
|
||||
sec = Some(tv.tv_sec as i64);
|
||||
usec = tv.tv_usec as i64;
|
||||
} else if level == libc::SOL_SOCKET && ty == libc::SO_RXQ_OVFL {
|
||||
drops = std::ptr::read_unaligned(libc::CMSG_DATA(c) as *const u32);
|
||||
}
|
||||
c = libc::CMSG_NXTHDR(&msg, c);
|
||||
}
|
||||
}
|
||||
let sec = sec.unwrap_or_else(|| {
|
||||
let t = crate::clock::realtime();
|
||||
usec = ((t.fract()) * 1e6) as i64;
|
||||
t.trunc() as i64
|
||||
});
|
||||
let new_drops = drops.wrapping_sub(self.kernel_drops);
|
||||
self.kernel_drops = drops;
|
||||
|
||||
Ok(Recv::Frame(
|
||||
Frame { sec, usec, id: frame.can_id, len: frame.can_dlc, data: frame.data },
|
||||
new_drops,
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn line(id: u32, data: &[u8]) -> String {
|
||||
let mut d = [0u8; 8];
|
||||
d[..data.len()].copy_from_slice(data);
|
||||
let mut s = String::new();
|
||||
Frame { sec: 1791531456, usec: 435801, id, len: data.len() as u8, data: d }.write_candump("can_pcan", &mut s);
|
||||
s
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn candump_format_matches_existing_captures() {
|
||||
assert_eq!(line(0x449, &[0x6F, 0x5F, 0x6E, 0x71, 0xFF, 0xFD, 0x33, 0xFF]),
|
||||
"(1791531456.435801) can_pcan 449#6F5F6E71FFFD33FF");
|
||||
assert_eq!(line(0x0C0, &[0x6D, 0xFF, 0, 7, 0x0C]), "(1791531456.435801) can_pcan 0C0#6DFF00070C");
|
||||
assert_eq!(line(0x18DAF110 | libc::CAN_EFF_FLAG, &[1]), "(1791531456.435801) can_pcan 18DAF110#01");
|
||||
assert_eq!(line(0x7DF | libc::CAN_RTR_FLAG, &[]), "(1791531456.435801) can_pcan 7DF#R");
|
||||
assert_eq!(line(0x04 | libc::CAN_ERR_FLAG, &[0; 8]), "(1791531456.435801) can_pcan 20000004#0000000000000000");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
//! The car unit has no RTC battery: the clock boots stale and is stepped later by
|
||||
//! chrony (NTP or GNSS). Recordings therefore carry enough to correct timestamps
|
||||
//! afterwards — sync state at start, monotonic time, and every step observed.
|
||||
|
||||
use serde_json::{json, Value};
|
||||
|
||||
fn clock(id: libc::clockid_t) -> f64 {
|
||||
let mut ts: libc::timespec = unsafe { std::mem::zeroed() };
|
||||
unsafe { libc::clock_gettime(id, &mut ts) };
|
||||
ts.tv_sec as f64 + ts.tv_nsec as f64 * 1e-9
|
||||
}
|
||||
|
||||
pub fn realtime() -> f64 {
|
||||
clock(libc::CLOCK_REALTIME)
|
||||
}
|
||||
|
||||
pub fn monotonic() -> f64 {
|
||||
clock(libc::CLOCK_MONOTONIC)
|
||||
}
|
||||
|
||||
/// realtime − monotonic. Constant while the clock only slews; a jump means a step.
|
||||
pub fn offset() -> f64 {
|
||||
realtime() - monotonic()
|
||||
}
|
||||
|
||||
pub fn boot_id() -> String {
|
||||
std::fs::read_to_string("/proc/sys/kernel/random/boot_id").map(|s| s.trim().to_string()).unwrap_or_default()
|
||||
}
|
||||
|
||||
/// Kernel view of synchronisation, read without privilege (adjtimex with modes = 0).
|
||||
pub fn sync_state() -> Value {
|
||||
let mut tx: libc::timex = unsafe { std::mem::zeroed() };
|
||||
let state = unsafe { libc::adjtimex(&mut tx) };
|
||||
let unsync = state == libc::TIME_ERROR || tx.status & libc::STA_UNSYNC != 0;
|
||||
json!({ "synced": !unsync, "maxerror_us": tx.maxerror, "esterror_us": tx.esterror })
|
||||
}
|
||||
|
||||
pub fn hostname() -> String {
|
||||
std::fs::read_to_string("/proc/sys/kernel/hostname").map(|s| s.trim().to_string()).unwrap_or_default()
|
||||
}
|
||||
@@ -0,0 +1,428 @@
|
||||
//! recorder — persists raw CAN and GNSS to disk, session by session.
|
||||
//!
|
||||
//! Reads SocketCAN and gpsd directly, not through the broker: capture is tier 1 and
|
||||
//! must not depend on anything that can be late or down.
|
||||
//!
|
||||
//! Sessions start when CAN frames arrive (auto) or on `ctl start` (manual), and an
|
||||
//! auto session ends after `--idle` seconds without frames: ignition off, adapter
|
||||
//! unplugged, or plugged in somewhere with no bus. An adapter with no traffic never
|
||||
//! creates a session, so leaving the PEAK attached at a desk produces nothing.
|
||||
|
||||
mod can;
|
||||
mod clock;
|
||||
mod session;
|
||||
mod sources;
|
||||
mod storage;
|
||||
|
||||
use std::io::{Read, Write};
|
||||
use std::os::unix::net::UnixStream;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::process::ExitCode;
|
||||
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
||||
use std::sync::mpsc::{sync_channel, RecvTimeoutError};
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use serde_json::json;
|
||||
|
||||
use session::{Limits, Session};
|
||||
use sources::Event;
|
||||
use storage::Bound;
|
||||
|
||||
const USAGE: &str = "\
|
||||
usage:
|
||||
recorder run [options]
|
||||
--iface NAME CAN interface (can_pcan)
|
||||
--root DIR recordings directory (/var/lib/yukart/recordings)
|
||||
--gpsd HOST:PORT gpsd, '' to disable (127.0.0.1:2947)
|
||||
--ctl PATH control socket (/run/yukart-recorder/ctl.sock)
|
||||
--idle SECS end an auto session after this long without frames (15)
|
||||
--min-secs SECS discard auto sessions shorter than this with no marks (3)
|
||||
--chunk-secs SECS rotate the CAN log after this long (300)
|
||||
--chunk-mb MB ... or this size (64)
|
||||
--cap-gb GB total size of all recordings (50)
|
||||
--min-free-gb GB keep at least this much free on the filesystem (20)
|
||||
--no-auto only record on 'ctl start'
|
||||
recorder ctl [--ctl PATH] COMMAND
|
||||
status | start [LABEL] | stop | mark TEXT | auto on|off";
|
||||
|
||||
struct Config {
|
||||
iface: String,
|
||||
root: PathBuf,
|
||||
gpsd: String,
|
||||
ctl: String,
|
||||
idle: Duration,
|
||||
min_secs: u64,
|
||||
limits: Limits,
|
||||
bound: Bound,
|
||||
auto: bool,
|
||||
}
|
||||
|
||||
fn parse_run(args: &[String]) -> Result<Config, String> {
|
||||
let mut c = Config {
|
||||
iface: "can_pcan".into(),
|
||||
root: "/var/lib/yukart/recordings".into(),
|
||||
gpsd: "127.0.0.1:2947".into(),
|
||||
ctl: "/run/yukart-recorder/ctl.sock".into(),
|
||||
idle: Duration::from_secs(15),
|
||||
min_secs: 3,
|
||||
limits: Limits { chunk_secs: 300, chunk_bytes: 64 << 20 },
|
||||
bound: Bound { cap_bytes: 50 << 30, min_free_bytes: 20 << 30 },
|
||||
auto: true,
|
||||
};
|
||||
let mut it = args.iter();
|
||||
while let Some(a) = it.next() {
|
||||
if a == "--no-auto" {
|
||||
c.auto = false;
|
||||
continue;
|
||||
}
|
||||
let v = it.next().ok_or(format!("{a} needs a value"))?;
|
||||
let num = || v.parse::<u64>().map_err(|_| format!("{a}: not a number: {v}"));
|
||||
match a.as_str() {
|
||||
"--iface" => c.iface = v.clone(),
|
||||
"--root" => c.root = v.into(),
|
||||
"--gpsd" => c.gpsd = v.clone(),
|
||||
"--ctl" => c.ctl = v.clone(),
|
||||
"--idle" => c.idle = Duration::from_secs(num()?),
|
||||
"--min-secs" => c.min_secs = num()?,
|
||||
"--chunk-secs" => c.limits.chunk_secs = num()?,
|
||||
"--chunk-mb" => c.limits.chunk_bytes = num()? << 20,
|
||||
"--cap-gb" => c.bound.cap_bytes = num()? << 30,
|
||||
"--min-free-gb" => c.bound.min_free_bytes = num()? << 30,
|
||||
_ => return Err(format!("unknown option {a}")),
|
||||
}
|
||||
}
|
||||
Ok(c)
|
||||
}
|
||||
|
||||
fn ctl_client(args: &[String]) -> ExitCode {
|
||||
let mut path = "/run/yukart-recorder/ctl.sock".to_string();
|
||||
let mut rest = args;
|
||||
if rest.first().map(String::as_str) == Some("--ctl") && rest.len() >= 2 {
|
||||
path = rest[1].clone();
|
||||
rest = &rest[2..];
|
||||
}
|
||||
if rest.is_empty() {
|
||||
eprintln!("{USAGE}");
|
||||
return ExitCode::from(2);
|
||||
}
|
||||
let run = || -> std::io::Result<String> {
|
||||
let mut s = UnixStream::connect(&path)?;
|
||||
s.set_read_timeout(Some(Duration::from_secs(6)))?;
|
||||
s.write_all(rest.join(" ").as_bytes())?;
|
||||
s.shutdown(std::net::Shutdown::Write)?;
|
||||
let mut out = String::new();
|
||||
s.read_to_string(&mut out)?;
|
||||
Ok(out)
|
||||
};
|
||||
match run() {
|
||||
Ok(out) => {
|
||||
print!("{out}");
|
||||
if out.starts_with("error") { ExitCode::from(1) } else { ExitCode::SUCCESS }
|
||||
}
|
||||
Err(e) => {
|
||||
eprintln!("recorder ctl: {path}: {e}");
|
||||
ExitCode::from(1)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct Recorder {
|
||||
cfg: Config,
|
||||
session: Option<Session>,
|
||||
compress: std::sync::mpsc::Sender<PathBuf>,
|
||||
auto: bool,
|
||||
/// After a manual stop, auto-start waits for the bus to go idle first,
|
||||
/// otherwise it would reopen a session immediately.
|
||||
suppressed: bool,
|
||||
last_frame: Option<Instant>,
|
||||
last_tpv: Option<(f64, String)>,
|
||||
can_link: (bool, String),
|
||||
gnss_link: (bool, String),
|
||||
kernel_drops: u64,
|
||||
queue_drops: Arc<AtomicU64>,
|
||||
queue_drops_seen: u64,
|
||||
}
|
||||
|
||||
impl Recorder {
|
||||
fn start(&mut self, trigger: &'static str, label: Option<String>) -> Result<(), String> {
|
||||
let deleted = storage::enforce(&self.cfg.root, &self.cfg.bound, None);
|
||||
let seq = storage::next_seq(&self.cfg.root);
|
||||
let mut s = Session::open(&self.cfg.root, seq, trigger, label, &self.cfg.iface, self.compress.clone())
|
||||
.map_err(|e| format!("cannot open session: {e}"))?;
|
||||
eprintln!("session {} started ({trigger})", s.name);
|
||||
if !deleted.is_empty() {
|
||||
let _ = s.event("storage_bound", json!({ "deleted": deleted }));
|
||||
}
|
||||
let _ = s.event("can_link", json!({ "up": self.can_link.0, "detail": self.can_link.1 }));
|
||||
let _ = s.event("gnss_link", json!({ "up": self.gnss_link.0, "detail": self.gnss_link.1 }));
|
||||
if let Some((rx, line)) = &self.last_tpv {
|
||||
let _ = s.gnss(*rx, line, true);
|
||||
}
|
||||
self.session = Some(s);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn stop(&mut self, reason: &str) {
|
||||
let Some(s) = self.session.take() else { return };
|
||||
let short = s.trigger == "auto" && s.marks == 0 && s.bus_span().as_secs() < self.cfg.min_secs;
|
||||
let name = s.name.clone();
|
||||
if short {
|
||||
// A wake-up blip (door unlock, a brief plug-in) is not worth a session.
|
||||
match s.discard() {
|
||||
Ok(()) => eprintln!("session {name} discarded: under {}s of traffic, no marks", self.cfg.min_secs),
|
||||
Err(e) => eprintln!("session {name} discard failed: {e}"),
|
||||
}
|
||||
return;
|
||||
}
|
||||
match s.close(reason) {
|
||||
Ok(_) => eprintln!("session {name} closed: {reason}"),
|
||||
Err(e) => eprintln!("session {name} close failed: {e}"),
|
||||
}
|
||||
}
|
||||
|
||||
fn ctl(&mut self, cmd: &str) -> String {
|
||||
let (verb, arg) = cmd.split_once(' ').map(|(v, a)| (v, a.trim())).unwrap_or((cmd, ""));
|
||||
match verb {
|
||||
"status" => {
|
||||
json!({
|
||||
"auto": self.auto,
|
||||
"suppressed": self.suppressed,
|
||||
"session": self.session.as_ref().map(Session::status),
|
||||
"can": { "up": self.can_link.0, "detail": self.can_link.1,
|
||||
"last_frame_age_s": self.last_frame.map(|t| t.elapsed().as_secs_f64()) },
|
||||
"gnss": { "up": self.gnss_link.0, "detail": self.gnss_link.1 },
|
||||
"storage": { "root": self.cfg.root, "used_bytes": storage::usage(&self.cfg.root),
|
||||
"free_bytes": storage::free_bytes(&self.cfg.root),
|
||||
"cap_bytes": self.cfg.bound.cap_bytes },
|
||||
"clock": clock::sync_state(),
|
||||
})
|
||||
.to_string()
|
||||
}
|
||||
"start" => {
|
||||
let label = (!arg.is_empty()).then(|| arg.to_string());
|
||||
if let Some(s) = &mut self.session {
|
||||
// Promote an auto session: it now ends only on 'stop'.
|
||||
s.trigger = "manual";
|
||||
let _ = s.event("promoted", json!({ "label": label }));
|
||||
return format!("ok: {} now manual", s.name);
|
||||
}
|
||||
match self.start("manual", label) {
|
||||
Ok(()) => format!("ok: {}", self.session.as_ref().unwrap().name),
|
||||
Err(e) => format!("error: {e}"),
|
||||
}
|
||||
}
|
||||
"stop" => match &self.session {
|
||||
Some(s) => {
|
||||
let name = s.name.clone();
|
||||
self.stop("manual stop");
|
||||
self.suppressed = true;
|
||||
format!("ok: {name} closed")
|
||||
}
|
||||
None => "ok: no session".into(),
|
||||
},
|
||||
"mark" => match &mut self.session {
|
||||
Some(s) if !arg.is_empty() => {
|
||||
let _ = s.event("mark", json!({ "text": arg }));
|
||||
format!("ok: mark {} in {}", s.marks, s.name)
|
||||
}
|
||||
Some(_) => "error: mark needs text".into(),
|
||||
None => "error: no session; 'start' first".into(),
|
||||
},
|
||||
"auto" => match arg {
|
||||
"on" => { self.auto = true; "ok: auto on".into() }
|
||||
"off" => { self.auto = false; "ok: auto off".into() }
|
||||
_ => "error: auto on|off".into(),
|
||||
},
|
||||
_ => format!("error: unknown command '{verb}'"),
|
||||
}
|
||||
}
|
||||
|
||||
fn handle(&mut self, ev: Event) {
|
||||
match ev {
|
||||
Event::Can(f) => {
|
||||
self.last_frame = Some(Instant::now());
|
||||
if self.session.is_none() && self.auto && !self.suppressed {
|
||||
if let Err(e) = self.start("auto", None) {
|
||||
eprintln!("{e}");
|
||||
}
|
||||
}
|
||||
if let Some(s) = &mut self.session {
|
||||
if let Err(e) = s.frame(&f) {
|
||||
eprintln!("write: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
Event::KernelDrops(n) => {
|
||||
self.kernel_drops += n as u64;
|
||||
if let Some(s) = &mut self.session {
|
||||
s.kernel_drops += n as u64;
|
||||
let _ = s.event("kernel_drops", json!({ "frames": n }));
|
||||
}
|
||||
}
|
||||
Event::CanLink(up, detail) => {
|
||||
eprintln!("can: {detail}");
|
||||
self.can_link = (up, detail.clone());
|
||||
if let Some(s) = &mut self.session {
|
||||
let _ = s.event("can_link", json!({ "up": up, "detail": detail }));
|
||||
}
|
||||
}
|
||||
Event::Gnss(rx, line) => {
|
||||
if line.contains("\"class\":\"TPV\"") {
|
||||
self.last_tpv = Some((rx, line.clone()));
|
||||
}
|
||||
if let Some(s) = &mut self.session {
|
||||
let _ = s.gnss(rx, &line, false);
|
||||
}
|
||||
}
|
||||
Event::GnssLink(up, detail) => {
|
||||
eprintln!("gnss: {detail}");
|
||||
self.gnss_link = (up, detail.clone());
|
||||
if let Some(s) = &mut self.session {
|
||||
let _ = s.event("gnss_link", json!({ "up": up, "detail": detail }));
|
||||
}
|
||||
}
|
||||
Event::Ctl(cmd, reply) => {
|
||||
let out = self.ctl(&cmd);
|
||||
let _ = reply.send(out);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn bus_idle(&self) -> bool {
|
||||
self.last_frame.map(|t| t.elapsed() >= self.cfg.idle).unwrap_or(true)
|
||||
}
|
||||
}
|
||||
|
||||
fn run(cfg: Config) -> ExitCode {
|
||||
if let Err(e) = std::fs::create_dir_all(&cfg.root) {
|
||||
eprintln!("recorder: {}: {e}", cfg.root.display());
|
||||
return ExitCode::from(1);
|
||||
}
|
||||
let compress = storage::spawn_compressor();
|
||||
for p in storage::recover(&cfg.root) {
|
||||
let _ = compress.send(p);
|
||||
}
|
||||
|
||||
let stop = Arc::new(AtomicBool::new(false));
|
||||
for sig in [signal_hook::consts::SIGTERM, signal_hook::consts::SIGINT] {
|
||||
let _ = signal_hook::flag::register(sig, stop.clone());
|
||||
}
|
||||
|
||||
// ~100 s of frames at the measured 890 frames/s.
|
||||
let (tx, rx) = sync_channel::<Event>(100_000);
|
||||
let queue_drops = Arc::new(AtomicU64::new(0));
|
||||
sources::spawn_can(cfg.iface.clone(), tx.clone(), queue_drops.clone(), stop.clone());
|
||||
if !cfg.gpsd.is_empty() {
|
||||
sources::spawn_gnss(cfg.gpsd.clone(), tx.clone(), stop.clone());
|
||||
}
|
||||
if let Some(dir) = Path::new(&cfg.ctl).parent() {
|
||||
let _ = std::fs::create_dir_all(dir);
|
||||
}
|
||||
if let Err(e) = sources::spawn_ctl(cfg.ctl.clone(), tx.clone()) {
|
||||
// Manual control is lost, auto capture is not: keep going.
|
||||
eprintln!("recorder: control socket {}: {e}", cfg.ctl);
|
||||
}
|
||||
drop(tx);
|
||||
|
||||
eprintln!("recorder {} on {} → {}", env!("CARGO_PKG_VERSION"), cfg.iface, cfg.root.display());
|
||||
let mut r = Recorder {
|
||||
auto: cfg.auto,
|
||||
cfg,
|
||||
session: None,
|
||||
compress,
|
||||
suppressed: false,
|
||||
last_frame: None,
|
||||
last_tpv: None,
|
||||
can_link: (false, "not yet opened".into()),
|
||||
gnss_link: (false, "not yet connected".into()),
|
||||
kernel_drops: 0,
|
||||
queue_drops,
|
||||
queue_drops_seen: 0,
|
||||
};
|
||||
|
||||
let mut last_sync = Instant::now();
|
||||
let mut last_bound = Instant::now();
|
||||
let mut clock_ref = clock::offset();
|
||||
loop {
|
||||
match rx.recv_timeout(Duration::from_millis(200)) {
|
||||
Ok(ev) => r.handle(ev),
|
||||
Err(RecvTimeoutError::Timeout) => {}
|
||||
Err(RecvTimeoutError::Disconnected) => break,
|
||||
}
|
||||
if stop.load(Ordering::Relaxed) {
|
||||
// Drain what is already queued so the last frames before shutdown are kept.
|
||||
while let Ok(ev) = rx.try_recv() {
|
||||
if !matches!(ev, Event::Ctl(..)) {
|
||||
r.handle(ev);
|
||||
}
|
||||
}
|
||||
r.stop("shutdown");
|
||||
break;
|
||||
}
|
||||
if r.bus_idle() {
|
||||
r.suppressed = false;
|
||||
if r.session.as_ref().map(|s| s.trigger == "auto").unwrap_or(false) {
|
||||
r.stop("bus idle");
|
||||
}
|
||||
}
|
||||
if last_sync.elapsed() >= Duration::from_secs(1) {
|
||||
last_sync = Instant::now();
|
||||
let offset = clock::offset();
|
||||
let step = offset - clock_ref;
|
||||
clock_ref = offset;
|
||||
let qd = r.queue_drops.load(Ordering::Relaxed);
|
||||
let new_qd = qd - r.queue_drops_seen;
|
||||
r.queue_drops_seen = qd;
|
||||
if let Some(s) = &mut r.session {
|
||||
if step.abs() > 0.05 {
|
||||
// chrony stepped the clock: everything before this is off by `step`.
|
||||
let _ = s.event("clock_step", json!({ "step_s": step, "clock": clock::sync_state() }));
|
||||
}
|
||||
if new_qd > 0 {
|
||||
s.queue_drops += new_qd;
|
||||
let _ = s.event("queue_drops", json!({ "frames": new_qd }));
|
||||
}
|
||||
if let Err(e) = s.sync() {
|
||||
eprintln!("sync: {e}");
|
||||
}
|
||||
match s.maybe_rotate(&r.cfg.limits) {
|
||||
Ok(true) => last_bound = Instant::now() - Duration::from_secs(3600),
|
||||
Ok(false) => {}
|
||||
Err(e) => eprintln!("rotate: {e}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
if last_bound.elapsed() >= Duration::from_secs(60) {
|
||||
last_bound = Instant::now();
|
||||
let active = r.session.as_ref().map(|s| s.dir.clone());
|
||||
let deleted = storage::enforce(&r.cfg.root, &r.cfg.bound, active.as_deref());
|
||||
if !deleted.is_empty() {
|
||||
eprintln!("storage bound: deleted {deleted:?}");
|
||||
if let Some(s) = &mut r.session {
|
||||
let _ = s.event("storage_bound", json!({ "deleted": deleted }));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
eprintln!("recorder: stopped");
|
||||
ExitCode::SUCCESS
|
||||
}
|
||||
|
||||
fn main() -> ExitCode {
|
||||
let args: Vec<String> = std::env::args().skip(1).collect();
|
||||
match args.first().map(String::as_str) {
|
||||
Some("run") => match parse_run(&args[1..]) {
|
||||
Ok(cfg) => run(cfg),
|
||||
Err(e) => {
|
||||
eprintln!("recorder: {e}\n{USAGE}");
|
||||
ExitCode::from(2)
|
||||
}
|
||||
},
|
||||
Some("ctl") => ctl_client(&args[1..]),
|
||||
_ => {
|
||||
eprintln!("{USAGE}");
|
||||
ExitCode::from(2)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,260 @@
|
||||
//! One recording session: a directory of append-only files.
|
||||
//!
|
||||
//! ```text
|
||||
//! 000042-20261009T163736/
|
||||
//! session.json metadata; "status": open → closed (or truncated, if found open at startup)
|
||||
//! can-0001.log.zst candump -L text, closed chunks compressed
|
||||
//! can-0002.log the open chunk, plain text, fdatasync'd every second
|
||||
//! gnss.ndjson every gpsd report, with the local receive time
|
||||
//! events.ndjson marks, clock steps, link changes, drops
|
||||
//! ```
|
||||
//!
|
||||
//! Durability: power loss costs at most the last second of the open chunk. Closed
|
||||
//! chunks are compressed to a temp file and renamed, so a crash mid-compression
|
||||
//! leaves the plain chunk intact.
|
||||
|
||||
use std::fs::{self, File, OpenOptions};
|
||||
use std::io::{self, BufWriter, Write};
|
||||
use std::os::unix::io::AsRawFd;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::mpsc::Sender;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use serde_json::{json, Value};
|
||||
|
||||
use crate::can::Frame;
|
||||
use crate::clock;
|
||||
|
||||
pub struct Limits {
|
||||
pub chunk_secs: u64,
|
||||
pub chunk_bytes: u64,
|
||||
}
|
||||
|
||||
pub struct Session {
|
||||
pub dir: PathBuf,
|
||||
pub name: String,
|
||||
pub trigger: &'static str,
|
||||
iface: String,
|
||||
meta: Value,
|
||||
started: Instant,
|
||||
chunk_idx: u32,
|
||||
chunk_started: Instant,
|
||||
chunk_bytes: u64,
|
||||
can: BufWriter<File>,
|
||||
gnss: BufWriter<File>,
|
||||
events: BufWriter<File>,
|
||||
line: String,
|
||||
pub frames: u64,
|
||||
first_frame: Option<Instant>,
|
||||
last_frame: Option<Instant>,
|
||||
pub marks: u32,
|
||||
pub kernel_drops: u64,
|
||||
pub queue_drops: u64,
|
||||
compress: Sender<PathBuf>,
|
||||
}
|
||||
|
||||
fn append(path: &Path) -> io::Result<BufWriter<File>> {
|
||||
Ok(BufWriter::with_capacity(256 << 10, OpenOptions::new().create(true).append(true).open(path)?))
|
||||
}
|
||||
|
||||
fn sync(w: &mut BufWriter<File>) -> io::Result<()> {
|
||||
w.flush()?;
|
||||
w.get_ref().sync_data()
|
||||
}
|
||||
|
||||
pub fn sync_dir(dir: &Path) {
|
||||
if let Ok(d) = File::open(dir) {
|
||||
unsafe { libc::fsync(d.as_raw_fd()) };
|
||||
}
|
||||
}
|
||||
|
||||
/// Atomic replace: readers see the old or the new file, never half of one.
|
||||
pub fn write_json(path: &Path, v: &Value) -> io::Result<()> {
|
||||
let tmp = path.with_extension("json.tmp");
|
||||
let mut f = File::create(&tmp)?;
|
||||
f.write_all(serde_json::to_string_pretty(v)?.as_bytes())?;
|
||||
f.write_all(b"\n")?;
|
||||
f.sync_all()?;
|
||||
fs::rename(&tmp, path)?;
|
||||
if let Some(d) = path.parent() {
|
||||
sync_dir(d);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn chunk_name(i: u32) -> String {
|
||||
format!("can-{i:04}.log")
|
||||
}
|
||||
|
||||
impl Session {
|
||||
pub fn open(root: &Path, seq: u64, trigger: &'static str, label: Option<String>, iface: &str,
|
||||
compress: Sender<PathBuf>) -> io::Result<Self> {
|
||||
let rt = clock::realtime();
|
||||
let stamp = {
|
||||
let t = rt as libc::time_t;
|
||||
let mut tm: libc::tm = unsafe { std::mem::zeroed() };
|
||||
unsafe { libc::localtime_r(&t, &mut tm) };
|
||||
format!("{:04}{:02}{:02}T{:02}{:02}{:02}", tm.tm_year + 1900, tm.tm_mon + 1, tm.tm_mday,
|
||||
tm.tm_hour, tm.tm_min, tm.tm_sec)
|
||||
};
|
||||
// The sequence number orders sessions; the timestamp is for humans and may be stale.
|
||||
let name = format!("{seq:06}-{stamp}");
|
||||
let dir = root.join(&name);
|
||||
fs::create_dir_all(&dir)?;
|
||||
let meta = json!({
|
||||
"format": 1,
|
||||
"name": name,
|
||||
"seq": seq,
|
||||
"status": "open",
|
||||
"trigger": trigger,
|
||||
"label": label,
|
||||
"host": clock::hostname(),
|
||||
"boot_id": clock::boot_id(),
|
||||
"iface": iface,
|
||||
"recorder": env!("CARGO_PKG_VERSION"),
|
||||
"start": { "realtime": rt, "monotonic": clock::monotonic(), "clock": clock::sync_state() },
|
||||
});
|
||||
write_json(&dir.join("session.json"), &meta)?;
|
||||
let s = Session {
|
||||
can: append(&dir.join(chunk_name(1)))?,
|
||||
gnss: append(&dir.join("gnss.ndjson"))?,
|
||||
events: append(&dir.join("events.ndjson"))?,
|
||||
dir, name, trigger, meta,
|
||||
iface: iface.to_string(),
|
||||
started: Instant::now(),
|
||||
chunk_idx: 1,
|
||||
chunk_started: Instant::now(),
|
||||
chunk_bytes: 0,
|
||||
line: String::with_capacity(64),
|
||||
frames: 0, first_frame: None, last_frame: None, marks: 0, kernel_drops: 0, queue_drops: 0,
|
||||
compress,
|
||||
};
|
||||
sync_dir(root);
|
||||
Ok(s)
|
||||
}
|
||||
|
||||
pub fn frame(&mut self, f: &Frame) -> io::Result<()> {
|
||||
self.line.clear();
|
||||
f.write_candump(&self.iface, &mut self.line);
|
||||
self.line.push('\n');
|
||||
self.can.write_all(self.line.as_bytes())?;
|
||||
self.chunk_bytes += self.line.len() as u64;
|
||||
self.frames += 1;
|
||||
let now = Instant::now();
|
||||
self.first_frame.get_or_insert(now);
|
||||
self.last_frame = Some(now);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn gnss(&mut self, rx: f64, raw: &str, stale: bool) -> io::Result<()> {
|
||||
let msg = serde_json::from_str::<Value>(raw).unwrap_or_else(|_| Value::String(raw.to_string()));
|
||||
let mut v = json!({ "rx": rx, "msg": msg });
|
||||
if stale {
|
||||
// The last report seen before this session began, so a session starts with a position.
|
||||
v["before_start"] = json!(true);
|
||||
}
|
||||
writeln!(self.gnss, "{v}")
|
||||
}
|
||||
|
||||
pub fn event(&mut self, kind: &str, mut extra: Value) -> io::Result<()> {
|
||||
if !extra.is_object() {
|
||||
extra = json!({});
|
||||
}
|
||||
extra["rt"] = json!(clock::realtime());
|
||||
extra["mono"] = json!(clock::monotonic());
|
||||
extra["kind"] = json!(kind);
|
||||
if kind == "mark" {
|
||||
self.marks += 1;
|
||||
}
|
||||
writeln!(self.events, "{extra}")
|
||||
}
|
||||
|
||||
/// First to last frame: the trailing idle wait that ended the session does not count.
|
||||
pub fn bus_span(&self) -> Duration {
|
||||
match (self.first_frame, self.last_frame) {
|
||||
(Some(a), Some(b)) => b - a,
|
||||
_ => Duration::ZERO,
|
||||
}
|
||||
}
|
||||
|
||||
/// Called about once a second: makes everything so far survive power loss.
|
||||
pub fn sync(&mut self) -> io::Result<()> {
|
||||
sync(&mut self.can)?;
|
||||
sync(&mut self.gnss)?;
|
||||
sync(&mut self.events)
|
||||
}
|
||||
|
||||
/// Returns true when a chunk was closed (a good moment to enforce the storage bound).
|
||||
pub fn maybe_rotate(&mut self, lim: &Limits) -> io::Result<bool> {
|
||||
if self.chunk_bytes == 0
|
||||
|| (self.chunk_started.elapsed().as_secs() < lim.chunk_secs && self.chunk_bytes < lim.chunk_bytes)
|
||||
{
|
||||
return Ok(false);
|
||||
}
|
||||
sync(&mut self.can)?;
|
||||
let closed = self.dir.join(chunk_name(self.chunk_idx));
|
||||
self.chunk_idx += 1;
|
||||
self.can = append(&self.dir.join(chunk_name(self.chunk_idx)))?;
|
||||
sync_dir(&self.dir);
|
||||
self.chunk_started = Instant::now();
|
||||
self.chunk_bytes = 0;
|
||||
let _ = self.compress.send(closed);
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
pub fn close(mut self, reason: &str) -> io::Result<PathBuf> {
|
||||
self.sync()?;
|
||||
let last = self.dir.join(chunk_name(self.chunk_idx));
|
||||
drop(self.can);
|
||||
if self.chunk_bytes == 0 {
|
||||
let _ = fs::remove_file(&last);
|
||||
} else {
|
||||
let _ = self.compress.send(last);
|
||||
}
|
||||
self.meta["status"] = json!("closed");
|
||||
self.meta["end"] = json!({
|
||||
"realtime": clock::realtime(), "monotonic": clock::monotonic(),
|
||||
"reason": reason, "clock": clock::sync_state(),
|
||||
});
|
||||
self.meta["frames"] = json!(self.frames);
|
||||
self.meta["chunks"] = json!(if self.chunk_bytes == 0 { self.chunk_idx - 1 } else { self.chunk_idx });
|
||||
self.meta["marks"] = json!(self.marks);
|
||||
self.meta["drops"] = json!({ "kernel": self.kernel_drops, "recorder_queue": self.queue_drops });
|
||||
write_json(&self.dir.join("session.json"), &self.meta)?;
|
||||
Ok(self.dir)
|
||||
}
|
||||
|
||||
/// Drop the session entirely, without queueing its chunk for compression.
|
||||
pub fn discard(self) -> io::Result<()> {
|
||||
let dir = self.dir.clone();
|
||||
drop(self);
|
||||
fs::remove_dir_all(dir)
|
||||
}
|
||||
|
||||
pub fn status(&self) -> Value {
|
||||
json!({
|
||||
"name": self.name, "trigger": self.trigger, "frames": self.frames,
|
||||
"seconds": self.started.elapsed().as_secs(), "chunk": self.chunk_idx, "marks": self.marks,
|
||||
"drops": { "kernel": self.kernel_drops, "recorder_queue": self.queue_drops },
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// Compresses closed chunks off the writer thread: `x.log` → `x.log.zst.tmp` → `x.log.zst`, then
|
||||
/// removes `x.log`. Every step leaves at least one complete copy on disk.
|
||||
pub fn compress_chunk(plain: &Path) -> io::Result<()> {
|
||||
let zst = plain.with_extension("log.zst");
|
||||
let tmp = plain.with_extension("log.zst.tmp");
|
||||
{
|
||||
let mut input = File::open(plain)?;
|
||||
let out = File::create(&tmp)?;
|
||||
let mut enc = zstd::stream::Encoder::new(out, 9)?;
|
||||
io::copy(&mut input, &mut enc)?;
|
||||
enc.finish()?.sync_all()?;
|
||||
}
|
||||
fs::rename(&tmp, &zst)?;
|
||||
if let Some(d) = plain.parent() {
|
||||
sync_dir(d);
|
||||
}
|
||||
fs::remove_file(plain)
|
||||
}
|
||||
@@ -0,0 +1,149 @@
|
||||
//! Producer threads. Each owns one input, reconnects on its own, and never blocks
|
||||
//! on the writer: CAN frames that do not fit the queue are counted, not waited for.
|
||||
|
||||
use std::io::{BufRead, BufReader, Read, Write};
|
||||
use std::net::{TcpStream, ToSocketAddrs};
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
use std::os::unix::net::UnixListener;
|
||||
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
||||
use std::sync::mpsc::{self, SyncSender, TrySendError};
|
||||
use std::sync::Arc;
|
||||
use std::thread;
|
||||
use std::time::Duration;
|
||||
|
||||
use crate::can::{CanSocket, Frame, Recv};
|
||||
use crate::clock;
|
||||
|
||||
pub enum Event {
|
||||
Can(Frame),
|
||||
/// Frames the kernel dropped before we read them (SO_RXQ_OVFL).
|
||||
KernelDrops(u32),
|
||||
CanLink(bool, String),
|
||||
Gnss(f64, String),
|
||||
GnssLink(bool, String),
|
||||
Ctl(String, mpsc::Sender<String>),
|
||||
}
|
||||
|
||||
pub fn spawn_can(iface: String, tx: SyncSender<Event>, queue_drops: Arc<AtomicU64>, stop: Arc<AtomicBool>) {
|
||||
thread::Builder::new().name("can".into()).spawn(move || {
|
||||
let mut reported: Option<String> = None;
|
||||
while !stop.load(Ordering::Relaxed) {
|
||||
let mut sock = match CanSocket::open(&iface, Duration::from_millis(200)) {
|
||||
Ok(s) => s,
|
||||
Err(e) => {
|
||||
// An unplugged adapter is a normal state; say so once, then poll quietly.
|
||||
let msg = format!("{iface}: {e}");
|
||||
if reported.as_deref() != Some(&msg) {
|
||||
let _ = tx.send(Event::CanLink(false, msg.clone()));
|
||||
reported = Some(msg);
|
||||
}
|
||||
thread::sleep(Duration::from_secs(1));
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let _ = tx.send(Event::CanLink(true, format!("{iface}: socket open")));
|
||||
reported = None;
|
||||
while !stop.load(Ordering::Relaxed) {
|
||||
match sock.recv() {
|
||||
Ok(Recv::Timeout) => {}
|
||||
Ok(Recv::Frame(f, drops)) => {
|
||||
if drops > 0 {
|
||||
let _ = tx.try_send(Event::KernelDrops(drops));
|
||||
}
|
||||
match tx.try_send(Event::Can(f)) {
|
||||
Ok(()) => {}
|
||||
Err(TrySendError::Full(_)) => {
|
||||
queue_drops.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
Err(TrySendError::Disconnected(_)) => return,
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
// ENETDOWN / ENODEV: the link went down or the adapter left.
|
||||
let _ = tx.send(Event::CanLink(false, format!("{iface}: {e}")));
|
||||
reported = Some(String::new());
|
||||
thread::sleep(Duration::from_millis(500));
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}).expect("spawn can thread");
|
||||
}
|
||||
|
||||
pub fn spawn_gnss(addr: String, tx: SyncSender<Event>, stop: Arc<AtomicBool>) {
|
||||
thread::Builder::new().name("gnss".into()).spawn(move || {
|
||||
// Report a failing gpsd once, not every retry.
|
||||
let mut reported_down = false;
|
||||
while !stop.load(Ordering::Relaxed) {
|
||||
let conn = addr
|
||||
.to_socket_addrs()
|
||||
.ok()
|
||||
.and_then(|mut a| a.next())
|
||||
.ok_or_else(|| std::io::Error::other("unresolvable"))
|
||||
.and_then(|a| TcpStream::connect_timeout(&a, Duration::from_secs(2)));
|
||||
let mut stream = match conn {
|
||||
Ok(s) => s,
|
||||
Err(e) => {
|
||||
if !reported_down {
|
||||
let _ = tx.send(Event::GnssLink(false, format!("gpsd {addr}: {e}")));
|
||||
reported_down = true;
|
||||
}
|
||||
thread::sleep(Duration::from_secs(3));
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let _ = stream.set_read_timeout(Some(Duration::from_secs(1)));
|
||||
if stream.write_all(b"?WATCH={\"enable\":true,\"json\":true};\n").is_err() {
|
||||
continue;
|
||||
}
|
||||
let _ = tx.send(Event::GnssLink(true, format!("gpsd {addr}: watching")));
|
||||
let mut reader = BufReader::new(stream);
|
||||
let mut line = String::new();
|
||||
loop {
|
||||
if stop.load(Ordering::Relaxed) {
|
||||
return;
|
||||
}
|
||||
match reader.read_line(&mut line) {
|
||||
Ok(0) => break,
|
||||
Ok(_) => {
|
||||
if line.ends_with('\n') {
|
||||
let rx = clock::realtime();
|
||||
let _ = tx.send(Event::Gnss(rx, line.trim_end().to_string()));
|
||||
line.clear();
|
||||
}
|
||||
}
|
||||
Err(e) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => {}
|
||||
Err(_) => break,
|
||||
}
|
||||
}
|
||||
let _ = tx.send(Event::GnssLink(false, format!("gpsd {addr}: connection lost")));
|
||||
reported_down = true;
|
||||
thread::sleep(Duration::from_secs(1));
|
||||
}
|
||||
}).expect("spawn gnss thread");
|
||||
}
|
||||
|
||||
/// One command per connection, one reply line back.
|
||||
pub fn spawn_ctl(path: String, tx: SyncSender<Event>) -> std::io::Result<()> {
|
||||
let _ = std::fs::remove_file(&path);
|
||||
let listener = UnixListener::bind(&path)?;
|
||||
std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o660))?;
|
||||
thread::Builder::new().name("ctl".into()).spawn(move || {
|
||||
for conn in listener.incoming() {
|
||||
let Ok(mut conn) = conn else { continue };
|
||||
let _ = conn.set_read_timeout(Some(Duration::from_secs(2)));
|
||||
let mut buf = Vec::new();
|
||||
let _ = (&mut conn).take(4096).read_to_end(&mut buf);
|
||||
let cmd = String::from_utf8_lossy(&buf).trim().to_string();
|
||||
let (rtx, rrx) = mpsc::channel();
|
||||
if tx.send(Event::Ctl(cmd, rtx)).is_err() {
|
||||
return;
|
||||
}
|
||||
let reply = rrx.recv_timeout(Duration::from_secs(5)).unwrap_or_else(|_| "error: no reply".into());
|
||||
let _ = conn.write_all(reply.as_bytes());
|
||||
let _ = conn.write_all(b"\n");
|
||||
}
|
||||
})?;
|
||||
Ok(())
|
||||
}
|
||||
@@ -0,0 +1,134 @@
|
||||
//! The storage bound, enforced by the writer: oldest closed sessions go first, the
|
||||
//! session in progress never does. A full disk would stop capture silently; losing
|
||||
//! the oldest drive is recoverable.
|
||||
|
||||
use std::fs;
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
use serde_json::Value;
|
||||
|
||||
use crate::session::{compress_chunk, write_json};
|
||||
|
||||
pub struct Bound {
|
||||
pub cap_bytes: u64,
|
||||
pub min_free_bytes: u64,
|
||||
}
|
||||
|
||||
/// Session directories, oldest first. Ordered by sequence number, not by clock.
|
||||
pub fn sessions(root: &Path) -> Vec<(u64, PathBuf)> {
|
||||
let mut v: Vec<(u64, PathBuf)> = fs::read_dir(root)
|
||||
.into_iter()
|
||||
.flatten()
|
||||
.flatten()
|
||||
.filter(|e| e.file_type().map(|t| t.is_dir()).unwrap_or(false))
|
||||
.filter_map(|e| {
|
||||
let seq = e.file_name().to_str()?.split('-').next()?.parse().ok()?;
|
||||
Some((seq, e.path()))
|
||||
})
|
||||
.collect();
|
||||
v.sort();
|
||||
v
|
||||
}
|
||||
|
||||
/// Never reuses a number, even after the bound has deleted every session: copies
|
||||
/// taken off the car must not collide with later sessions of the same name.
|
||||
pub fn next_seq(root: &Path) -> u64 {
|
||||
let mark = root.join(".last-seq");
|
||||
let stored = fs::read_to_string(&mark).ok().and_then(|s| s.trim().parse().ok()).unwrap_or(0);
|
||||
let seen = sessions(root).last().map(|(s, _)| *s).unwrap_or(0);
|
||||
let next = stored.max(seen) + 1;
|
||||
let tmp = root.join(".last-seq.tmp");
|
||||
if fs::write(&tmp, format!("{next}\n")).is_ok() {
|
||||
let _ = fs::rename(&tmp, &mark);
|
||||
}
|
||||
next
|
||||
}
|
||||
|
||||
fn du(p: &Path) -> u64 {
|
||||
fs::read_dir(p)
|
||||
.into_iter()
|
||||
.flatten()
|
||||
.flatten()
|
||||
.map(|e| match e.file_type() {
|
||||
Ok(t) if t.is_dir() => du(&e.path()),
|
||||
_ => e.metadata().map(|m| m.len()).unwrap_or(0),
|
||||
})
|
||||
.sum()
|
||||
}
|
||||
|
||||
pub fn free_bytes(p: &Path) -> u64 {
|
||||
let c = std::ffi::CString::new(p.as_os_str().as_encoded_bytes()).unwrap_or_default();
|
||||
let mut s: libc::statvfs = unsafe { std::mem::zeroed() };
|
||||
if unsafe { libc::statvfs(c.as_ptr(), &mut s) } != 0 {
|
||||
return u64::MAX;
|
||||
}
|
||||
s.f_bavail as u64 * s.f_frsize as u64
|
||||
}
|
||||
|
||||
pub fn usage(root: &Path) -> u64 {
|
||||
du(root)
|
||||
}
|
||||
|
||||
/// Returns the names of sessions deleted.
|
||||
pub fn enforce(root: &Path, b: &Bound, active: Option<&Path>) -> Vec<String> {
|
||||
let mut deleted = Vec::new();
|
||||
let mut total = du(root);
|
||||
for (_, dir) in sessions(root) {
|
||||
if total <= b.cap_bytes && free_bytes(root) >= b.min_free_bytes {
|
||||
break;
|
||||
}
|
||||
if Some(dir.as_path()) == active {
|
||||
continue;
|
||||
}
|
||||
let size = du(&dir);
|
||||
if fs::remove_dir_all(&dir).is_ok() {
|
||||
total = total.saturating_sub(size);
|
||||
deleted.push(dir.file_name().unwrap_or_default().to_string_lossy().into_owned());
|
||||
}
|
||||
}
|
||||
deleted
|
||||
}
|
||||
|
||||
/// At startup no session is active, so anything still marked open was cut short
|
||||
/// (power loss, crash). Mark it, compress its chunks, and drop half-written temp files.
|
||||
pub fn recover(root: &Path) -> Vec<PathBuf> {
|
||||
let mut pending = Vec::new();
|
||||
for (_, dir) in sessions(root) {
|
||||
let meta_path = dir.join("session.json");
|
||||
if let Some(mut meta) = fs::read_to_string(&meta_path).ok().and_then(|s| serde_json::from_str::<Value>(&s).ok()) {
|
||||
if meta["status"] == "open" {
|
||||
meta["status"] = Value::from("truncated");
|
||||
let _ = write_json(&meta_path, &meta);
|
||||
}
|
||||
}
|
||||
for e in fs::read_dir(&dir).into_iter().flatten().flatten() {
|
||||
let p = e.path();
|
||||
let name = p.file_name().unwrap_or_default().to_string_lossy().into_owned();
|
||||
if name.ends_with(".tmp") {
|
||||
let _ = fs::remove_file(&p);
|
||||
} else if name.starts_with("can-") && name.ends_with(".log") {
|
||||
if e.metadata().map(|m| m.len() == 0).unwrap_or(false) {
|
||||
let _ = fs::remove_file(&p);
|
||||
} else {
|
||||
pending.push(p);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
pending
|
||||
}
|
||||
|
||||
pub fn spawn_compressor() -> std::sync::mpsc::Sender<PathBuf> {
|
||||
let (tx, rx) = std::sync::mpsc::channel::<PathBuf>();
|
||||
std::thread::Builder::new()
|
||||
.name("compress".into())
|
||||
.spawn(move || {
|
||||
for p in rx {
|
||||
if let Err(e) = compress_chunk(&p) {
|
||||
eprintln!("compress {}: {e} (plain chunk kept)", p.display());
|
||||
}
|
||||
}
|
||||
})
|
||||
.expect("spawn compressor");
|
||||
tx
|
||||
}
|
||||
Reference in New Issue
Block a user