sva_engine/cache/
watch.rs

1// Concern: tells a tier's one subscriber what each call over it is doing, working and at rest | Non-concern: counting a lookup, what a call computes | IO: (phase, samples, work, tally) -> EngineStats
2
3use std::cell::Cell;
4use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
5use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
6
7use super::memory::{Counters, Memory};
8use super::stats::Tally;
9
10#[derive(Clone, Copy, Debug, PartialEq, Eq)]
11pub enum Kind {
12    Render,
13    Stream,
14    Edit,
15    Persist,
16}
17
18/// Samples are delivered in `Delivering` alone; the last three end a call.
19#[derive(Clone, Copy, Debug, PartialEq, Eq)]
20pub enum Phase {
21    Parsing,
22    Typing,
23    LookingUp,
24    Planning,
25    /// Stored samples read ahead of the first block.
26    Fetching,
27    Delivering,
28    /// Past the last sample.
29    Finishing,
30    Storing,
31    Done,
32    Failed,
33    Abandoned,
34}
35
36/// Samples of values: one unit for what memory answered and what was computed.
37#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
38pub struct Gauge {
39    /// Answered by what memory or the disk held before the call began, each once a call.
40    pub cached_samples: u64,
41    /// Those of `cached_samples` the disk answered.
42    pub stored_samples: u64,
43    /// Summed over every value its pulls computed.
44    pub computed_samples: u64,
45}
46
47impl Gauge {
48    pub fn done(&self) -> u64 {
49        self.cached_samples + self.computed_samples
50    }
51
52    pub fn plus(self, more: Gauge) -> Gauge {
53        self.each(more, u64::wrapping_add)
54    }
55
56    fn minus(self, less: Gauge) -> Gauge {
57        self.each(less, u64::wrapping_sub)
58    }
59
60    fn each(self, other: Gauge, by: fn(u64, u64) -> u64) -> Gauge {
61        Gauge {
62            cached_samples: by(self.cached_samples, other.cached_samples),
63            stored_samples: by(self.stored_samples, other.stored_samples),
64            computed_samples: by(self.computed_samples, other.computed_samples),
65        }
66    }
67}
68
69/// A call's work, and all it expects: `None` over a range with no end.
70#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
71pub struct Progress {
72    pub work: Gauge,
73    pub planned_samples: Option<u64>,
74}
75
76#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
77pub struct Typed {
78    pub instances: usize,
79    /// Those this call types.
80    pub nodes: usize,
81    pub typed: usize,
82}
83
84#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
85pub struct Worked {
86    pub samples: u64,
87    pub work: Gauge,
88    pub lookups: Tally,
89    /// Memory's disk traffic meanwhile, whichever call asked.
90    pub tier: Counters,
91}
92
93#[derive(Clone, Debug, PartialEq, Eq)]
94pub struct CallStats {
95    pub id: u64,
96    pub kind: Kind,
97    pub target: String,
98    pub phase: Phase,
99    /// `None` until its range is decided, and for an open end.
100    pub total_samples: Option<u64>,
101    /// What `counts.work.done()` ends at, where its range ends.
102    pub planned_samples: Option<u64>,
103    pub typing: Option<Typed>,
104    pub counts: Worked,
105}
106
107/// A step of the engine starting.
108#[derive(Clone, Copy, Debug, PartialEq, Eq)]
109pub enum BootPhase {
110    StartingThreads,
111    ThreadsStarted,
112    ThreadsFailed,
113    OpeningStore,
114    ReadingIndex,
115    IndexRead,
116    StoreFailed,
117    Opened,
118}
119
120#[derive(Clone, Copy, Debug, PartialEq, Eq)]
121pub struct StoreBoot {
122    pub max_bytes: u64,
123    /// The index file's size, its entries and their bytes.
124    pub read: Option<(u64, usize, u64)>,
125}
126
127#[derive(Clone, Copy, Debug, PartialEq, Eq)]
128pub struct Boot {
129    pub phase: BootPhase,
130    pub threads: Option<usize>,
131    pub store: Option<StoreBoot>,
132}
133
134impl Boot {
135    pub fn under_way(&self) -> bool {
136        use BootPhase::{Opened, ThreadsFailed, ThreadsStarted};
137        !matches!(self.phase, ThreadsStarted | ThreadsFailed | Opened)
138    }
139}
140
141#[derive(Clone, Debug, PartialEq, Eq)]
142pub struct EngineStats {
143    /// Renders, opens, edits and persists in flight, a stream's reads never: zero is idle.
144    pub working: usize,
145    /// No tier reports one.
146    pub boot: Option<Boot>,
147    /// `None` in the one report a subscription opens with.
148    pub call: Option<CallStats>,
149    pub totals: Worked,
150}
151
152pub type Hears = Arc<dyn Fn(&EngineStats) + Send + Sync>;
153
154struct Heard {
155    hears: Hears,
156    since: Counters,
157    done: Told,
158}
159
160#[derive(Clone, Copy, Default)]
161struct Told {
162    samples: u64,
163    work: Gauge,
164    lookups: Tally,
165}
166
167/// Unsubscribed, a call reads one flag at each point it would report.
168pub(crate) struct Watch {
169    memory: Memory,
170    heard: Mutex<Option<Heard>>,
171    subscribed: AtomicBool,
172    calls: AtomicU64,
173    working: AtomicUsize,
174}
175
176impl Watch {
177    pub(crate) fn over(memory: Memory) -> Watch {
178        Watch {
179            memory,
180            heard: Mutex::new(None),
181            subscribed: AtomicBool::new(false),
182            calls: AtomicU64::new(0),
183            working: AtomicUsize::new(0),
184        }
185    }
186
187    fn heard(&self) -> MutexGuard<'_, Option<Heard>> {
188        self.heard.lock().unwrap_or_else(PoisonError::into_inner)
189    }
190
191    pub(crate) fn subscribe(&self, hears: Hears) {
192        let since = self.memory.counters();
193        *self.heard() = Some(Heard {
194            hears,
195            since,
196            done: Told::default(),
197        });
198        self.subscribed.store(true, Ordering::Relaxed);
199        self.tell(None);
200    }
201
202    pub(crate) fn unsubscribe(&self) {
203        self.subscribed.store(false, Ordering::Relaxed);
204        *self.heard() = None;
205    }
206
207    /// The subscriber is called with nothing held, so it may subscribe anew or call the engine.
208    fn tell(&self, call: Option<(&Call, Told)>) {
209        let mut heard = self.heard();
210        let Some(held) = heard.as_mut() else {
211            return;
212        };
213        let now = self.memory.counters();
214        let call = call.map(|(call, more)| {
215            held.done.samples = held.done.samples.wrapping_add(more.samples);
216            held.done.work = held.done.work.plus(more.work);
217            held.done.lookups = held.done.lookups.plus(more.lookups);
218            call.stats(now)
219        });
220        let stats = EngineStats {
221            working: self.working.load(Ordering::Relaxed),
222            boot: None,
223            call,
224            totals: counts(held.done, now.since(held.since)),
225        };
226        let hears = Arc::clone(&held.hears);
227        drop(heard);
228        hears(&stats);
229    }
230}
231
232fn counts(told: Told, tier: Counters) -> Worked {
233    Worked {
234        samples: told.samples,
235        work: told.work,
236        lookups: told.lookups,
237        tier,
238    }
239}
240
241pub struct Call {
242    watch: Arc<Watch>,
243    id: u64,
244    kind: Kind,
245    target: String,
246    since: Counters,
247    phase: Cell<Phase>,
248    total_samples: Cell<Option<u64>>,
249    planned_samples: Cell<Option<u64>>,
250    typing: Cell<Option<Typed>>,
251    marked: Cell<(u64, usize)>,
252    told: Cell<Told>,
253    summed: Cell<Told>,
254    reported: Cell<u64>,
255    occupies: Cell<bool>,
256    ended: Cell<bool>,
257}
258
259impl Call {
260    pub(crate) fn begin(watch: &Arc<Watch>, kind: Kind, target: &str, phase: Phase) -> Call {
261        watch.working.fetch_add(1, Ordering::Relaxed);
262        let call = Call {
263            watch: Arc::clone(watch),
264            id: watch.calls.fetch_add(1, Ordering::Relaxed) + 1,
265            kind,
266            target: target.to_string(),
267            since: watch.memory.counters(),
268            phase: Cell::new(phase),
269            total_samples: Cell::new(None),
270            planned_samples: Cell::new(None),
271            typing: Cell::new(None),
272            marked: Cell::new((0, 0)),
273            told: Cell::new(Told::default()),
274            summed: Cell::new(Told::default()),
275            reported: Cell::new(0),
276            occupies: Cell::new(true),
277            ended: Cell::new(false),
278        };
279        call.report();
280        call
281    }
282
283    pub fn kind(&self) -> Kind {
284        self.kind
285    }
286
287    pub fn target(&self) -> &str {
288        &self.target
289    }
290
291    pub fn subscribed(&self) -> bool {
292        self.watch.subscribed.load(Ordering::Relaxed)
293    }
294
295    fn stats(&self, now: Counters) -> CallStats {
296        CallStats {
297            id: self.id,
298            kind: self.kind,
299            target: self.target.clone(),
300            phase: self.phase.get(),
301            total_samples: self.total_samples.get(),
302            planned_samples: self.planned_samples.get(),
303            typing: self.typing.get(),
304            counts: counts(self.told.get(), now.since(self.since)),
305        }
306    }
307
308    fn report(&self) {
309        if !self.subscribed() {
310            return;
311        }
312        let (told, summed) = (self.told.get(), self.summed.get());
313        let more = Told {
314            samples: told.samples.wrapping_sub(summed.samples),
315            work: told.work.minus(summed.work),
316            lookups: told.lookups.minus(summed.lookups),
317        };
318        self.summed.set(told);
319        self.reported.set(told.samples);
320        let typed = self.typing.get().map_or(0, |typing| typing.typed);
321        self.marked.set((told.work.done(), typed));
322        self.watch.tell(Some((self, more)));
323    }
324
325    pub fn phase(&self, phase: Phase) {
326        if self.phase.replace(phase) != phase {
327            self.report();
328        }
329    }
330
331    pub(crate) fn looked(&self, phase: Phase, lookups: impl FnOnce() -> Tally) {
332        self.phase.set(phase);
333        if self.subscribed() {
334            let told = self.told.get();
335            self.told.set(Told {
336                lookups: lookups(),
337                ..told
338            });
339            self.report();
340        }
341    }
342
343    /// Reported once a hundredth more of its nodes are typed.
344    pub(crate) fn typed(&self, typing: Typed) {
345        let last = self.typing.replace(Some(typing));
346        if !self.subscribed() {
347            return;
348        }
349        let step = typing.nodes.div_ceil(100).max(1);
350        let due = typing.typed >= self.marked.get().1.saturating_add(step);
351        if last.is_none() || due || typing.typed == typing.nodes {
352            self.report();
353        }
354    }
355
356    /// Reported once `every` samples passed, at the total, and where the phase or total is new.
357    pub(crate) fn delivered(
358        &self,
359        (samples, total_samples, every): (u64, Option<u64>, u64),
360        work: impl FnOnce() -> (Progress, Tally),
361    ) {
362        let entered = self.phase.replace(Phase::Delivering) != Phase::Delivering;
363        let ranged = self.total_samples.replace(total_samples) != total_samples;
364        if !self.subscribed() {
365            return;
366        }
367        let due = samples >= self.reported.get().saturating_add(every);
368        if entered || ranged || due || Some(samples) == total_samples {
369            let (progress, lookups) = work();
370            self.planned_samples.set(progress.planned_samples);
371            self.told.set(Told {
372                samples,
373                work: progress.work,
374                lookups,
375            });
376            self.report();
377        }
378    }
379
380    /// Inside a pull: reported once a hundredth more of the work planned is done.
381    pub(crate) fn worked(&self, progress: Progress) {
382        let Some(planned) = progress.planned_samples.filter(|_| self.subscribed()) else {
383            return;
384        };
385        let step = planned.div_ceil(100).max(1);
386        if progress.work.done() < self.marked.get().0.saturating_add(step) {
387            return;
388        }
389        self.planned_samples.set(Some(planned));
390        self.told.set(Told {
391            work: progress.work,
392            ..self.told.get()
393        });
394        self.report();
395    }
396
397    /// A root memory answered, read by none: all it holds is the call's work.
398    pub(crate) fn answered(&self, cached_samples: u64, from_store: bool, lookups: Tally) {
399        if !self.subscribed() {
400            return;
401        }
402        let work = Gauge {
403            cached_samples,
404            stored_samples: if from_store { cached_samples } else { 0 },
405            computed_samples: 0,
406        };
407        self.planned_samples.set(Some(cached_samples));
408        self.told.set(Told {
409            work,
410            lookups,
411            ..self.told.get()
412        });
413        self.report();
414    }
415
416    /// A stream once open: its reads occupy nothing, and let go from here on it is done.
417    pub(crate) fn rest(&self) {
418        if self.occupies.replace(false) {
419            self.watch.working.fetch_sub(1, Ordering::Relaxed);
420        }
421    }
422
423    pub(crate) fn end(&self, phase: Phase) {
424        if self.ended.replace(true) {
425            return;
426        }
427        if self.occupies.replace(false) {
428            self.watch.working.fetch_sub(1, Ordering::Relaxed);
429        }
430        self.phase.set(phase);
431        self.report();
432    }
433}
434
435impl Drop for Call {
436    fn drop(&mut self) {
437        self.end(match self.occupies.get() {
438            true => Phase::Failed,
439            false => Phase::Done,
440        });
441    }
442}
sva_engine/cache/watch.rs

sva_engine/cache/
watch.rs
