crates/nornir-trace/src/wire.rs
//! **A run's events, compact on the wire.** A million events as JSON is ~60 MB;
//! as this columnar-ish varint stream ~5 MB, so a filing is a few chunks of a few
//! MB under gRPC's message limit.
//!
//! ```text
//! "NTR1" · varint n · n × (varint thread · varint Δt_ns · varint func · u8 kind)
//! ```
//!
//! `Δt_ns` is from the previous event of the chunk (the first: from 0, so a chunk
//! stands alone); `seq_no` is implicit — the chunk's base plus the event's place.
use anyhow::{bail, Result};
use crate::{EventKind, TraceEvent};
const MAGIC: &[u8; 4] = b"NTR1";
fn put(out: &mut Vec<u8>, mut v: u64) {
loop {
let b = (v & 0x7f) as u8;
v >>= 7;
if v == 0 {
out.push(b);
return;
}
out.push(b | 0x80);
}
}
fn get(b: &[u8], o: &mut usize) -> Result<u64> {
let mut v = 0u64;
let mut shift = 0;
loop {
let Some(&c) = b.get(*o) else { bail!("a trace chunk ends inside a number") };
*o += 1;
if shift >= 64 {
bail!("a trace chunk holds an over-long number");
}
v |= ((c & 0x7f) as u64) << shift;
if c & 0x80 == 0 {
return Ok(v);
}
shift += 7;
}
}
/// Encode events (in time order) as one chunk.
pub fn encode(events: &[TraceEvent]) -> Vec<u8> {
let mut out = Vec::with_capacity(8 + events.len() * 6);
out.extend_from_slice(MAGIC);
put(&mut out, events.len() as u64);
let mut prev = 0u64;
for e in events {
put(&mut out, e.thread as u64);
put(&mut out, e.t_ns.saturating_sub(prev));
put(&mut out, e.func as u64);
out.push(match e.kind {
EventKind::Enter => 0,
EventKind::Exit => 1,
});
prev = e.t_ns;
}
out
}
/// Decode one chunk whose first event is the run's `base_seq`-th.
pub fn decode(b: &[u8], base_seq: u64) -> Result<Vec<TraceEvent>> {
if b.len() < 4 || &b[..4] != MAGIC {
bail!("not a trace chunk (no NTR1 magic)");
}
let mut o = 4;
let n = get(b, &mut o)?;
if n > (b.len() as u64) {
bail!("a trace chunk claims {n} events in {} bytes", b.len());
}
let mut out = Vec::with_capacity(n as usize);
let mut t = 0u64;
for i in 0..n {
let thread = get(b, &mut o)? as u32;
t += get(b, &mut o)?;
let func = get(b, &mut o)? as u32;
let Some(&k) = b.get(o) else { bail!("a trace chunk ends inside an event") };
o += 1;
let kind = match k {
0 => EventKind::Enter,
1 => EventKind::Exit,
_ => bail!("a trace chunk holds event kind {k}"),
};
out.push(TraceEvent { thread, seq_no: base_seq + i, t_ns: t, func, kind });
}
Ok(out)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_chunk_round_trips_and_is_small() {
let events: Vec<TraceEvent> = (0..10_000u64)
.map(|i| TraceEvent {
thread: 4000 + (i % 3) as u32,
seq_no: 500 + i,
t_ns: 1_000_000 + i * 37,
func: (i % 250) as u32,
kind: if i % 2 == 0 { EventKind::Enter } else { EventKind::Exit },
})
.collect();
let b = encode(&events);
assert!(b.len() < events.len() * 7, "{} bytes for {} events", b.len(), events.len());
assert_eq!(decode(&b, 500).unwrap(), events);
assert!(decode(&b[..b.len() - 1], 500).is_err());
assert!(decode(b"JSON", 0).is_err());
}
}
// ── a filing: one self-standing chunk of a run ─────────────────────────────────
use serde::{Deserialize, Serialize};
use crate::{TraceRun, TracedBinary};
/// A function as a filing names it: its symbol and where its code is, the file
/// RELATIVE to the repo it lives in (`repo` = the workspace member the file is
/// under; `""` = outside every checkout — std, a registry crate).
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
pub struct FiledFn {
pub symbol: String,
#[serde(default)]
pub repo: String,
#[serde(default)]
pub file: String,
#[serde(default)]
pub line: u32,
}
/// What travels beside one chunk of events (`trace_json` of a `trace` filing).
/// A chunk stands alone: its events index ITS `functions`.
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
pub struct ChunkFiling {
pub chunk: u32,
pub chunks: u32,
/// The run-wide sequence number of the chunk's first event.
pub base_seq: u64,
/// What ran (`test code_room`, `bin dwarves --help`).
pub target: String,
pub functions: Vec<FiledFn>,
// run-wide facts, the same on every chunk
pub binaries: Vec<TracedBinary>,
pub total_events: u64,
pub truncated: bool,
pub threads: u32,
pub duration_ns: u64,
}
/// One chunk ready to file: its facts and its encoded events.
#[derive(Debug, Clone)]
pub struct Chunk {
pub filing: ChunkFiling,
pub events: Vec<u8>,
}
/// **Cut a run into self-standing chunks** of at most `per_chunk` events, each
/// with the functions its events name (re-indexed from 0), `place` saying where
/// each run function lives (repo, relative file).
pub fn chunks(run: &TraceRun, target: &str, per_chunk: usize, place: impl Fn(&crate::TracedFn) -> FiledFn) -> Vec<Chunk> {
let per_chunk = per_chunk.max(1);
let placed: Vec<FiledFn> = run.functions.iter().map(&place).collect();
let parts: Vec<&[crate::TraceEvent]> = if run.events.is_empty() { vec![&[]] } else { run.events.chunks(per_chunk).collect() };
let n = parts.len() as u32;
let threads = run.threads().len() as u32;
parts
.into_iter()
.enumerate()
.map(|(i, evs)| {
let mut local: std::collections::HashMap<u32, u32> = Default::default();
let mut functions = Vec::new();
let evs: Vec<crate::TraceEvent> = evs
.iter()
.map(|e| {
let f = *local.entry(e.func).or_insert_with(|| {
functions.push(placed[e.func as usize].clone());
(functions.len() - 1) as u32
});
crate::TraceEvent { func: f, ..*e }
})
.collect();
Chunk {
filing: ChunkFiling {
chunk: i as u32,
chunks: n,
base_seq: evs.first().map(|e| e.seq_no).unwrap_or(0),
target: target.to_string(),
functions,
binaries: run.binaries.clone(),
total_events: run.total_events,
truncated: run.truncated,
threads,
duration_ns: run.duration_ns(),
},
events: encode(&evs),
}
})
.collect()
}
/// A profile as filed (`trace_json` of a `profile` filing).
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
pub struct ProfileFiling {
pub target: String,
pub samples: u64,
pub total_ns: u64,
pub functions: Vec<FiledHeat>,
}
/// One function's heat, placed like a [`FiledFn`].
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
pub struct FiledHeat {
#[serde(flatten)]
pub at: FiledFn,
pub self_ns: u64,
pub total_ns: u64,
pub self_samples: u64,
pub total_samples: u64,
}
#[cfg(test)]
mod chunk_tests {
use super::*;
use crate::{EventKind, TraceEvent, TracedFn};
#[test]
fn a_run_cut_into_chunks_reindexes_each_chunks_functions() {
let run = TraceRun {
functions: ["a", "b", "c"].map(|s| TracedFn { symbol: s.into(), file: format!("/w/x/{s}.rs"), line: 1 }).to_vec(),
events: (0..5u64)
.map(|i| TraceEvent { thread: 1, seq_no: i, t_ns: i * 10, func: [0, 2, 2, 1, 0][i as usize], kind: EventKind::Enter })
.collect(),
total_events: 5,
..Default::default()
};
let cs = chunks(&run, "test t", 3, |f| FiledFn { symbol: f.symbol.clone(), repo: "x".into(), file: f.file.trim_start_matches("/w/x/").into(), line: f.line });
assert_eq!(cs.len(), 2);
assert_eq!((cs[0].filing.chunk, cs[0].filing.chunks, cs[0].filing.base_seq), (0, 2, 0));
assert_eq!(cs[0].filing.functions.iter().map(|f| f.symbol.as_str()).collect::<Vec<_>>(), ["a", "c"]);
assert_eq!(cs[1].filing.functions.iter().map(|f| f.symbol.as_str()).collect::<Vec<_>>(), ["b", "a"]);
assert_eq!(cs[1].filing.functions[0].file, "b.rs");
let back = decode(&cs[1].events, cs[1].filing.base_seq).unwrap();
assert_eq!(back.iter().map(|e| (e.seq_no, e.t_ns, e.func)).collect::<Vec<_>>(), vec![(3, 30, 0), (4, 40, 1)]);
}
}