latexml_core/
telemetry.rs

1//! Per-job telemetry: phase wall times, counts, and resource peaks.
2//!
3//! See `docs/performance/TELEMETRY.md` for the design contract.
4//!
5//! Default-on instrumentation. Coarse phase wrappers cost ~20ns each
6//! (one `Instant::now()` call); the math-parse histogram update is
7//! the only per-formula instrumentation and is a single atomic
8//! increment of one of 9 `u32` slots.
9//!
10//! Thread-local state. Aggregate at end-of-process via `take()`.
11//! All times in microseconds; counts in their natural unit.
12
13use std::{cell::RefCell, time::Instant};
14
15/// Coarse phase enum. 17 values; bumping requires updating
16/// `Telemetry::write_json` and `tools/perf_phase_summary.py`.
17///
18/// Phase ordering reflects the conversion pipeline order
19/// (Bootstrap → Digest → Build → Rewrite → MathParse →
20/// PostXmlParse → PostScan → Bibliography → Crossref → Graphics →
21/// MathImages → MathmlPres → MathmlCont → Split → Xslt →
22/// Html5Fixups → Serialize) so flat dumps read in execution order.
23#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
24#[repr(u8)]
25pub enum Phase {
26  Bootstrap = 0,
27  Digest = 1,
28  Build = 2,
29  Rewrite = 3,
30  MathParse = 4,
31  /// Parses the XML emitted by core into a `PostDocument`.
32  /// Size-proportional; material on large papers.
33  PostXmlParse = 5,
34  /// Scan phase of latexml_post: ID assignment, label resolution, etc.
35  PostScan = 6,
36  Bibliography = 7,
37  Crossref = 8,
38  /// External-tool dispatch for `\includegraphics` (mutool / pdftocairo / convert / gs).
39  Graphics = 9,
40  /// External-tool dispatch for picture/latex/math image rendering
41  /// (`picture_images.rs` + `latex_images.rs` + `math_images.rs`).
42  MathImages = 10,
43  MathmlPres = 11,
44  MathmlCont = 12,
45  /// Document splitting (only when `--split` is on; otherwise 0).
46  Split = 13,
47  Xslt = 14,
48  /// Final HTML tweaks after XSLT (asset paths, header/footer).
49  Html5Fixups = 15,
50  Serialize = 16,
51}
52
53impl Phase {
54  pub const COUNT: usize = 17;
55
56  pub fn as_str(self) -> &'static str {
57    match self {
58      Phase::Bootstrap => "bootstrap",
59      Phase::Digest => "digest",
60      Phase::Build => "build",
61      Phase::Rewrite => "rewrite",
62      Phase::MathParse => "math_parse",
63      Phase::PostXmlParse => "post_xml_parse",
64      Phase::PostScan => "post_scan",
65      Phase::Bibliography => "bibliography",
66      Phase::Crossref => "crossref",
67      Phase::Graphics => "graphics",
68      Phase::MathImages => "math_images",
69      Phase::MathmlPres => "mathml_pres",
70      Phase::MathmlCont => "mathml_cont",
71      Phase::Split => "split",
72      Phase::Xslt => "xslt",
73      Phase::Html5Fixups => "html5_fixups",
74      Phase::Serialize => "serialize",
75    }
76  }
77}
78
79/// Math-parse time bucket boundaries in microseconds.
80/// `bucket(us)` returns 0..=8.
81const BUCKET_BOUNDS_US: [u64; 8] = [500, 1_000, 2_000, 5_000, 10_000, 20_000, 50_000, 100_000];
82
83fn bucket_for(us: u64) -> usize {
84  for (i, b) in BUCKET_BOUNDS_US.iter().enumerate() {
85    if us < *b {
86      return i;
87    }
88  }
89  8
90}
91
92/// Per-job telemetry record.
93///
94/// Identifier fields (`paper_id`, `cmdline`, `host`, `git_sha`) are
95/// filled in by the binary entry point, not by the engine. Phase
96/// wall times and counts are populated by the engine via this
97/// module's API.
98#[derive(Clone, Debug)]
99pub struct Telemetry {
100  // Identifiers (set by the binary)
101  pub paper_id:       String,
102  pub git_sha:        String,
103  pub cmdline:        String,
104  pub host:           String,
105  pub timeout_s:      u32,
106  pub schema_version: u32,
107
108  // Wall (microseconds)
109  pub wall_us:  u64,
110  pub phase_us: [u64; Phase::COUNT],
111
112  // Counts
113  pub formulae:                  u32,
114  pub math_parse_attempts:       u32,
115  pub math_parse_count:          u64,
116  pub math_parse_buckets:        [u32; 9],
117  pub graphics_assets:           u32,
118  pub graphics_subprocess_count: u32,
119  pub db_objects:                u32,
120  pub output_bytes:              u64,
121  pub warnings:                  u32,
122  pub errors:                    u32,
123  pub fatal_errors:              u32,
124  pub external_tool_count:       u32,
125
126  // Resource
127  pub max_rss_kb:    u64,
128  pub child_user_us: u64,
129  pub child_sys_us:  u64,
130
131  // Outcome (set by the binary at end)
132  pub category:  String,
133  pub exit_code: i32,
134}
135
136impl Default for Telemetry {
137  fn default() -> Self {
138    Telemetry {
139      paper_id:                  String::new(),
140      git_sha:                   option_env!("LATEXML_GIT_SHA").unwrap_or("").to_string(),
141      cmdline:                   String::new(),
142      host:                      String::new(),
143      timeout_s:                 0,
144      schema_version:            1,
145      wall_us:                   0,
146      phase_us:                  [0; Phase::COUNT],
147      formulae:                  0,
148      math_parse_attempts:       0,
149      math_parse_count:          0,
150      math_parse_buckets:        [0; 9],
151      graphics_assets:           0,
152      graphics_subprocess_count: 0,
153      db_objects:                0,
154      output_bytes:              0,
155      warnings:                  0,
156      errors:                    0,
157      fatal_errors:              0,
158      external_tool_count:       0,
159      max_rss_kb:                0,
160      child_user_us:             0,
161      child_sys_us:              0,
162      category:                  String::new(),
163      exit_code:                 0,
164    }
165  }
166}
167
168thread_local! {
169  static STATE: RefCell<Telemetry> = RefCell::new(Telemetry::default());
170  // Phase stack: each entry is (phase, started_at). Time accrues only
171  // to the innermost (top-of-stack) phase.
172  static STACK: RefCell<Vec<(Phase, Instant)>> = const { RefCell::new(Vec::new()) };
173}
174
175/// Begin a phase. Subsequent time accrues to this phase until the
176/// matching `phase_exit()` (or the `PhaseGuard` returned by
177/// [`phase`]) drops.
178pub fn phase_enter(p: Phase) {
179  let now = Instant::now();
180  STACK.with(|s| {
181    let mut stack = s.borrow_mut();
182    // If there's a parent phase, charge accumulated wall to it
183    // before we steal the clock.
184    if let Some((parent, started)) = stack.last_mut() {
185      let dt = now.saturating_duration_since(*started).as_micros() as u64;
186      let parent = *parent;
187      STATE.with(|st| st.borrow_mut().phase_us[parent as usize] += dt);
188      *started = now;
189    }
190    stack.push((p, now));
191  });
192}
193
194/// End the innermost phase.
195pub fn phase_exit() {
196  let now = Instant::now();
197  STACK.with(|s| {
198    let mut stack = s.borrow_mut();
199    let (p, started) = stack
200      .pop()
201      .expect("telemetry::phase_exit called without matching phase_enter");
202    let dt = now.saturating_duration_since(started).as_micros() as u64;
203    STATE.with(|st| st.borrow_mut().phase_us[p as usize] += dt);
204    // Reset parent's start so it doesn't double-count our time.
205    if let Some((_, started_parent)) = stack.last_mut() {
206      *started_parent = now;
207    }
208  });
209}
210
211/// RAII guard returned by [`phase`]. Calls `phase_exit` on drop.
212pub struct PhaseGuard {
213  _private: (),
214}
215
216impl Drop for PhaseGuard {
217  fn drop(&mut self) { phase_exit(); }
218}
219
220/// Convenience: `let _g = telemetry::phase(Phase::Digest);`
221pub fn phase(p: Phase) -> PhaseGuard {
222  phase_enter(p);
223  PhaseGuard { _private: () }
224}
225
226// ─── counters ───────────────────────────────────────────────────────────────
227
228pub fn incr_formulae() { STATE.with(|s| s.borrow_mut().formulae += 1); }
229
230/// Set the formulae count directly. Use when the document-wide count
231/// is known up front (e.g., right before `MathParser::parse_math` is
232/// invoked over all `<XMath>` nodes).
233pub fn set_formulae(n: u32) { STATE.with(|s| s.borrow_mut().formulae = n); }
234
235/// Add to the formulae count. For streaming pass 2, which parses math one
236/// SEGMENT at a time and so never knows the document-wide count up front —
237/// `set_formulae` there would record only the last segment's tally.
238pub fn add_formulae(n: u32) { STATE.with(|s| s.borrow_mut().formulae += n); }
239
240/// Record one math parse: total time and number of successful parses
241/// returned (the Marpa parser may produce multiple ASF derivations
242/// for one input). Updates the histogram bucket for the elapsed time.
243pub fn record_math_parse(us: u64, parses: u32) {
244  STATE.with(|s| {
245    let mut t = s.borrow_mut();
246    t.math_parse_attempts += 1;
247    t.math_parse_count += parses as u64;
248    t.math_parse_buckets[bucket_for(us)] += 1;
249  });
250}
251
252pub fn incr_graphics_asset() { STATE.with(|s| s.borrow_mut().graphics_assets += 1); }
253pub fn set_graphics_assets(n: u32) { STATE.with(|s| s.borrow_mut().graphics_assets = n); }
254pub fn incr_graphics_subprocess() { STATE.with(|s| s.borrow_mut().graphics_subprocess_count += 1); }
255/// Bulk-add subprocess counts from a worker-pool tally. Used after
256/// `std::thread::scope` joins because per-worker `thread_local!` STATE
257/// is discarded on thread exit; counts accumulated in a shared
258/// `AtomicU32` are merged here.
259pub fn add_graphics_subprocess(n: u32) {
260  STATE.with(|s| s.borrow_mut().graphics_subprocess_count += n);
261}
262pub fn incr_external_tool() { STATE.with(|s| s.borrow_mut().external_tool_count += 1); }
263pub fn set_db_objects(n: u32) { STATE.with(|s| s.borrow_mut().db_objects = n); }
264pub fn set_output_bytes(n: u64) { STATE.with(|s| s.borrow_mut().output_bytes = n); }
265pub fn incr_warning() { STATE.with(|s| s.borrow_mut().warnings += 1); }
266pub fn incr_error() { STATE.with(|s| s.borrow_mut().errors += 1); }
267pub fn incr_fatal_error() { STATE.with(|s| s.borrow_mut().fatal_errors += 1); }
268/// Bulk-set status counts at finalize time from `common::error::REPORT`
269/// (the canonical Error!/Warn!/Fatal! counter). Avoids double-bookkeeping
270/// in every macro invocation; just snapshot once before serialization.
271pub fn set_status_counts(warnings: u32, errors: u32, fatal_errors: u32) {
272  STATE.with(|s| {
273    let mut t = s.borrow_mut();
274    t.warnings = warnings;
275    t.errors = errors;
276    t.fatal_errors = fatal_errors;
277  });
278}
279
280// ─── identifiers (binary-set) ───────────────────────────────────────────────
281
282pub fn set_paper_id(id: &str) { STATE.with(|s| s.borrow_mut().paper_id = id.to_string()); }
283pub fn set_cmdline(s: &str) { STATE.with(|st| st.borrow_mut().cmdline = s.to_string()); }
284pub fn set_host(h: &str) { STATE.with(|s| s.borrow_mut().host = h.to_string()); }
285pub fn set_timeout_s(t: u32) { STATE.with(|s| s.borrow_mut().timeout_s = t); }
286pub fn set_category(c: &str) { STATE.with(|s| s.borrow_mut().category = c.to_string()); }
287pub fn set_exit_code(e: i32) { STATE.with(|s| s.borrow_mut().exit_code = e); }
288pub fn set_wall_us(w: u64) { STATE.with(|s| s.borrow_mut().wall_us = w); }
289pub fn set_max_rss_kb(r: u64) { STATE.with(|s| s.borrow_mut().max_rss_kb = r); }
290pub fn set_child_rusage_us(user: u64, sys: u64) {
291  STATE.with(|s| {
292    let mut t = s.borrow_mut();
293    t.child_user_us = user;
294    t.child_sys_us = sys;
295  });
296}
297
298/// Take the current telemetry record, replacing it with a fresh
299/// default. Use at end-of-process to serialize the result.
300pub fn take() -> Telemetry { STATE.with(|s| std::mem::take(&mut *s.borrow_mut())) }
301
302/// Reset per-conversion telemetry WITHOUT reading it.
303///
304/// `take` is the only other reset, and it is called from the binaries' end-of-job
305/// finalizers — which return early when no telemetry sink is configured
306/// (`write_telemetry_record`: `let Some(path) = path else { return }`). So in a
307/// process that converts repeatedly with telemetry off (the `--server` LSP, a
308/// test harness), state carried over between documents. That was self-correcting
309/// while every counter was `set`; it is not once any counter accumulates —
310/// streaming's `add_formulae` must sum across segments, so it would sum across
311/// DOCUMENTS too. Phase timings had the same latent flaw.
312pub fn reset() { STATE.with(|s| *s.borrow_mut() = Telemetry::default()); }
313
314/// Read-only view for tests / instrumented assertions.
315pub fn with<R>(f: impl FnOnce(&Telemetry) -> R) -> R { STATE.with(|s| f(&s.borrow())) }
316
317// ─── JSON serialization ─────────────────────────────────────────────────────
318
319fn write_json_string(out: &mut String, s: &str) {
320  out.push('"');
321  for c in s.chars() {
322    match c {
323      '"' => out.push_str("\\\""),
324      '\\' => out.push_str("\\\\"),
325      '\n' => out.push_str("\\n"),
326      '\r' => out.push_str("\\r"),
327      '\t' => out.push_str("\\t"),
328      c if (c as u32) < 0x20 => {
329        use std::fmt::Write;
330        write!(out, "\\u{:04x}", c as u32).unwrap();
331      },
332      c => out.push(c),
333    }
334  }
335  out.push('"');
336}
337
338impl Telemetry {
339  /// Serialize as a single-line JSON object (suitable for JSONL).
340  /// Hand-written to avoid pulling serde into latexml_core.
341  // The `field!` macro always writes `first = false` after emitting; on the
342  // very last field that final write is naturally dead. Silence the warning.
343  #[allow(unused_assignments)]
344  pub fn to_json_line(&self) -> String {
345    use std::fmt::Write;
346    let mut s = String::with_capacity(1024);
347    s.push('{');
348
349    let mut first = true;
350    macro_rules! field {
351      ($name:literal, $val:expr_2021) => {{
352        if !first {
353          s.push(',');
354        }
355        first = false;
356        s.push('"');
357        s.push_str($name);
358        s.push_str("\":");
359        write!(s, "{}", $val).unwrap();
360      }};
361    }
362    macro_rules! field_str {
363      ($name:literal, $val:expr_2021) => {{
364        if !first {
365          s.push(',');
366        }
367        first = false;
368        s.push('"');
369        s.push_str($name);
370        s.push_str("\":");
371        write_json_string(&mut s, $val);
372      }};
373    }
374    macro_rules! field_array_u64 {
375      ($name:literal, $arr:expr_2021) => {{
376        if !first {
377          s.push(',');
378        }
379        first = false;
380        s.push('"');
381        s.push_str($name);
382        s.push_str("\":[");
383        for (i, v) in $arr.iter().enumerate() {
384          if i > 0 {
385            s.push(',');
386          }
387          write!(s, "{}", v).unwrap();
388        }
389        s.push(']');
390      }};
391    }
392    macro_rules! field_array_u32 {
393      ($name:literal, $arr:expr_2021) => {{
394        field_array_u64!($name, $arr);
395      }};
396    }
397
398    field_str!("paper_id", &self.paper_id);
399    field_str!("git_sha", &self.git_sha);
400    field_str!("cmdline", &self.cmdline);
401    field_str!("host", &self.host);
402    field!("timeout_s", self.timeout_s);
403    field!("schema_version", self.schema_version);
404    field!("wall_us", self.wall_us);
405    field_array_u64!("phase_us", &self.phase_us);
406    // Per-phase aliases for grep convenience
407    for (i, val) in self.phase_us.iter().enumerate() {
408      let phase = match i {
409        0 => Phase::Bootstrap,
410        1 => Phase::Digest,
411        2 => Phase::Build,
412        3 => Phase::Rewrite,
413        4 => Phase::MathParse,
414        5 => Phase::PostXmlParse,
415        6 => Phase::PostScan,
416        7 => Phase::Bibliography,
417        8 => Phase::Crossref,
418        9 => Phase::Graphics,
419        10 => Phase::MathImages,
420        11 => Phase::MathmlPres,
421        12 => Phase::MathmlCont,
422        13 => Phase::Split,
423        14 => Phase::Xslt,
424        15 => Phase::Html5Fixups,
425        16 => Phase::Serialize,
426        _ => unreachable!(),
427      };
428      s.push_str(",\"phase_");
429      s.push_str(phase.as_str());
430      s.push_str("_us\":");
431      write!(s, "{}", val).unwrap();
432    }
433    field!("formulae", self.formulae);
434    field!("math_parse_attempts", self.math_parse_attempts);
435    field!("math_parse_count", self.math_parse_count);
436    field_array_u32!("math_parse_buckets", &self.math_parse_buckets);
437    field!("graphics_assets", self.graphics_assets);
438    field!("graphics_subprocess_count", self.graphics_subprocess_count);
439    field!("db_objects", self.db_objects);
440    field!("output_bytes", self.output_bytes);
441    field!("warnings", self.warnings);
442    field!("errors", self.errors);
443    field!("fatal_errors", self.fatal_errors);
444    field!("external_tool_count", self.external_tool_count);
445    field!("max_rss_kb", self.max_rss_kb);
446    field!("child_user_us", self.child_user_us);
447    field!("child_sys_us", self.child_sys_us);
448    field_str!("category", &self.category);
449    field!("exit_code", self.exit_code);
450
451    s.push('}');
452    s
453  }
454}
455
456// ─── tests ──────────────────────────────────────────────────────────────────
457
458#[cfg(test)]
459mod tests {
460  use std::{thread::sleep, time::Duration};
461
462  use super::*;
463
464  #[test]
465  fn bucket_boundaries() {
466    assert_eq!(bucket_for(0), 0);
467    assert_eq!(bucket_for(499), 0);
468    assert_eq!(bucket_for(500), 1);
469    assert_eq!(bucket_for(999), 1);
470    assert_eq!(bucket_for(1_000), 2);
471    assert_eq!(bucket_for(99_999), 7);
472    assert_eq!(bucket_for(100_000), 8);
473    assert_eq!(bucket_for(1_000_000), 8);
474  }
475
476  #[test]
477  fn nested_phases_charge_innermost_only() {
478    let _g_outer = phase(Phase::Bootstrap);
479    sleep(Duration::from_micros(200));
480    {
481      let _g_inner = phase(Phase::Digest);
482      sleep(Duration::from_micros(500));
483    }
484    sleep(Duration::from_micros(200));
485    drop(_g_outer);
486
487    let t = take();
488    // Bootstrap should have accumulated time outside the Digest scope.
489    assert!(
490      t.phase_us[Phase::Bootstrap as usize] >= 300,
491      "bootstrap got {}",
492      t.phase_us[0]
493    );
494    // Digest should have ~500us (with slack for sleep imprecision).
495    assert!(
496      t.phase_us[Phase::Digest as usize] >= 400,
497      "digest got {}",
498      t.phase_us[Phase::Digest as usize]
499    );
500  }
501
502  #[test]
503  fn json_round_trip_basic_fields() {
504    set_paper_id("0901.0001");
505    set_host("test-host");
506    set_timeout_s(120);
507    set_category("ok");
508    set_exit_code(0);
509    incr_formulae();
510    incr_formulae();
511    record_math_parse(750, 3);
512    record_math_parse(50_000, 1);
513    let t = take();
514    let json = t.to_json_line();
515    assert!(json.starts_with('{'));
516    assert!(json.ends_with('}'));
517    assert!(json.contains("\"paper_id\":\"0901.0001\""));
518    assert!(json.contains("\"formulae\":2"));
519    assert!(json.contains("\"math_parse_attempts\":2"));
520    assert!(json.contains("\"math_parse_count\":4"));
521    // 750us → bucket 1 (>= 500, < 1000)
522    // 50_000us → bucket 7 (>= 20_000, < 100_000)
523    assert!(
524      json.contains("\"math_parse_buckets\":[0,1,0,0,0,0,0,1,0]"),
525      "buckets in json: {}",
526      json
527    );
528  }
529
530  #[test]
531  fn json_escapes_string_fields() {
532    set_paper_id("a \"weird\" id\nwith newline");
533    set_cmdline("cmd\twith\ttabs");
534    let t = take();
535    let json = t.to_json_line();
536    assert!(json.contains("\\\"weird\\\""));
537    assert!(json.contains("\\n"));
538    assert!(json.contains("\\t"));
539  }
540
541  /// `reset` must zero accumulating state without a sink configured. The only
542  /// other reset is `take()`, which the binaries skip when no telemetry path is
543  /// set — and streaming's `add_formulae` ACCUMULATES, so without this a
544  /// long-lived process (the `--server` LSP, a worker pool thread) would sum
545  /// formulae and phase time across documents.
546  #[test]
547  fn reset_clears_accumulating_state() {
548    add_formulae(100);
549    phase_enter(Phase::MathParse);
550    phase_exit();
551    with(|t| {
552      assert_eq!(t.formulae, 100);
553    });
554    reset();
555    with(|t| {
556      assert_eq!(
557        t.formulae, 0,
558        "formulae must not survive a conversion boundary"
559      );
560      assert_eq!(
561        t.phase_us,
562        [0; Phase::COUNT],
563        "phase time must not survive either"
564      );
565    });
566  }
567}
latexml_core/telemetry.rs

latexml_core/
telemetry.rs
