rete_core/
build_trace.rs

1//! The **build trace**: where a build spent its wall clock, and what its
2//! external sorts did — the debugging half of the build record that `rete build`
3//! stores in the kind-7 build-info section.
4//!
5//! It is a lap timer, not a profiler. The builders call [`lap`] at the end of
6//! each phase; the time since the previous lap is charged to that phase, so the
7//! phases **partition** the build's wall clock and never overlap. Counters
8//! ([`add`], [`max`]) record what the memory-bounded build's spill did.
9//!
10//! Recording happens only while a [`Collector`] is alive on the **calling
11//! thread** — the CLI starts one around a build. Every other caller (the
12//! bindings, the browser builder, tests that build in parallel threads) pays
13//! one thread-local check per phase and records nothing. That check also keeps
14//! the clock out of `wasm32-unknown-unknown`, where `std::time::Instant::now()`
15//! panics: without a collector it is never called, and on wasm32 the functions
16//! are no-ops.
17
18#[cfg(not(target_arch = "wasm32"))]
19use std::cell::RefCell;
20use std::time::Duration;
21
22/// One phase of a build. A closed set, so a reader of the record knows every
23/// name it can meet, and a writer can size a record before the build has run.
24#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
25pub enum Phase {
26    /// Reading the input `.rete` back (`repyramid`, `recard`).
27    Read,
28    /// Reading and parsing the inputs into memory (the in-RAM text path).
29    Parse,
30    /// `--materialize` / `--reason`: the reasoner over the default graph.
31    Reason,
32    /// The two-pass streaming build's first pass: read + parse + observe every
33    /// term, then sort the dictionary.
34    ParseDictionary,
35    /// The two-pass streaming build's second pass: read + parse + encode.
36    ParseEncode,
37    /// The memory-bounded build's single pass: read + parse + spill chunks.
38    ParseChunk,
39    /// Building the dictionary from parsed statements (in-RAM), or merging the
40    /// chunk dictionaries (memory-bounded).
41    Dictionary,
42    /// Encoding statements to id-triples (in-RAM), or remapping the chunks'
43    /// local ids to global ones (memory-bounded).
44    Encode,
45    /// Deriving the Dataset Card.
46    CardDerive,
47    /// The community pyramid.
48    Pyramid,
49    /// The full-text index.
50    TextIndex,
51    /// The LABEL_INDEX and SCHEMA sections.
52    LabelsSchema,
53    /// Compressing and serializing the dictionary section.
54    DictionarySection,
55    /// Sorting and tiling the index permutations (and, in-RAM, the named
56    /// graphs' indexes).
57    IndexPermutations,
58    /// The memory-bounded build's named-graph indexes or quad index.
59    NamedGraphs,
60    /// Assembling the file image in memory.
61    Serialize,
62    /// Running the card's starter queries against the finished file.
63    StarterQueries,
64    /// Writing the file to disk.
65    Write,
66    /// Folding the content hash over the written sections for a card placed
67    /// after the data (the memory-bounded build, `rete recard --in-place`):
68    /// the card is hashed first, so the rest of the hash can only be folded
69    /// once it exists — one read of every hashed section.
70    ContentHash,
71}
72
73impl Phase {
74    /// Every phase, in declaration order.
75    pub const ALL: [Phase; 19] = [
76        Phase::Read,
77        Phase::Parse,
78        Phase::Reason,
79        Phase::ParseDictionary,
80        Phase::ParseEncode,
81        Phase::ParseChunk,
82        Phase::Dictionary,
83        Phase::Encode,
84        Phase::CardDerive,
85        Phase::Pyramid,
86        Phase::TextIndex,
87        Phase::LabelsSchema,
88        Phase::DictionarySection,
89        Phase::IndexPermutations,
90        Phase::NamedGraphs,
91        Phase::Serialize,
92        Phase::StarterQueries,
93        Phase::Write,
94        Phase::ContentHash,
95    ];
96
97    /// The name the build record stores.
98    pub fn as_str(self) -> &'static str {
99        match self {
100            Phase::Read => "read",
101            Phase::Parse => "parse",
102            Phase::Reason => "reason",
103            Phase::ParseDictionary => "parse+dictionary",
104            Phase::ParseEncode => "parse+encode",
105            Phase::ParseChunk => "parse+chunk",
106            Phase::Dictionary => "dictionary",
107            Phase::Encode => "encode",
108            Phase::CardDerive => "card_derive",
109            Phase::Pyramid => "pyramid",
110            Phase::TextIndex => "text_index",
111            Phase::LabelsSchema => "labels+schema",
112            Phase::DictionarySection => "dictionary_section",
113            Phase::IndexPermutations => "index_permutations",
114            Phase::NamedGraphs => "named_graphs",
115            Phase::Serialize => "serialize",
116            Phase::StarterQueries => "starter_queries",
117            Phase::Write => "write",
118            Phase::ContentHash => "content_hash",
119        }
120    }
121}
122
123/// What a [`Collector`] recorded.
124#[derive(Debug, Clone, Default, PartialEq, Eq)]
125pub struct Trace {
126    /// Wall-clock time per phase, in the order each phase first ended. A phase
127    /// that ends more than once (a lap per permutation, say) is summed.
128    pub phases: Vec<(Phase, Duration)>,
129    /// Named counters, sorted by name. The memory-bounded build records
130    /// `chunks`, `sort_runs`, `merge_passes` and `scratch_peak_bytes`.
131    pub counters: Vec<(&'static str, u64)>,
132}
133
134impl Trace {
135    /// A counter's value, if it was recorded.
136    pub fn counter(&self, name: &str) -> Option<u64> {
137        self.counters
138            .iter()
139            .find(|(n, _)| *n == name)
140            .map(|&(_, v)| v)
141    }
142}
143
144#[cfg(not(target_arch = "wasm32"))]
145struct State {
146    last: std::time::Instant,
147    trace: Trace,
148}
149
150#[cfg(not(target_arch = "wasm32"))]
151thread_local! {
152    static ACTIVE: RefCell<Option<State>> = const { RefCell::new(None) };
153}
154
155/// Records the build trace of the current thread for as long as it lives.
156/// Starting a second collector on a thread replaces the first.
157pub struct Collector {
158    _not_send: std::marker::PhantomData<*const ()>,
159}
160
161impl Collector {
162    /// Start recording on this thread; the first lap is timed from now.
163    pub fn start() -> Self {
164        #[cfg(not(target_arch = "wasm32"))]
165        ACTIVE.with(|a| {
166            *a.borrow_mut() = Some(State {
167                last: std::time::Instant::now(),
168                trace: Trace::default(),
169            })
170        });
171        Collector {
172            _not_send: std::marker::PhantomData,
173        }
174    }
175
176    /// What has been recorded so far (the collector keeps recording).
177    pub fn snapshot(&self) -> Trace {
178        #[cfg(not(target_arch = "wasm32"))]
179        {
180            ACTIVE.with(|a| {
181                a.borrow()
182                    .as_ref()
183                    .map(|s| s.trace.clone())
184                    .unwrap_or_default()
185            })
186        }
187        #[cfg(target_arch = "wasm32")]
188        Trace::default()
189    }
190}
191
192impl Drop for Collector {
193    fn drop(&mut self) {
194        #[cfg(not(target_arch = "wasm32"))]
195        ACTIVE.with(|a| *a.borrow_mut() = None);
196    }
197}
198
199/// End `phase`: charge it the wall-clock time since the previous lap (or since
200/// the collector started). A no-op without a collector on this thread.
201pub fn lap(phase: Phase) {
202    #[cfg(not(target_arch = "wasm32"))]
203    ACTIVE.with(|a| {
204        if let Some(s) = a.borrow_mut().as_mut() {
205            let now = std::time::Instant::now();
206            let d = now.duration_since(s.last);
207            s.last = now;
208            match s.trace.phases.iter_mut().find(|(p, _)| *p == phase) {
209                Some((_, total)) => *total += d,
210                None => s.trace.phases.push((phase, d)),
211            }
212        }
213    });
214    #[cfg(target_arch = "wasm32")]
215    let _ = phase;
216}
217
218/// Restart the lap clock without charging anything — for time a caller does
219/// not want attributed to the next phase.
220pub fn skip() {
221    #[cfg(not(target_arch = "wasm32"))]
222    ACTIVE.with(|a| {
223        if let Some(s) = a.borrow_mut().as_mut() {
224            s.last = std::time::Instant::now();
225        }
226    });
227}
228
229fn update(name: &'static str, f: impl FnOnce(u64) -> u64) {
230    #[cfg(not(target_arch = "wasm32"))]
231    ACTIVE.with(|a| {
232        if let Some(s) = a.borrow_mut().as_mut() {
233            let c = &mut s.trace.counters;
234            match c.binary_search_by(|(n, _)| (*n).cmp(name)) {
235                Ok(i) => c[i].1 = f(c[i].1),
236                Err(i) => c.insert(i, (name, f(0))),
237            }
238        }
239    });
240    #[cfg(target_arch = "wasm32")]
241    let _ = (name, f);
242}
243
244/// Add `v` to the counter `name`. A no-op without a collector.
245pub fn add(name: &'static str, v: u64) {
246    update(name, |c| c.saturating_add(v));
247}
248
249/// Raise the counter `name` to `v` if it is lower. A no-op without a collector.
250pub fn max(name: &'static str, v: u64) {
251    update(name, |c| c.max(v));
252}
253
254/// One k-way merge over `inputs` spilled, sorted files (the memory-bounded
255/// build): every such merge reads each of its inputs once, start to end.
256#[cfg(not(target_arch = "wasm32"))]
257pub(crate) fn note_merge(inputs: usize) {
258    add("merge_passes", 1);
259    add("sort_runs", inputs as u64);
260}
261
262/// True while a collector records on this thread — for a measurement that
263/// costs something (a directory walk) and is worth skipping otherwise.
264pub fn active() -> bool {
265    #[cfg(not(target_arch = "wasm32"))]
266    {
267        ACTIVE.with(|a| a.borrow().is_some())
268    }
269    #[cfg(target_arch = "wasm32")]
270    false
271}
272
273/// The Cargo features this `rete-core` was compiled with that change what a
274/// build writes or how fast it runs — recorded with the engine identity.
275pub fn enabled_features() -> Vec<&'static str> {
276    let mut f = Vec::new();
277    if cfg!(feature = "compression") {
278        f.push("compression");
279    }
280    if cfg!(feature = "parallel") {
281        f.push("parallel");
282    }
283    if cfg!(feature = "rdf12-turtle") {
284        f.push("rdf12-turtle");
285    }
286    if cfg!(feature = "unicode-regex") {
287        f.push("unicode-regex");
288    }
289    f
290}
291
292#[cfg(all(test, not(target_arch = "wasm32")))]
293mod tests {
294    use super::*;
295
296    #[test]
297    fn laps_partition_and_accumulate_only_under_a_collector() {
298        lap(Phase::Parse);
299        add("chunks", 3);
300        assert!(!active());
301        let c = Collector::start();
302        assert!(active());
303        lap(Phase::Parse);
304        lap(Phase::Encode);
305        lap(Phase::Parse);
306        add("sort_runs", 2);
307        add("sort_runs", 5);
308        max("scratch_peak_bytes", 10);
309        max("scratch_peak_bytes", 4);
310        let t = c.snapshot();
311        let names: Vec<_> = t.phases.iter().map(|(p, _)| p.as_str()).collect();
312        assert_eq!(
313            names,
314            ["parse", "encode"],
315            "first-end order, repeats summed"
316        );
317        assert_eq!(t.counter("sort_runs"), Some(7));
318        assert_eq!(t.counter("scratch_peak_bytes"), Some(10));
319        assert_eq!(t.counter("chunks"), None, "nothing before the collector");
320        drop(c);
321        assert!(!active());
322        lap(Phase::Write);
323    }
324
325    #[test]
326    fn phase_names_are_distinct() {
327        let mut names: Vec<_> = Phase::ALL.iter().map(|p| p.as_str()).collect();
328        names.sort_unstable();
329        names.dedup();
330        assert_eq!(names.len(), Phase::ALL.len());
331    }
332}
rete_core/build_trace.rs

rete_core/
build_trace.rs
