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}
67
68impl Phase {
69    /// Every phase, in declaration order.
70    pub const ALL: [Phase; 18] = [
71        Phase::Read,
72        Phase::Parse,
73        Phase::Reason,
74        Phase::ParseDictionary,
75        Phase::ParseEncode,
76        Phase::ParseChunk,
77        Phase::Dictionary,
78        Phase::Encode,
79        Phase::CardDerive,
80        Phase::Pyramid,
81        Phase::TextIndex,
82        Phase::LabelsSchema,
83        Phase::DictionarySection,
84        Phase::IndexPermutations,
85        Phase::NamedGraphs,
86        Phase::Serialize,
87        Phase::StarterQueries,
88        Phase::Write,
89    ];
90
91    /// The name the build record stores.
92    pub fn as_str(self) -> &'static str {
93        match self {
94            Phase::Read => "read",
95            Phase::Parse => "parse",
96            Phase::Reason => "reason",
97            Phase::ParseDictionary => "parse+dictionary",
98            Phase::ParseEncode => "parse+encode",
99            Phase::ParseChunk => "parse+chunk",
100            Phase::Dictionary => "dictionary",
101            Phase::Encode => "encode",
102            Phase::CardDerive => "card_derive",
103            Phase::Pyramid => "pyramid",
104            Phase::TextIndex => "text_index",
105            Phase::LabelsSchema => "labels+schema",
106            Phase::DictionarySection => "dictionary_section",
107            Phase::IndexPermutations => "index_permutations",
108            Phase::NamedGraphs => "named_graphs",
109            Phase::Serialize => "serialize",
110            Phase::StarterQueries => "starter_queries",
111            Phase::Write => "write",
112        }
113    }
114}
115
116/// What a [`Collector`] recorded.
117#[derive(Debug, Clone, Default, PartialEq, Eq)]
118pub struct Trace {
119    /// Wall-clock time per phase, in the order each phase first ended. A phase
120    /// that ends more than once (a lap per permutation, say) is summed.
121    pub phases: Vec<(Phase, Duration)>,
122    /// Named counters, sorted by name. The memory-bounded build records
123    /// `chunks`, `sort_runs`, `merge_passes` and `scratch_peak_bytes`.
124    pub counters: Vec<(&'static str, u64)>,
125}
126
127impl Trace {
128    /// A counter's value, if it was recorded.
129    pub fn counter(&self, name: &str) -> Option<u64> {
130        self.counters
131            .iter()
132            .find(|(n, _)| *n == name)
133            .map(|&(_, v)| v)
134    }
135}
136
137#[cfg(not(target_arch = "wasm32"))]
138struct State {
139    last: std::time::Instant,
140    trace: Trace,
141}
142
143#[cfg(not(target_arch = "wasm32"))]
144thread_local! {
145    static ACTIVE: RefCell<Option<State>> = const { RefCell::new(None) };
146}
147
148/// Records the build trace of the current thread for as long as it lives.
149/// Starting a second collector on a thread replaces the first.
150pub struct Collector {
151    _not_send: std::marker::PhantomData<*const ()>,
152}
153
154impl Collector {
155    /// Start recording on this thread; the first lap is timed from now.
156    pub fn start() -> Self {
157        #[cfg(not(target_arch = "wasm32"))]
158        ACTIVE.with(|a| {
159            *a.borrow_mut() = Some(State {
160                last: std::time::Instant::now(),
161                trace: Trace::default(),
162            })
163        });
164        Collector {
165            _not_send: std::marker::PhantomData,
166        }
167    }
168
169    /// What has been recorded so far (the collector keeps recording).
170    pub fn snapshot(&self) -> Trace {
171        #[cfg(not(target_arch = "wasm32"))]
172        {
173            ACTIVE.with(|a| {
174                a.borrow()
175                    .as_ref()
176                    .map(|s| s.trace.clone())
177                    .unwrap_or_default()
178            })
179        }
180        #[cfg(target_arch = "wasm32")]
181        Trace::default()
182    }
183}
184
185impl Drop for Collector {
186    fn drop(&mut self) {
187        #[cfg(not(target_arch = "wasm32"))]
188        ACTIVE.with(|a| *a.borrow_mut() = None);
189    }
190}
191
192/// End `phase`: charge it the wall-clock time since the previous lap (or since
193/// the collector started). A no-op without a collector on this thread.
194pub fn lap(phase: Phase) {
195    #[cfg(not(target_arch = "wasm32"))]
196    ACTIVE.with(|a| {
197        if let Some(s) = a.borrow_mut().as_mut() {
198            let now = std::time::Instant::now();
199            let d = now.duration_since(s.last);
200            s.last = now;
201            match s.trace.phases.iter_mut().find(|(p, _)| *p == phase) {
202                Some((_, total)) => *total += d,
203                None => s.trace.phases.push((phase, d)),
204            }
205        }
206    });
207    #[cfg(target_arch = "wasm32")]
208    let _ = phase;
209}
210
211/// Restart the lap clock without charging anything — for time a caller does
212/// not want attributed to the next phase.
213pub fn skip() {
214    #[cfg(not(target_arch = "wasm32"))]
215    ACTIVE.with(|a| {
216        if let Some(s) = a.borrow_mut().as_mut() {
217            s.last = std::time::Instant::now();
218        }
219    });
220}
221
222fn update(name: &'static str, f: impl FnOnce(u64) -> u64) {
223    #[cfg(not(target_arch = "wasm32"))]
224    ACTIVE.with(|a| {
225        if let Some(s) = a.borrow_mut().as_mut() {
226            let c = &mut s.trace.counters;
227            match c.binary_search_by(|(n, _)| (*n).cmp(name)) {
228                Ok(i) => c[i].1 = f(c[i].1),
229                Err(i) => c.insert(i, (name, f(0))),
230            }
231        }
232    });
233    #[cfg(target_arch = "wasm32")]
234    let _ = (name, f);
235}
236
237/// Add `v` to the counter `name`. A no-op without a collector.
238pub fn add(name: &'static str, v: u64) {
239    update(name, |c| c.saturating_add(v));
240}
241
242/// Raise the counter `name` to `v` if it is lower. A no-op without a collector.
243pub fn max(name: &'static str, v: u64) {
244    update(name, |c| c.max(v));
245}
246
247/// One k-way merge over `inputs` spilled, sorted files (the memory-bounded
248/// build): every such merge reads each of its inputs once, start to end.
249#[cfg(not(target_arch = "wasm32"))]
250pub(crate) fn note_merge(inputs: usize) {
251    add("merge_passes", 1);
252    add("sort_runs", inputs as u64);
253}
254
255/// True while a collector records on this thread — for a measurement that
256/// costs something (a directory walk) and is worth skipping otherwise.
257pub fn active() -> bool {
258    #[cfg(not(target_arch = "wasm32"))]
259    {
260        ACTIVE.with(|a| a.borrow().is_some())
261    }
262    #[cfg(target_arch = "wasm32")]
263    false
264}
265
266/// The Cargo features this `rete-core` was compiled with that change what a
267/// build writes or how fast it runs — recorded with the engine identity.
268pub fn enabled_features() -> Vec<&'static str> {
269    let mut f = Vec::new();
270    if cfg!(feature = "compression") {
271        f.push("compression");
272    }
273    if cfg!(feature = "parallel") {
274        f.push("parallel");
275    }
276    if cfg!(feature = "rdf12-turtle") {
277        f.push("rdf12-turtle");
278    }
279    if cfg!(feature = "unicode-regex") {
280        f.push("unicode-regex");
281    }
282    f
283}
284
285#[cfg(all(test, not(target_arch = "wasm32")))]
286mod tests {
287    use super::*;
288
289    #[test]
290    fn laps_partition_and_accumulate_only_under_a_collector() {
291        lap(Phase::Parse);
292        add("chunks", 3);
293        assert!(!active());
294        let c = Collector::start();
295        assert!(active());
296        lap(Phase::Parse);
297        lap(Phase::Encode);
298        lap(Phase::Parse);
299        add("sort_runs", 2);
300        add("sort_runs", 5);
301        max("scratch_peak_bytes", 10);
302        max("scratch_peak_bytes", 4);
303        let t = c.snapshot();
304        let names: Vec<_> = t.phases.iter().map(|(p, _)| p.as_str()).collect();
305        assert_eq!(
306            names,
307            ["parse", "encode"],
308            "first-end order, repeats summed"
309        );
310        assert_eq!(t.counter("sort_runs"), Some(7));
311        assert_eq!(t.counter("scratch_peak_bytes"), Some(10));
312        assert_eq!(t.counter("chunks"), None, "nothing before the collector");
313        drop(c);
314        assert!(!active());
315        lap(Phase::Write);
316    }
317
318    #[test]
319    fn phase_names_are_distinct() {
320        let mut names: Vec<_> = Phase::ALL.iter().map(|p| p.as_str()).collect();
321        names.sort_unstable();
322        names.dedup();
323        assert_eq!(names.len(), Phase::ALL.len());
324    }
325}
rete_core/build_trace.rs

rete_core/
build_trace.rs
