use core::fmt;
use core::fmt::Write as _;
use core::marker::PhantomData;
use core::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
use std::cell::RefCell;
use std::sync::{Arc, Mutex, MutexGuard, OnceLock, PoisonError};
use crate::time::{MonotonicTime, TimeSource as _};
pub const RECORDS_PER_THREAD: usize = 4096;
pub const MAX_TRACKS: usize = 64;
const MAX_DEPTH: usize = RECORDS_PER_THREAD / 2;
static ENABLED: AtomicBool = AtomicBool::new(false);
static NEXT_TRACK: AtomicU32 = AtomicU32::new(0);
static REFUSED_TRACKS: AtomicU64 = AtomicU64::new(0);
static TRACKS: Mutex<Vec<Arc<TrackEntry>>> = Mutex::new(Vec::new());
static EPOCH: OnceLock<MonotonicTime> = OnceLock::new();
pub fn set_enabled(enabled: bool) -> bool {
ENABLED.swap(enabled, Ordering::Relaxed)
}
#[inline]
#[must_use]
pub fn is_enabled() -> bool {
ENABLED.load(Ordering::Relaxed)
}
pub const ENV_VAR: &str = "CRCBL_TRACE";
pub const DEFAULT_ENABLED: bool = false;
pub fn init_from_env() -> bool {
if cfg!(target_arch = "wasm32") {
return false;
}
let enabled = std::env::var(ENV_VAR)
.ok()
.and_then(|value| parse_enabled(&value))
.unwrap_or(DEFAULT_ENABLED);
set_enabled(enabled);
if enabled {
log::info!("trace: recording CPU spans ({ENV_VAR} is set)");
}
enabled
}
pub(crate) fn parse_enabled(value: &str) -> Option<bool> {
match value.trim().to_ascii_lowercase().as_str() {
"1" | "true" | "on" | "yes" => Some(true),
"0" | "false" | "off" | "no" => Some(false),
_ => None,
}
}
fn now_nanos() -> u64 {
let epoch = EPOCH.get_or_init(MonotonicTime::new);
u64::try_from(epoch.elapsed().as_nanos()).unwrap_or(u64::MAX)
}
fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
mutex.lock().unwrap_or_else(PoisonError::into_inner)
}
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Debug)]
pub struct ThreadTrack(u32);
impl ThreadTrack {
#[inline]
#[must_use]
pub const fn get(self) -> u32 {
self.0
}
#[inline]
#[must_use]
pub const fn new(number: u32) -> Self {
Self(number)
}
}
impl fmt::Display for ThreadTrack {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "thread {}", self.0)
}
}
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum RecordKind {
SpanBegin,
SpanEnd,
Counter(u64),
}
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub struct Record {
pub name: &'static str,
pub kind: RecordKind,
pub depth: u16,
pub nanos: u64,
}
#[derive(Clone, PartialEq, Eq, Debug)]
pub struct ThreadTrace {
pub track: ThreadTrack,
pub name: Option<String>,
pub records: Vec<Record>,
pub dropped: u64,
}
#[derive(Clone, Default, PartialEq, Eq, Debug)]
pub struct Snapshot {
pub threads: Vec<ThreadTrace>,
pub refused_tracks: u64,
}
impl Snapshot {
#[must_use]
pub fn record_count(&self) -> usize {
self.threads.iter().map(|thread| thread.records.len()).sum()
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.threads.is_empty() && self.refused_tracks == 0
}
#[must_use]
pub fn dropped(&self) -> u64 {
self.threads.iter().map(|thread| thread.dropped).sum()
}
#[must_use]
pub fn report(&self) -> String {
if self.is_empty() {
return "trace: nothing recorded\n".to_string();
}
let mut out = String::new();
let _ = writeln!(
out,
"trace: {} record(s) on {} thread(s), {} dropped",
self.record_count(),
self.threads.len(),
self.dropped(),
);
for thread in &self.threads {
let _ = writeln!(
out,
" {:<12} {:<16} {:>6} record(s), {:>6} dropped",
thread.track.to_string(),
thread.name.as_deref().unwrap_or("-"),
thread.records.len(),
thread.dropped,
);
}
if self.refused_tracks != 0 {
let _ = writeln!(
out,
" {} thread(s) refused a track: MAX_TRACKS were already taken",
self.refused_tracks,
);
}
out
}
}
#[derive(Debug)]
#[must_use = "a span runs for as long as its guard lives; `trace::span(name)` \
on its own opens and closes it in the same instant"]
pub struct Span {
open: bool,
_not_send: PhantomData<*const ()>,
}
impl Drop for Span {
#[inline]
fn drop(&mut self) {
if self.open {
record_end();
}
}
}
#[inline]
pub fn span(name: &'static str) -> Span {
if !is_enabled() {
return Span {
open: false,
_not_send: PhantomData,
};
}
Span {
open: record_begin(name),
_not_send: PhantomData,
}
}
#[inline]
pub fn counter(name: &'static str, value: u64) {
if !is_enabled() {
return;
}
let nanos = now_nanos();
with_buffer(|buffer| buffer.sample(name, value, nanos));
}
pub fn drain() -> Snapshot {
let mut tracks = lock(&TRACKS);
let mut threads = Vec::new();
tracks.retain(|entry| {
let mut buffer = lock(&entry.buffer);
if !buffer.records.is_empty() || buffer.dropped != 0 {
threads.push(ThreadTrace {
track: entry.track,
name: entry.name.clone(),
records: buffer.records.clone(),
dropped: buffer.dropped,
});
buffer.records.clear();
buffer.dropped = 0;
}
drop(buffer);
Arc::strong_count(entry) > 1
});
drop(tracks);
Snapshot {
threads,
refused_tracks: REFUSED_TRACKS.swap(0, Ordering::Relaxed),
}
}
#[derive(Debug)]
struct TrackBuffer {
records: Vec<Record>,
open: Vec<&'static str>,
dropped: u64,
}
impl TrackBuffer {
fn new() -> Self {
Self {
records: Vec::with_capacity(RECORDS_PER_THREAD),
open: Vec::with_capacity(MAX_DEPTH),
dropped: 0,
}
}
fn has_room_for(&self, extra: usize) -> bool {
self.records.len() + self.open.len() + extra <= RECORDS_PER_THREAD
}
fn push(&mut self, name: &'static str, kind: RecordKind, nanos: u64) {
debug_assert!(
self.records.len() < self.records.capacity(),
"the trace buffer is about to reallocate on the recording path"
);
let depth = u16::try_from(self.open.len()).unwrap_or(u16::MAX);
self.records.push(Record {
name,
kind,
depth,
nanos,
});
}
fn begin(&mut self, name: &'static str, nanos: u64) -> bool {
if !self.has_room_for(2) {
self.dropped += 1;
return false;
}
self.push(name, RecordKind::SpanBegin, nanos);
self.open.push(name);
true
}
fn end(&mut self, nanos: u64) {
let Some(name) = self.open.pop() else {
return;
};
self.push(name, RecordKind::SpanEnd, nanos);
}
fn sample(&mut self, name: &'static str, value: u64, nanos: u64) {
if !self.has_room_for(1) {
self.dropped += 1;
return;
}
self.push(name, RecordKind::Counter(value), nanos);
}
}
#[derive(Debug)]
struct TrackEntry {
track: ThreadTrack,
name: Option<String>,
buffer: Mutex<TrackBuffer>,
}
#[derive(Debug)]
enum Registration {
Untried,
Refused,
Track(Arc<TrackEntry>),
}
thread_local! {
static LOCAL: RefCell<Registration> = const { RefCell::new(Registration::Untried) };
}
fn register() -> Option<Arc<TrackEntry>> {
let mut tracks = lock(&TRACKS);
if tracks.len() >= MAX_TRACKS {
REFUSED_TRACKS.fetch_add(1, Ordering::Relaxed);
return None;
}
let entry = Arc::new(TrackEntry {
track: ThreadTrack(NEXT_TRACK.fetch_add(1, Ordering::Relaxed)),
name: std::thread::current().name().map(str::to_owned),
buffer: Mutex::new(TrackBuffer::new()),
});
tracks.push(Arc::clone(&entry));
drop(tracks);
Some(entry)
}
fn with_buffer<R>(f: impl FnOnce(&mut TrackBuffer) -> R) -> Option<R> {
LOCAL
.try_with(|slot| {
let mut slot = slot.try_borrow_mut().ok()?;
if matches!(*slot, Registration::Untried) {
*slot = match register() {
Some(entry) => Registration::Track(entry),
None => Registration::Refused,
};
}
let Registration::Track(entry) = &*slot else {
return None;
};
Some(f(&mut lock(&entry.buffer)))
})
.ok()
.flatten()
}
fn record_begin(name: &'static str) -> bool {
let nanos = now_nanos();
with_buffer(|buffer| buffer.begin(name, nanos)).unwrap_or(false)
}
fn record_end() {
let nanos = now_nanos();
with_buffer(|buffer| buffer.end(nanos));
}
#[cfg(test)]
mod tests {
use std::time::Duration;
use proptest::prelude::*;
use super::*;
static TRACE_TESTS: Mutex<()> = Mutex::new(());
#[derive(Debug)]
struct TraceTest {
_guard: MutexGuard<'static, ()>,
}
impl Drop for TraceTest {
fn drop(&mut self) {
set_enabled(false);
drop(drain());
}
}
fn trace_test() -> TraceTest {
let guard = TRACE_TESTS.lock().unwrap_or_else(PoisonError::into_inner);
set_enabled(false);
drop(drain());
set_enabled(true);
TraceTest { _guard: guard }
}
type Shape = (&'static str, RecordKind, u16);
fn shape(record: &Record) -> Shape {
(record.name, record.kind, record.depth)
}
fn drain_this_thread() -> Vec<Record> {
let snapshot = drain();
assert!(
snapshot.threads.len() <= 1,
"another thread is recording: {}",
snapshot.report()
);
assert_eq!(snapshot.refused_tracks, 0);
snapshot
.threads
.into_iter()
.next()
.map_or_else(Vec::new, |thread| {
assert_eq!(thread.dropped, 0, "records were refused unexpectedly");
thread.records
})
}
fn shapes_of(records: &[Record]) -> Vec<Shape> {
records.iter().map(shape).collect()
}
#[test]
fn nesting_is_recorded_in_order_and_at_the_right_depth() {
let _test = trace_test();
{
let _frame = span("frame");
{
let _tick = span("tick");
let _physics = span("physics");
}
let _render = span("render");
}
assert_eq!(
shapes_of(&drain_this_thread()),
vec![
("frame", RecordKind::SpanBegin, 0),
("tick", RecordKind::SpanBegin, 1),
("physics", RecordKind::SpanBegin, 2),
("physics", RecordKind::SpanEnd, 2),
("tick", RecordKind::SpanEnd, 1),
("render", RecordKind::SpanBegin, 1),
("render", RecordKind::SpanEnd, 1),
("frame", RecordKind::SpanEnd, 0),
]
);
}
#[test]
fn the_disabled_gate_records_nothing() {
let _test = trace_test();
set_enabled(false);
assert!(!is_enabled());
{
let _frame = span("frame");
let _tick = span("tick");
counter("draws", 4096);
}
let snapshot = drain();
assert!(
snapshot.is_empty(),
"the disabled trace recorded something: {}",
snapshot.report()
);
assert_eq!(snapshot.record_count(), 0);
assert_eq!(snapshot.dropped(), 0);
assert_eq!(snapshot.report(), "trace: nothing recorded\n");
}
#[test]
fn the_gate_reports_what_it_was_set_to() {
let _test = trace_test();
assert!(
set_enabled(true),
"`trace_test` hands the gate over enabled"
);
assert!(is_enabled());
assert!(set_enabled(false), "the previous value was `true`");
assert!(!is_enabled());
assert!(!set_enabled(true), "the previous value was `false`");
assert!(is_enabled());
}
#[test]
fn the_variables_spellings_are_read_the_way_crcbl_log_reads_a_directive() {
for on in ["1", "true", "on", "yes", "TRUE", " On ", "YES"] {
assert_eq!(parse_enabled(on), Some(true), "{on:?}");
}
for off in ["0", "false", "off", "no", "OFF", " False "] {
assert_eq!(parse_enabled(off), Some(false), "{off:?}");
}
for nonsense in ["", " ", "yse", "2", "enabled", "trace", "-1"] {
assert_eq!(parse_enabled(nonsense), None, "{nonsense:?}");
}
}
#[test]
#[cfg_attr(miri, ignore = "miri cannot spawn the child process this needs")]
fn the_environment_variable_is_what_turns_the_gate_on() {
let _test = trace_test();
set_enabled(false);
if std::env::var(ENV_VAR).is_ok() {
assert!(init_from_env(), "the child was launched with {ENV_VAR}=1");
assert!(is_enabled(), "and the gate is what it set");
return;
}
assert!(!init_from_env(), "{ENV_VAR} is not set in this environment");
assert!(!is_enabled(), "so the gate is left at DEFAULT_ENABLED");
let output = crcbl_spawn::command(
std::env::current_exe().expect("a test binary knows its own path"),
)
.args([
"--exact",
"trace::tests::the_environment_variable_is_what_turns_the_gate_on",
])
.env(ENV_VAR, "1")
.output()
.expect("re-running this test binary");
let report = String::from_utf8_lossy(&output.stdout);
assert!(output.status.success(), "the child failed:\n{report}");
assert!(
report.contains("1 passed"),
"the child ran no test — has this one been renamed?\n{report}"
);
}
#[test]
fn a_guard_survives_the_gate_flipping_under_it() {
let _test = trace_test();
{
let _outer = span("opened-enabled");
set_enabled(false);
}
assert_eq!(
shapes_of(&drain_this_thread()),
vec![
("opened-enabled", RecordKind::SpanBegin, 0),
("opened-enabled", RecordKind::SpanEnd, 0),
]
);
{
let _outer = span("opened-disabled");
set_enabled(true);
}
assert_eq!(shapes_of(&drain_this_thread()), Vec::new());
{
let _after = span("after");
}
assert_eq!(
shapes_of(&drain_this_thread()),
vec![
("after", RecordKind::SpanBegin, 0),
("after", RecordKind::SpanEnd, 0),
]
);
}
#[test]
fn counters_carry_their_value_and_the_depth_they_were_sampled_at() {
let _test = trace_test();
counter("jobs", 7);
{
let _frame = span("frame");
counter("draws", 128);
counter("instances", u64::MAX);
}
assert_eq!(
shapes_of(&drain_this_thread()),
vec![
("jobs", RecordKind::Counter(7), 0),
("frame", RecordKind::SpanBegin, 0),
("draws", RecordKind::Counter(128), 1),
("instances", RecordKind::Counter(u64::MAX), 1),
("frame", RecordKind::SpanEnd, 0),
]
);
}
#[test]
fn a_full_buffer_refuses_and_counts_instead_of_growing() {
let _test = trace_test();
let attempted = u64::try_from(RECORDS_PER_THREAD).unwrap() + 16;
{
let _frame = span("frame");
for value in 0..attempted {
counter("filler", value);
}
let _refused = span("refused");
}
let capacity_before = buffer_capacity();
let snapshot = drain();
let thread = &snapshot.threads[0];
assert_eq!(
thread.records.len(),
RECORDS_PER_THREAD,
"the buffer should be exactly full"
);
let admitted = thread
.records
.iter()
.filter(|record| matches!(record.kind, RecordKind::Counter(_)))
.count();
assert!(admitted > 0 && u64::try_from(admitted).unwrap() < attempted);
assert_eq!(
thread.dropped,
attempted - u64::try_from(admitted).unwrap() + 1
);
assert_eq!(snapshot.dropped(), thread.dropped);
assert_eq!(
shape(&thread.records[0]),
("frame", RecordKind::SpanBegin, 0)
);
let last = thread.records.last().expect("the buffer is not empty");
assert_eq!(shape(last), ("frame", RecordKind::SpanEnd, 0));
assert!(thread.records.iter().all(|record| record.name != "refused"));
assert!(
snapshot.report().contains("dropped"),
"the report hides the hole: {}",
snapshot.report()
);
assert_eq!(capacity_before, RECORDS_PER_THREAD);
assert_eq!(
buffer_capacity(),
RECORDS_PER_THREAD,
"draining reallocated"
);
}
fn buffer_capacity() -> usize {
LOCAL.with(|slot| {
let slot = slot.borrow();
let Registration::Track(entry) = &*slot else {
panic!("this thread has no track");
};
lock(&entry.buffer).records.capacity()
})
}
#[test]
fn a_worker_thread_gets_a_track_of_its_own() {
let _test = trace_test();
{
let _frame = span("frame");
std::thread::Builder::new()
.name("worker-0".to_owned())
.spawn(|| {
let _job = span("job");
counter("steals", 3);
})
.expect("spawn a worker")
.join()
.expect("the worker did not panic");
}
let snapshot = drain();
assert_eq!(snapshot.threads.len(), 2, "{}", snapshot.report());
let worker = snapshot
.threads
.iter()
.find(|thread| thread.name.as_deref() == Some("worker-0"))
.unwrap_or_else(|| panic!("no worker track: {}", snapshot.report()));
assert_eq!(
shapes_of(&worker.records),
vec![
("job", RecordKind::SpanBegin, 0),
("steals", RecordKind::Counter(3), 1),
("job", RecordKind::SpanEnd, 0),
]
);
let main = snapshot
.threads
.iter()
.find(|thread| thread.track != worker.track)
.expect("the main thread's track");
assert_ne!(main.track, worker.track);
assert_eq!(
shapes_of(&main.records),
vec![
("frame", RecordKind::SpanBegin, 0),
("frame", RecordKind::SpanEnd, 0),
]
);
assert_eq!(drain_this_thread(), Vec::new());
}
#[test]
fn threads_past_the_track_bound_are_refused_and_counted() {
let _test = trace_test();
let surplus = 4;
for index in 0..MAX_TRACKS + surplus {
std::thread::Builder::new()
.name(format!("crowd-{index}"))
.spawn(|| {
let _job = span("job");
})
.expect("spawn")
.join()
.expect("the thread did not panic");
}
let snapshot = drain();
assert!(
snapshot.threads.len() <= MAX_TRACKS,
"{} tracks past a bound of {MAX_TRACKS}",
snapshot.threads.len()
);
assert!(
snapshot.refused_tracks >= u64::try_from(surplus).unwrap(),
"{} refusals: {}",
snapshot.refused_tracks,
snapshot.report()
);
assert!(snapshot.report().contains("refused a track"));
drop(span("after"));
assert_eq!(drain().threads.len(), 1);
assert_eq!(drain().refused_tracks, 0);
}
#[test]
fn a_spans_timestamps_bracket_what_ran_inside_it() {
let _test = trace_test();
let nap = Duration::from_millis(5);
{
let _outer = span("outer");
let _inner = span("inner");
std::thread::sleep(nap);
}
let records = drain_this_thread();
let [outer_begin, inner_begin, inner_end, outer_end] = records.as_slice() else {
panic!("expected two nested spans, got {records:?}");
};
let floor = u64::try_from(nap.as_nanos()).unwrap();
assert!(
inner_end.nanos - inner_begin.nanos >= floor,
"the inner span slept {nap:?} and reports {} ns",
inner_end.nanos - inner_begin.nanos
);
assert!(
outer_end.nanos - outer_begin.nanos >= inner_end.nanos - inner_begin.nanos,
"the outer span is shorter than the inner one it contains"
);
assert!(outer_begin.nanos <= inner_begin.nanos);
assert!(inner_end.nanos <= outer_end.nanos);
}
#[test]
fn a_drain_splits_a_span_that_is_open_across_it() {
let _test = trace_test();
let frame = span("frame");
assert_eq!(
shapes_of(&drain_this_thread()),
vec![("frame", RecordKind::SpanBegin, 0)]
);
drop(frame);
assert_eq!(
shapes_of(&drain_this_thread()),
vec![("frame", RecordKind::SpanEnd, 0)]
);
}
#[test]
fn tracks_are_numbered_and_displayed() {
assert_eq!(ThreadTrack(7).get(), 7);
assert_eq!(ThreadTrack(7).to_string(), "thread 7");
assert!(ThreadTrack(1) < ThreadTrack(2));
assert!(Snapshot::default().is_empty());
assert_eq!(Snapshot::default().record_count(), 0);
}
#[derive(Clone, Copy, Debug)]
enum Step {
Open(&'static str),
Close,
Sample(u64),
}
const NAMES: [&str; 4] = ["a", "b", "c", "d"];
fn any_step() -> impl Strategy<Value = Step> {
prop_oneof![
5 => (0..NAMES.len()).prop_map(|index| Step::Open(NAMES[index])),
4 => Just(Step::Close),
2 => any::<u64>().prop_map(Step::Sample),
]
}
proptest! {
#[test]
fn prop_recorded_depth_matches_a_stack_walk(steps in proptest::collection::vec(any_step(), 1..200)) {
let _test = trace_test();
let mut open: Vec<Span> = Vec::new();
for step in &steps {
match *step {
Step::Open(name) => open.push(span(name)),
Step::Close => drop(open.pop()),
Step::Sample(value) => counter("sample", value),
}
}
while open.pop().is_some() {}
let records = drain_this_thread();
prop_assert!(records.len() <= RECORDS_PER_THREAD);
let mut stack: Vec<&'static str> = Vec::new();
for record in &records {
match record.kind {
RecordKind::SpanBegin => {
prop_assert_eq!(usize::from(record.depth), stack.len());
stack.push(record.name);
}
RecordKind::SpanEnd => {
let opened = stack.pop();
prop_assert_eq!(opened, Some(record.name));
prop_assert_eq!(usize::from(record.depth), stack.len());
}
RecordKind::Counter(_) => {
prop_assert_eq!(usize::from(record.depth), stack.len());
}
}
}
prop_assert!(stack.is_empty(), "unclosed spans: {:?}", stack);
let recording = steps.iter().any(|step| !matches!(step, Step::Close));
prop_assert_eq!(records.is_empty(), !recording);
}
}
}