rete_core/
build_progress.rs

1//! **Build progress**: where a running build is, for a renderer on another
2//! thread — what `rete build|merge|recard --progress` draws.
3//!
4//! [`crate::build_trace`] records what a build did, on the build's own thread,
5//! for the build record. This is its live sibling: the builders publish the
6//! phase they enter, how far through it they are, and a few running totals into
7//! process-wide atomics, which a renderer samples whenever it likes.
8//!
9//! Nothing is published until someone [`listen`]s. Until then every call is one
10//! relaxed atomic load and a branch, and a [`Ticker`] — the per-record counter
11//! the hot loops use — adds to a local integer and publishes once per
12//! [`Ticker::EVERY`] records. On wasm32 every function is a no-op.
13//!
14//! The figures are progress, not accounting: a phase's `done` may stop short
15//! of its `total` (a deduplicating merge reads fewer records than were
16//! spilled), and a renderer clamps it to the total.
17
18use crate::build_trace::Phase;
19#[cfg(not(target_arch = "wasm32"))]
20use std::sync::atomic::{AtomicU64, AtomicU8, AtomicUsize, Ordering};
21
22/// What a phase's `done` and `total` count.
23#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
24pub enum Unit {
25    /// Nothing countable: a renderer shows a spinner.
26    #[default]
27    None,
28    /// Input bytes as stored (under any gzip decoder), or bytes written.
29    Bytes,
30    /// Statements read or remapped.
31    Statements,
32    /// Chunk-local dictionary terms merged.
33    Terms,
34    /// Sorted records merged, over every permutation of the phase.
35    Records,
36    /// Starter queries run.
37    Queries,
38}
39
40impl Unit {
41    #[cfg(not(target_arch = "wasm32"))]
42    const ALL: [Unit; 6] = [
43        Unit::None,
44        Unit::Bytes,
45        Unit::Statements,
46        Unit::Terms,
47        Unit::Records,
48        Unit::Queries,
49    ];
50
51    /// The name `--progress=json` prints.
52    pub fn as_str(self) -> &'static str {
53        match self {
54            Unit::None => "none",
55            Unit::Bytes => "bytes",
56            Unit::Statements => "statements",
57            Unit::Terms => "terms",
58            Unit::Records => "records",
59            Unit::Queries => "queries",
60        }
61    }
62}
63
64/// One sample of the published state.
65#[derive(Debug, Clone, Default, PartialEq, Eq)]
66pub struct Snapshot {
67    /// The phase begun last (`None` before the first).
68    pub phase: Option<Phase>,
69    /// How many phases have begun: changes whenever `phase` does.
70    pub seq: u64,
71    /// Every phase begun since [`listen`], with when, in order — so a
72    /// renderer sampling now and then still learns of a phase that began and
73    /// ended between two samples, and how long each one took.
74    pub begun: Vec<(Phase, std::time::Instant)>,
75    /// How far through the phase, in `unit`s.
76    pub done: u64,
77    /// The phase's total, when known.
78    pub total: Option<u64>,
79    /// What `done` and `total` count.
80    pub unit: Unit,
81    /// Statements read so far (the parse), then as counted by the builder.
82    pub statements: u64,
83    /// Distinct dictionary terms, once the dictionary is built.
84    pub terms: u64,
85    /// Chunks the memory-bounded build has sealed.
86    pub chunks: u64,
87    /// The memory-bounded build's spill directory while it exists.
88    pub spill_dir: Option<std::path::PathBuf>,
89}
90
91impl Snapshot {
92    /// Whether `phase` has begun since [`listen`].
93    pub fn has_begun(&self, phase: Phase) -> bool {
94        self.begun.iter().any(|(p, _)| *p == phase)
95    }
96}
97
98/// `phase`'s position in [`Phase::ALL`].
99#[cfg(not(target_arch = "wasm32"))]
100fn phase_index(phase: Phase) -> usize {
101    Phase::ALL.iter().position(|p| *p == phase).unwrap_or(0)
102}
103
104#[cfg(not(target_arch = "wasm32"))]
105mod state {
106    use super::*;
107    pub(super) static LISTENERS: AtomicUsize = AtomicUsize::new(0);
108    /// `Phase::ALL` index + 1; 0 = none yet.
109    pub(super) static PHASE: AtomicU8 = AtomicU8::new(0);
110    pub(super) static SEQ: AtomicU64 = AtomicU64::new(0);
111    pub(super) static BEGUN: std::sync::Mutex<Vec<(Phase, std::time::Instant)>> =
112        std::sync::Mutex::new(Vec::new());
113    pub(super) static DONE: AtomicU64 = AtomicU64::new(0);
114    /// `u64::MAX` = unknown.
115    pub(super) static TOTAL: AtomicU64 = AtomicU64::new(u64::MAX);
116    pub(super) static UNIT: AtomicU8 = AtomicU8::new(0);
117    pub(super) static STATEMENTS: AtomicU64 = AtomicU64::new(0);
118    pub(super) static TERMS: AtomicU64 = AtomicU64::new(0);
119    pub(super) static CHUNKS: AtomicU64 = AtomicU64::new(0);
120    pub(super) static SPILL_DIR: std::sync::Mutex<Option<std::path::PathBuf>> =
121        std::sync::Mutex::new(None);
122}
123
124/// Publishes for as long as it lives; see [`listen`].
125pub struct Listener {
126    _private: (),
127}
128
129/// Start publishing, from a clean slate. Builds that run while the returned
130/// guard lives publish their progress; drop it to stop.
131pub fn listen() -> Listener {
132    #[cfg(not(target_arch = "wasm32"))]
133    {
134        use state::*;
135        PHASE.store(0, Ordering::Relaxed);
136        DONE.store(0, Ordering::Relaxed);
137        TOTAL.store(u64::MAX, Ordering::Relaxed);
138        UNIT.store(0, Ordering::Relaxed);
139        if let Ok(mut b) = BEGUN.lock() {
140            b.clear();
141        }
142        for c in [&STATEMENTS, &TERMS, &CHUNKS] {
143            c.store(0, Ordering::Relaxed);
144        }
145        if let Ok(mut d) = SPILL_DIR.lock() {
146            *d = None;
147        }
148        SEQ.store(0, Ordering::Release);
149        LISTENERS.fetch_add(1, Ordering::AcqRel);
150        #[cfg(test)]
151        THIS_THREAD_LISTENS.with(|c| c.set(true));
152    }
153    Listener { _private: () }
154}
155
156impl Drop for Listener {
157    fn drop(&mut self) {
158        #[cfg(not(target_arch = "wasm32"))]
159        state::LISTENERS.fetch_sub(1, Ordering::AcqRel);
160        #[cfg(test)]
161        THIS_THREAD_LISTENS.with(|c| c.set(false));
162    }
163}
164
165#[cfg(test)]
166thread_local! {
167    static THIS_THREAD_LISTENS: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
168}
169
170/// True while a [`Listener`] lives.
171#[inline]
172pub fn listening() -> bool {
173    #[cfg(not(target_arch = "wasm32"))]
174    {
175        // In this crate's own tests only the thread that listens publishes:
176        // the state is process-wide, and every other test that builds a graph
177        // (they run in parallel) would otherwise advance a listening test's
178        // figures under it.
179        #[cfg(test)]
180        if !THIS_THREAD_LISTENS.with(|c| c.get()) {
181            return false;
182        }
183        state::LISTENERS.load(Ordering::Relaxed) > 0
184    }
185    #[cfg(target_arch = "wasm32")]
186    false
187}
188
189/// Enter `phase`: `done` restarts at 0 against `total` (`None` when nothing
190/// bounds it).
191pub fn begin(phase: Phase, total: Option<u64>, unit: Unit) {
192    #[cfg(not(target_arch = "wasm32"))]
193    if listening() {
194        use state::*;
195        let idx = phase_index(phase);
196        DONE.store(0, Ordering::Relaxed);
197        TOTAL.store(total.unwrap_or(u64::MAX), Ordering::Relaxed);
198        UNIT.store(unit as u8, Ordering::Relaxed);
199        PHASE.store(idx as u8 + 1, Ordering::Relaxed);
200        if let Ok(mut b) = BEGUN.lock() {
201            b.push((phase, std::time::Instant::now()));
202        }
203        SEQ.fetch_add(1, Ordering::Release);
204    }
205    #[cfg(target_arch = "wasm32")]
206    let _ = (phase, total, unit);
207}
208
209/// Set the current phase's total once it is known.
210pub fn set_total(total: Option<u64>, unit: Unit) {
211    #[cfg(not(target_arch = "wasm32"))]
212    if listening() {
213        state::TOTAL.store(total.unwrap_or(u64::MAX), Ordering::Relaxed);
214        state::UNIT.store(unit as u8, Ordering::Relaxed);
215    }
216    #[cfg(target_arch = "wasm32")]
217    let _ = (total, unit);
218}
219
220/// Add `n` to the current phase's `done`.
221#[inline]
222pub fn advance(n: u64) {
223    #[cfg(not(target_arch = "wasm32"))]
224    if listening() {
225        state::DONE.fetch_add(n, Ordering::Relaxed);
226    }
227    #[cfg(target_arch = "wasm32")]
228    let _ = n;
229}
230
231/// Set the current phase's `done`.
232pub fn set_done(n: u64) {
233    #[cfg(not(target_arch = "wasm32"))]
234    if listening() {
235        state::DONE.store(n, Ordering::Relaxed);
236    }
237    #[cfg(target_arch = "wasm32")]
238    let _ = n;
239}
240
241/// Running totals a renderer shows beside the phase.
242#[derive(Debug, Clone, Copy)]
243pub enum Stat {
244    /// Statements read so far.
245    Statements,
246    /// Distinct dictionary terms.
247    Terms,
248    /// Chunks sealed by the memory-bounded build.
249    Chunks,
250}
251
252/// Set a running total.
253#[inline]
254pub fn set_stat(stat: Stat, v: u64) {
255    #[cfg(not(target_arch = "wasm32"))]
256    if listening() {
257        match stat {
258            Stat::Statements => &state::STATEMENTS,
259            Stat::Terms => &state::TERMS,
260            Stat::Chunks => &state::CHUNKS,
261        }
262        .store(v, Ordering::Relaxed);
263    }
264    #[cfg(target_arch = "wasm32")]
265    let _ = (stat, v);
266}
267
268/// Name the directory the build spills into (`None` once it is gone), so a
269/// renderer can show how much it holds.
270pub fn set_spill_dir(dir: Option<&std::path::Path>) {
271    #[cfg(not(target_arch = "wasm32"))]
272    if listening() {
273        if let Ok(mut d) = state::SPILL_DIR.lock() {
274            *d = dir.map(std::path::Path::to_path_buf);
275        }
276    }
277    #[cfg(target_arch = "wasm32")]
278    let _ = dir;
279}
280
281/// Read the published state. The phase, its `done`, `total` and `unit` are
282/// read as one: a sample never pairs a new phase with the last one's figures.
283pub fn snapshot() -> Snapshot {
284    #[cfg(not(target_arch = "wasm32"))]
285    {
286        use state::*;
287        loop {
288            let seq = SEQ.load(Ordering::Acquire);
289            let phase = PHASE.load(Ordering::Relaxed);
290            let done = DONE.load(Ordering::Relaxed);
291            let total = TOTAL.load(Ordering::Relaxed);
292            let unit = UNIT.load(Ordering::Relaxed);
293            let begun = BEGUN.lock().map(|b| b.clone()).unwrap_or_default();
294            if SEQ.load(Ordering::Acquire) != seq {
295                continue;
296            }
297            return Snapshot {
298                phase: (phase as usize)
299                    .checked_sub(1)
300                    .and_then(|i| Phase::ALL.get(i).copied()),
301                seq,
302                begun,
303                done,
304                total: (total != u64::MAX).then_some(total),
305                unit: Unit::ALL.get(unit as usize).copied().unwrap_or_default(),
306                statements: STATEMENTS.load(Ordering::Relaxed),
307                terms: TERMS.load(Ordering::Relaxed),
308                chunks: CHUNKS.load(Ordering::Relaxed),
309                spill_dir: SPILL_DIR.lock().ok().and_then(|d| d.clone()),
310            };
311        }
312    }
313    #[cfg(target_arch = "wasm32")]
314    Snapshot::default()
315}
316
317/// A per-record counter for a hot loop: [`tick`](Self::tick) adds to a local
318/// integer and publishes ([`advance`]) once per [`Self::EVERY`] records, and
319/// what is left when it is dropped.
320#[derive(Debug, Default)]
321pub struct Ticker {
322    pending: u64,
323}
324
325impl Ticker {
326    /// Records between two publications.
327    pub const EVERY: u64 = 1 << 16;
328
329    /// A counter with nothing counted.
330    pub fn new() -> Self {
331        Ticker { pending: 0 }
332    }
333
334    /// Count `n` records.
335    #[inline]
336    pub fn tick(&mut self, n: u64) {
337        #[cfg(not(target_arch = "wasm32"))]
338        {
339            self.pending += n;
340            if self.pending >= Self::EVERY {
341                self.flush();
342            }
343        }
344        #[cfg(target_arch = "wasm32")]
345        let _ = n;
346    }
347
348    /// Publish what has been counted.
349    pub fn flush(&mut self) {
350        if self.pending > 0 {
351            advance(self.pending);
352            self.pending = 0;
353        }
354    }
355}
356
357impl Drop for Ticker {
358    fn drop(&mut self) {
359        self.flush();
360    }
361}
362
363#[cfg(all(test, not(target_arch = "wasm32")))]
364mod tests {
365    use super::*;
366
367    /// The one test in this crate that listens: the state is process-wide, so
368    /// a second listening test running in parallel would see this one's
369    /// figures.
370    #[test]
371    fn publishes_only_while_listened_to() {
372        begin(Phase::Parse, Some(10), Unit::Bytes);
373        advance(5);
374        assert_eq!(snapshot().phase, None, "nothing before a listener");
375        {
376            let mut t = Ticker::new();
377            t.tick(3);
378        }
379        let l = listen();
380        assert!(listening());
381        begin(Phase::ParseChunk, None, Unit::Bytes);
382        set_total(Some(100), Unit::Bytes);
383        advance(40);
384        set_stat(Stat::Statements, 7);
385        let s = snapshot();
386        assert_eq!(s.phase, Some(Phase::ParseChunk));
387        assert_eq!((s.done, s.total, s.unit), (40, Some(100), Unit::Bytes));
388        assert_eq!(s.statements, 7);
389        assert!(s.has_begun(Phase::ParseChunk) && !s.has_begun(Phase::Parse));
390        let seq = s.seq;
391        begin(Phase::Dictionary, Some(1), Unit::Terms);
392        {
393            let mut t = Ticker::new();
394            for _ in 0..(Ticker::EVERY + 5) {
395                t.tick(1);
396            }
397            assert_eq!(snapshot().done, Ticker::EVERY, "published once per EVERY");
398        }
399        let s = snapshot();
400        assert_eq!(s.done, Ticker::EVERY + 5, "the rest on drop");
401        assert_eq!(s.seq, seq + 1);
402        assert_eq!(s.unit.as_str(), "terms");
403        drop(l);
404        assert!(!listening());
405        begin(Phase::Write, None, Unit::None);
406        assert_eq!(
407            snapshot().phase,
408            Some(Phase::Dictionary),
409            "frozen once dropped"
410        );
411    }
412}
rete_core/build_progress.rs

rete_core/
build_progress.rs
