diff --git a/Cargo.lock b/Cargo.lock index 59936218079..b5bb4c2eb5b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -23879,10 +23879,14 @@ dependencies = [ name = "ztracing" version = "0.1.0" dependencies = [ + "async-channel 2.5.0", + "js-sys", "tracing", "tracing-subscriber", "tracing-tracy", "tracy-client", + "wasm-bindgen", + "web-sys", "zlog", "ztracing_macro", ] diff --git a/crates/ztracing/Cargo.toml b/crates/ztracing/Cargo.toml index ba4fea9a34e..81d3fc0d09f 100644 --- a/crates/ztracing/Cargo.toml +++ b/crates/ztracing/Cargo.toml @@ -10,13 +10,22 @@ workspace = true [features] tracy = ["tracing-tracy", "tracy-client"] +web = ["dep:async-channel", "dep:js-sys", "dep:wasm-bindgen", "dep:web-sys"] [dependencies] zlog.workspace = true tracing.workspace = true tracing-subscriber = "0.3.22" + +ztracing_macro.workspace = true + +[target.'cfg(not(target_family = "wasm"))'.dependencies] tracing-tracy = { version = "0.11.4", optional = true, features = ["enable", "ondemand"] } tracy-client = { version = "0.18.2", optional = true, features = ["enable", "ondemand"] } -ztracing_macro.workspace = true +[target.'cfg(target_family = "wasm")'.dependencies] +async-channel = { workspace = true, optional = true } +js-sys = { workspace = true, optional = true } +wasm-bindgen = { workspace = true, optional = true } +web-sys = { version = "0.3", features = ["console", "Performance"], optional = true } diff --git a/crates/ztracing/src/lib.rs b/crates/ztracing/src/lib.rs index 853633ac48c..ea3c493b771 100644 --- a/crates/ztracing/src/lib.rs +++ b/crates/ztracing/src/lib.rs @@ -1,37 +1,37 @@ pub use tracing::{Level, field}; -#[cfg(ztracing)] +#[cfg(any(ztracing, all(target_family = "wasm", feature = "web")))] pub use tracing::{ Span, debug_span, error_span, event, info_span, instrument, span, trace_span, warn_span, }; -#[cfg(not(ztracing))] +#[cfg(not(any(ztracing, all(target_family = "wasm", feature = "web"))))] pub use ztracing_macro::instrument; -#[cfg(ztracing)] +#[cfg(all(ztracing, not(target_family = "wasm")))] const MAX_CALLSTACK_DEPTH: u16 = 16; -#[cfg(all(ztracing, ztracing_with_memory))] +#[cfg(all(ztracing, ztracing_with_memory, not(target_family = "wasm")))] #[global_allocator] static GLOBAL: tracy_client::ProfiledAllocator = tracy_client::ProfiledAllocator::new(std::alloc::System, MAX_CALLSTACK_DEPTH); -#[cfg(not(ztracing))] +#[cfg(not(any(ztracing, all(target_family = "wasm", feature = "web"))))] pub use __consume_all_tokens as trace_span; -#[cfg(not(ztracing))] +#[cfg(not(any(ztracing, all(target_family = "wasm", feature = "web"))))] pub use __consume_all_tokens as info_span; -#[cfg(not(ztracing))] +#[cfg(not(any(ztracing, all(target_family = "wasm", feature = "web"))))] pub use __consume_all_tokens as debug_span; -#[cfg(not(ztracing))] +#[cfg(not(any(ztracing, all(target_family = "wasm", feature = "web"))))] pub use __consume_all_tokens as warn_span; -#[cfg(not(ztracing))] +#[cfg(not(any(ztracing, all(target_family = "wasm", feature = "web"))))] pub use __consume_all_tokens as error_span; -#[cfg(not(ztracing))] +#[cfg(not(any(ztracing, all(target_family = "wasm", feature = "web"))))] pub use __consume_all_tokens as event; -#[cfg(not(ztracing))] +#[cfg(not(any(ztracing, all(target_family = "wasm", feature = "web"))))] pub use __consume_all_tokens as span; -#[cfg(not(ztracing))] +#[cfg(not(any(ztracing, all(target_family = "wasm", feature = "web"))))] #[macro_export] macro_rules! __consume_all_tokens { ($($t:tt)*) => { @@ -39,10 +39,10 @@ macro_rules! __consume_all_tokens { }; } -#[cfg(not(ztracing))] +#[cfg(not(any(ztracing, all(target_family = "wasm", feature = "web"))))] pub struct Span; -#[cfg(not(ztracing))] +#[cfg(not(any(ztracing, all(target_family = "wasm", feature = "web"))))] impl Span { pub fn current() -> Self { Self @@ -53,7 +53,7 @@ impl Span { pub fn record(&self, _t: T, _s: S) {} } -#[cfg(ztracing)] +#[cfg(all(ztracing, not(target_family = "wasm")))] pub fn init() { use tracing_subscriber::fmt::format::DefaultFields; use tracing_subscriber::prelude::*; @@ -91,5 +91,27 @@ pub fn init() { .expect("setup tracy layer"); } -#[cfg(not(ztracing))] +#[cfg(all(target_family = "wasm", feature = "web"))] +mod web; + +#[cfg(all(target_family = "wasm", feature = "web"))] +pub use web::{PerformanceLayer, PerformanceReporter, performance_layer}; + +#[cfg(all(target_family = "wasm", feature = "web"))] +pub fn init(start_reporter: impl FnOnce(PerformanceReporter)) { + use tracing_subscriber::prelude::*; + + let (performance_layer, reporter) = performance_layer(); + if tracing::subscriber::set_global_default( + tracing_subscriber::registry().with(performance_layer), + ) + .is_ok() + { + start_reporter(reporter); + } else { + zlog::error!("failed to set browser performance tracing subscriber"); + } +} + +#[cfg(all(not(ztracing), not(all(target_family = "wasm", feature = "web"))))] pub fn init() {} diff --git a/crates/ztracing/src/web.rs b/crates/ztracing/src/web.rs new file mode 100644 index 00000000000..96f33b7b3ab --- /dev/null +++ b/crates/ztracing/src/web.rs @@ -0,0 +1,441 @@ +use std::{ + cell::{Cell, OnceCell}, + collections::HashMap, + sync::{ + Arc, + atomic::{AtomicU32, AtomicUsize, Ordering}, + }, +}; + +use tracing::{Id, Subscriber, span::Attributes}; +use tracing_subscriber::{Layer, registry::LookupSpan}; +use wasm_bindgen::{JsCast, JsValue, prelude::wasm_bindgen}; +use web_sys::Performance; + +const EVENT_BUFFER_CAPACITY: usize = 65_536; +const REPORT_BATCH_SIZE: usize = 4_096; + +static NEXT_SOURCE_THREAD_ID: AtomicU32 = AtomicU32::new(0); + +thread_local! { + static PERFORMANCE: OnceCell> = const { OnceCell::new() }; + static SOURCE_THREAD_ID: Cell> = const { Cell::new(None) }; +} + +#[wasm_bindgen] +extern "C" { + type PerformanceMeasureApi; + + #[wasm_bindgen(catch, method, js_class = "Performance", js_name = measure)] + fn measure_with_options( + this: &PerformanceMeasureApi, + measure_name: &str, + options: &JsValue, + ) -> Result; +} + +pub struct PerformanceLayer { + event_sender: async_channel::Sender, + dropped_event_count: Arc, +} + +pub struct PerformanceReporter { + event_receiver: async_channel::Receiver, + dropped_event_count: Arc, +} + +enum Event { + NewSpan { + span_id: u64, + target: &'static str, + name: &'static str, + }, + Enter { + span_id: u64, + source_thread_id: u32, + timestamp: f64, + }, + Exit { + span_id: u64, + source_thread_id: u32, + timestamp: f64, + }, + Close { + span_id: u64, + }, +} + +struct SpanState { + display_name: String, + active_entries: HashMap>, +} + +struct CompletedMeasure { + name: String, + source_thread_id: u32, + start: f64, + duration: f64, +} + +#[derive(Default)] +struct ReporterState { + spans: HashMap, +} + +pub fn performance_layer() -> (PerformanceLayer, PerformanceReporter) { + let (event_sender, event_receiver) = async_channel::bounded(EVENT_BUFFER_CAPACITY); + let dropped_event_count = Arc::new(AtomicUsize::new(0)); + ( + PerformanceLayer { + event_sender, + dropped_event_count: dropped_event_count.clone(), + }, + PerformanceReporter { + event_receiver, + dropped_event_count, + }, + ) +} + +impl PerformanceLayer { + pub fn dropped_event_count(&self) -> usize { + self.dropped_event_count.load(Ordering::Relaxed) + } + + fn send(&self, event: Event) { + if self.event_sender.try_send(event).is_err() { + self.dropped_event_count.fetch_add(1, Ordering::Relaxed); + } + } +} + +impl Layer for PerformanceLayer +where + S: Subscriber + for<'lookup> LookupSpan<'lookup>, +{ + fn on_new_span( + &self, + _attributes: &Attributes<'_>, + id: &Id, + context: tracing_subscriber::layer::Context<'_, S>, + ) { + let Some(span) = context.span(id) else { + return; + }; + let metadata = span.metadata(); + self.send(Event::NewSpan { + span_id: id.into_u64(), + target: metadata.target(), + name: metadata.name(), + }); + } + + fn on_enter(&self, id: &Id, _context: tracing_subscriber::layer::Context<'_, S>) { + let Some(timestamp) = absolute_now() else { + return; + }; + self.send(Event::Enter { + span_id: id.into_u64(), + source_thread_id: source_thread_id(), + timestamp, + }); + } + + fn on_exit(&self, id: &Id, _context: tracing_subscriber::layer::Context<'_, S>) { + let Some(timestamp) = absolute_now() else { + return; + }; + self.send(Event::Exit { + span_id: id.into_u64(), + source_thread_id: source_thread_id(), + timestamp, + }); + } + + fn on_close(&self, id: Id, _context: tracing_subscriber::layer::Context<'_, S>) { + self.send(Event::Close { + span_id: id.into_u64(), + }); + } +} + +impl PerformanceReporter { + pub async fn run(self) { + let Some(performance) = performance() else { + web_sys::console::error_1(&JsValue::from_str( + "browser performance tracing requires globalThis.performance", + )); + return; + }; + let reporter_time_origin = performance.time_origin(); + let mut state = ReporterState::default(); + let mut completed_measures = Vec::new(); + let mut reported_dropped_event_count = 0; + + while let Ok(event) = self.event_receiver.recv().await { + state.process(event, &mut completed_measures); + let mut processed_event_count = 1; + while processed_event_count < REPORT_BATCH_SIZE { + let Ok(event) = self.event_receiver.try_recv() else { + break; + }; + state.process(event, &mut completed_measures); + processed_event_count += 1; + } + + for measure in completed_measures.drain(..) { + report_measure(&performance, reporter_time_origin, measure); + } + + let dropped_event_count = self.dropped_event_count.load(Ordering::Relaxed); + if dropped_event_count != reported_dropped_event_count { + web_sys::console::warn_1(&JsValue::from_str(&format!( + "browser performance tracing dropped {} events", + dropped_event_count - reported_dropped_event_count + ))); + reported_dropped_event_count = dropped_event_count; + } + } + } +} + +impl ReporterState { + fn process(&mut self, event: Event, completed_measures: &mut Vec) { + match event { + Event::NewSpan { + span_id, + target, + name, + } => { + self.spans.insert( + span_id, + SpanState { + display_name: format!("{target}::{name}"), + active_entries: HashMap::new(), + }, + ); + } + Event::Enter { + span_id, + source_thread_id, + timestamp, + } => { + let Some(span) = self.spans.get_mut(&span_id) else { + return; + }; + span.active_entries + .entry(source_thread_id) + .or_default() + .push(timestamp); + } + Event::Exit { + span_id, + source_thread_id, + timestamp, + } => { + let Some(span) = self.spans.get_mut(&span_id) else { + return; + }; + let Some(active_entries) = span.active_entries.get_mut(&source_thread_id) else { + return; + }; + let Some(start) = active_entries.pop() else { + return; + }; + if active_entries.is_empty() { + span.active_entries.remove(&source_thread_id); + } + completed_measures.push(CompletedMeasure { + name: span.display_name.clone(), + source_thread_id, + start, + duration: (timestamp - start).max(0.0), + }); + } + Event::Close { span_id } => { + self.spans.remove(&span_id); + } + } + } +} + +fn absolute_now() -> Option { + with_performance(|performance| performance.time_origin() + performance.now()) +} + +fn source_thread_id() -> u32 { + SOURCE_THREAD_ID.with(|source_thread_id| { + if let Some(source_thread_id) = source_thread_id.get() { + source_thread_id + } else { + let new_source_thread_id = NEXT_SOURCE_THREAD_ID.fetch_add(1, Ordering::Relaxed); + source_thread_id.set(Some(new_source_thread_id)); + new_source_thread_id + } + }) +} + +fn performance() -> Option { + with_performance(Clone::clone) +} + +fn with_performance(callback: impl FnOnce(&Performance) -> T) -> Option { + PERFORMANCE.with(|performance| { + performance + .get_or_init(|| { + js_sys::Reflect::get(&js_sys::global(), &JsValue::from_str("performance")) + .ok() + .and_then(|performance| performance.dyn_into().ok()) + }) + .as_ref() + .map(callback) + }) +} + +fn report_measure(performance: &Performance, reporter_time_origin: f64, measure: CompletedMeasure) { + let start = measure.start - reporter_time_origin; + if start < 0.0 { + web_sys::console::warn_1(&JsValue::from_str(&format!( + "browser performance tracing dropped {} because it predates the reporter worker", + measure.name + ))); + return; + } + + let options = js_sys::Object::new(); + if let Err(error) = js_sys::Reflect::set( + &options, + &JsValue::from_str("start"), + &JsValue::from_f64(start), + ) { + log_performance_error("set measure start", error); + return; + } + if let Err(error) = js_sys::Reflect::set( + &options, + &JsValue::from_str("duration"), + &JsValue::from_f64(measure.duration), + ) { + log_performance_error("set measure duration", error); + return; + } + let detail = js_sys::Object::new(); + if let Err(error) = js_sys::Reflect::set( + &detail, + &JsValue::from_str("sourceThreadId"), + &JsValue::from_f64(f64::from(measure.source_thread_id)), + ) { + log_performance_error("set measure source thread", error); + return; + } + if let Err(error) = + js_sys::Reflect::set(&options, &JsValue::from_str("detail"), detail.as_ref()) + { + log_performance_error("set measure detail", error); + return; + } + let performance: &PerformanceMeasureApi = performance.unchecked_ref(); + if let Err(error) = performance.measure_with_options(&measure.name, options.as_ref()) { + log_performance_error("measure", error); + } +} + +fn log_performance_error(operation: &str, error: JsValue) { + web_sys::console::error_2( + &JsValue::from_str(&format!("performance.{operation} failed")), + &error, + ); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pairs_span_events() { + let mut state = ReporterState::default(); + let mut completed_measures = Vec::new(); + + state.process( + Event::NewSpan { + span_id: 1, + target: "editor", + name: "layout", + }, + &mut completed_measures, + ); + state.process( + Event::Enter { + span_id: 1, + source_thread_id: 2, + timestamp: 10.0, + }, + &mut completed_measures, + ); + state.process( + Event::Exit { + span_id: 1, + source_thread_id: 2, + timestamp: 15.0, + }, + &mut completed_measures, + ); + + assert_eq!(completed_measures.len(), 1); + let Some(measure) = completed_measures.first() else { + return; + }; + assert_eq!(measure.name, "editor::layout"); + assert_eq!(measure.source_thread_id, 2); + assert_eq!(measure.start, 10.0); + assert_eq!(measure.duration, 5.0); + } + + #[test] + fn pairs_reentrant_spans_in_stack_order() { + let mut state = ReporterState::default(); + let mut completed_measures = Vec::new(); + + state.process( + Event::NewSpan { + span_id: 1, + target: "editor", + name: "layout", + }, + &mut completed_measures, + ); + for timestamp in [10.0, 12.0] { + state.process( + Event::Enter { + span_id: 1, + source_thread_id: 2, + timestamp, + }, + &mut completed_measures, + ); + } + for timestamp in [14.0, 16.0] { + state.process( + Event::Exit { + span_id: 1, + source_thread_id: 2, + timestamp, + }, + &mut completed_measures, + ); + } + + assert_eq!(completed_measures.len(), 2); + let mut measures = completed_measures.iter(); + let Some(inner_measure) = measures.next() else { + return; + }; + let Some(outer_measure) = measures.next() else { + return; + }; + assert_eq!(inner_measure.start, 12.0); + assert_eq!(inner_measure.duration, 2.0); + assert_eq!(outer_measure.start, 10.0); + assert_eq!(outer_measure.duration, 6.0); + } +}