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    }
151    Listener { _private: () }
152}
153
154impl Drop for Listener {
155    fn drop(&mut self) {
156        #[cfg(not(target_arch = "wasm32"))]
157        state::LISTENERS.fetch_sub(1, Ordering::AcqRel);
158    }
159}
160
161/// True while a [`Listener`] lives.
162#[inline]
163pub fn listening() -> bool {
164    #[cfg(not(target_arch = "wasm32"))]
165    {
166        state::LISTENERS.load(Ordering::Relaxed) > 0
167    }
168    #[cfg(target_arch = "wasm32")]
169    false
170}
171
172/// Enter `phase`: `done` restarts at 0 against `total` (`None` when nothing
173/// bounds it).
174pub fn begin(phase: Phase, total: Option<u64>, unit: Unit) {
175    #[cfg(not(target_arch = "wasm32"))]
176    if listening() {
177        use state::*;
178        let idx = phase_index(phase);
179        DONE.store(0, Ordering::Relaxed);
180        TOTAL.store(total.unwrap_or(u64::MAX), Ordering::Relaxed);
181        UNIT.store(unit as u8, Ordering::Relaxed);
182        PHASE.store(idx as u8 + 1, Ordering::Relaxed);
183        if let Ok(mut b) = BEGUN.lock() {
184            b.push((phase, std::time::Instant::now()));
185        }
186        SEQ.fetch_add(1, Ordering::Release);
187    }
188    #[cfg(target_arch = "wasm32")]
189    let _ = (phase, total, unit);
190}
191
192/// Set the current phase's total once it is known.
193pub fn set_total(total: Option<u64>, unit: Unit) {
194    #[cfg(not(target_arch = "wasm32"))]
195    if listening() {
196        state::TOTAL.store(total.unwrap_or(u64::MAX), Ordering::Relaxed);
197        state::UNIT.store(unit as u8, Ordering::Relaxed);
198    }
199    #[cfg(target_arch = "wasm32")]
200    let _ = (total, unit);
201}
202
203/// Add `n` to the current phase's `done`.
204#[inline]
205pub fn advance(n: u64) {
206    #[cfg(not(target_arch = "wasm32"))]
207    if listening() {
208        state::DONE.fetch_add(n, Ordering::Relaxed);
209    }
210    #[cfg(target_arch = "wasm32")]
211    let _ = n;
212}
213
214/// Set the current phase's `done`.
215pub fn set_done(n: u64) {
216    #[cfg(not(target_arch = "wasm32"))]
217    if listening() {
218        state::DONE.store(n, Ordering::Relaxed);
219    }
220    #[cfg(target_arch = "wasm32")]
221    let _ = n;
222}
223
224/// Running totals a renderer shows beside the phase.
225#[derive(Debug, Clone, Copy)]
226pub enum Stat {
227    /// Statements read so far.
228    Statements,
229    /// Distinct dictionary terms.
230    Terms,
231    /// Chunks sealed by the memory-bounded build.
232    Chunks,
233}
234
235/// Set a running total.
236#[inline]
237pub fn set_stat(stat: Stat, v: u64) {
238    #[cfg(not(target_arch = "wasm32"))]
239    if listening() {
240        match stat {
241            Stat::Statements => &state::STATEMENTS,
242            Stat::Terms => &state::TERMS,
243            Stat::Chunks => &state::CHUNKS,
244        }
245        .store(v, Ordering::Relaxed);
246    }
247    #[cfg(target_arch = "wasm32")]
248    let _ = (stat, v);
249}
250
251/// Name the directory the build spills into (`None` once it is gone), so a
252/// renderer can show how much it holds.
253pub fn set_spill_dir(dir: Option<&std::path::Path>) {
254    #[cfg(not(target_arch = "wasm32"))]
255    if listening() {
256        if let Ok(mut d) = state::SPILL_DIR.lock() {
257            *d = dir.map(std::path::Path::to_path_buf);
258        }
259    }
260    #[cfg(target_arch = "wasm32")]
261    let _ = dir;
262}
263
264/// Read the published state. The phase, its `done`, `total` and `unit` are
265/// read as one: a sample never pairs a new phase with the last one's figures.
266pub fn snapshot() -> Snapshot {
267    #[cfg(not(target_arch = "wasm32"))]
268    {
269        use state::*;
270        loop {
271            let seq = SEQ.load(Ordering::Acquire);
272            let phase = PHASE.load(Ordering::Relaxed);
273            let done = DONE.load(Ordering::Relaxed);
274            let total = TOTAL.load(Ordering::Relaxed);
275            let unit = UNIT.load(Ordering::Relaxed);
276            let begun = BEGUN.lock().map(|b| b.clone()).unwrap_or_default();
277            if SEQ.load(Ordering::Acquire) != seq {
278                continue;
279            }
280            return Snapshot {
281                phase: (phase as usize)
282                    .checked_sub(1)
283                    .and_then(|i| Phase::ALL.get(i).copied()),
284                seq,
285                begun,
286                done,
287                total: (total != u64::MAX).then_some(total),
288                unit: Unit::ALL.get(unit as usize).copied().unwrap_or_default(),
289                statements: STATEMENTS.load(Ordering::Relaxed),
290                terms: TERMS.load(Ordering::Relaxed),
291                chunks: CHUNKS.load(Ordering::Relaxed),
292                spill_dir: SPILL_DIR.lock().ok().and_then(|d| d.clone()),
293            };
294        }
295    }
296    #[cfg(target_arch = "wasm32")]
297    Snapshot::default()
298}
299
300/// A per-record counter for a hot loop: [`tick`](Self::tick) adds to a local
301/// integer and publishes ([`advance`]) once per [`Self::EVERY`] records, and
302/// what is left when it is dropped.
303#[derive(Debug, Default)]
304pub struct Ticker {
305    pending: u64,
306}
307
308impl Ticker {
309    /// Records between two publications.
310    pub const EVERY: u64 = 1 << 16;
311
312    /// A counter with nothing counted.
313    pub fn new() -> Self {
314        Ticker { pending: 0 }
315    }
316
317    /// Count `n` records.
318    #[inline]
319    pub fn tick(&mut self, n: u64) {
320        #[cfg(not(target_arch = "wasm32"))]
321        {
322            self.pending += n;
323            if self.pending >= Self::EVERY {
324                self.flush();
325            }
326        }
327        #[cfg(target_arch = "wasm32")]
328        let _ = n;
329    }
330
331    /// Publish what has been counted.
332    pub fn flush(&mut self) {
333        if self.pending > 0 {
334            advance(self.pending);
335            self.pending = 0;
336        }
337    }
338}
339
340impl Drop for Ticker {
341    fn drop(&mut self) {
342        self.flush();
343    }
344}
345
346#[cfg(all(test, not(target_arch = "wasm32")))]
347mod tests {
348    use super::*;
349
350    /// The one test in this crate that listens: the state is process-wide, so
351    /// a second listening test running in parallel would see this one's
352    /// figures.
353    #[test]
354    fn publishes_only_while_listened_to() {
355        begin(Phase::Parse, Some(10), Unit::Bytes);
356        advance(5);
357        assert_eq!(snapshot().phase, None, "nothing before a listener");
358        {
359            let mut t = Ticker::new();
360            t.tick(3);
361        }
362        let l = listen();
363        assert!(listening());
364        begin(Phase::ParseChunk, None, Unit::Bytes);
365        set_total(Some(100), Unit::Bytes);
366        advance(40);
367        set_stat(Stat::Statements, 7);
368        let s = snapshot();
369        assert_eq!(s.phase, Some(Phase::ParseChunk));
370        assert_eq!((s.done, s.total, s.unit), (40, Some(100), Unit::Bytes));
371        assert_eq!(s.statements, 7);
372        assert!(s.has_begun(Phase::ParseChunk) && !s.has_begun(Phase::Parse));
373        let seq = s.seq;
374        begin(Phase::Dictionary, Some(1), Unit::Terms);
375        {
376            let mut t = Ticker::new();
377            for _ in 0..(Ticker::EVERY + 5) {
378                t.tick(1);
379            }
380            assert_eq!(snapshot().done, Ticker::EVERY, "published once per EVERY");
381        }
382        let s = snapshot();
383        assert_eq!(s.done, Ticker::EVERY + 5, "the rest on drop");
384        assert_eq!(s.seq, seq + 1);
385        assert_eq!(s.unit.as_str(), "terms");
386        drop(l);
387        assert!(!listening());
388        begin(Phase::Write, None, Unit::None);
389        assert_eq!(
390            snapshot().phase,
391            Some(Phase::Dictionary),
392            "frozen once dropped"
393        );
394    }
395}
rete_core/build_progress.rs

rete_core/
build_progress.rs
