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, 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#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
37pub struct Worked {
38    pub samples: u64,
39    /// Summed over every value its pulls computed.
40    pub computed_samples: u64,
41    pub lookups: Tally,
42    /// Memory's disk traffic meanwhile, whichever call asked.
43    pub tier: Counters,
44}
45
46#[derive(Clone, Debug, PartialEq, Eq)]
47pub struct CallStats {
48    pub id: u64,
49    pub kind: Kind,
50    pub target: String,
51    pub phase: Phase,
52    /// `None` until its range is decided, and for an open end.
53    pub total_samples: Option<u64>,
54    pub counts: Worked,
55}
56
57#[derive(Clone, Debug, PartialEq, Eq)]
58pub struct EngineStats {
59    /// Renders, opens, edits and persists in flight, a stream's reads never: zero is idle.
60    pub working: usize,
61    /// `None` in the one report a subscription opens with.
62    pub call: Option<CallStats>,
63    pub totals: Worked,
64}
65
66pub type Hears = Arc<dyn Fn(&EngineStats) + Send + Sync>;
67
68struct Heard {
69    hears: Hears,
70    since: Counters,
71    done: Told,
72}
73
74#[derive(Clone, Copy, Default)]
75struct Told {
76    samples: u64,
77    computed_samples: u64,
78    lookups: Tally,
79}
80
81/// Unsubscribed, a call reads one flag at each point it would report.
82pub(crate) struct Watch {
83    memory: Memory,
84    heard: Mutex<Option<Heard>>,
85    subscribed: AtomicBool,
86    calls: AtomicU64,
87    working: AtomicUsize,
88}
89
90impl Watch {
91    pub(crate) fn over(memory: Memory) -> Watch {
92        Watch {
93            memory,
94            heard: Mutex::new(None),
95            subscribed: AtomicBool::new(false),
96            calls: AtomicU64::new(0),
97            working: AtomicUsize::new(0),
98        }
99    }
100
101    fn heard(&self) -> MutexGuard<'_, Option<Heard>> {
102        self.heard.lock().unwrap_or_else(PoisonError::into_inner)
103    }
104
105    pub(crate) fn subscribe(&self, hears: Hears) {
106        let since = self.memory.counters();
107        *self.heard() = Some(Heard {
108            hears,
109            since,
110            done: Told::default(),
111        });
112        self.subscribed.store(true, Ordering::Relaxed);
113        self.tell(None);
114    }
115
116    pub(crate) fn unsubscribe(&self) {
117        self.subscribed.store(false, Ordering::Relaxed);
118        *self.heard() = None;
119    }
120
121    /// The subscriber is called with nothing held, so it may subscribe anew or call the engine.
122    fn tell(&self, call: Option<(&Call, Told)>) {
123        let mut heard = self.heard();
124        let Some(held) = heard.as_mut() else {
125            return;
126        };
127        let now = self.memory.counters();
128        let call = call.map(|(call, more)| {
129            held.done.samples = held.done.samples.wrapping_add(more.samples);
130            held.done.computed_samples =
131                (held.done.computed_samples).wrapping_add(more.computed_samples);
132            held.done.lookups = held.done.lookups.plus(more.lookups);
133            call.stats(now)
134        });
135        let stats = EngineStats {
136            working: self.working.load(Ordering::Relaxed),
137            call,
138            totals: counts(held.done, now.since(held.since)),
139        };
140        let hears = Arc::clone(&held.hears);
141        drop(heard);
142        hears(&stats);
143    }
144}
145
146fn counts(told: Told, tier: Counters) -> Worked {
147    Worked {
148        samples: told.samples,
149        computed_samples: told.computed_samples,
150        lookups: told.lookups,
151        tier,
152    }
153}
154
155pub struct Call {
156    watch: Arc<Watch>,
157    id: u64,
158    kind: Kind,
159    target: String,
160    since: Counters,
161    phase: Cell<Phase>,
162    total_samples: Cell<Option<u64>>,
163    told: Cell<Told>,
164    summed: Cell<Told>,
165    reported: Cell<u64>,
166    occupies: Cell<bool>,
167    ended: Cell<bool>,
168}
169
170impl Call {
171    pub(crate) fn begin(watch: &Arc<Watch>, kind: Kind, target: &str, phase: Phase) -> Call {
172        watch.working.fetch_add(1, Ordering::Relaxed);
173        let call = Call {
174            watch: Arc::clone(watch),
175            id: watch.calls.fetch_add(1, Ordering::Relaxed) + 1,
176            kind,
177            target: target.to_string(),
178            since: watch.memory.counters(),
179            phase: Cell::new(phase),
180            total_samples: Cell::new(None),
181            told: Cell::new(Told::default()),
182            summed: Cell::new(Told::default()),
183            reported: Cell::new(0),
184            occupies: Cell::new(true),
185            ended: Cell::new(false),
186        };
187        call.report();
188        call
189    }
190
191    pub fn kind(&self) -> Kind {
192        self.kind
193    }
194
195    pub fn target(&self) -> &str {
196        &self.target
197    }
198
199    fn subscribed(&self) -> bool {
200        self.watch.subscribed.load(Ordering::Relaxed)
201    }
202
203    fn stats(&self, now: Counters) -> CallStats {
204        CallStats {
205            id: self.id,
206            kind: self.kind,
207            target: self.target.clone(),
208            phase: self.phase.get(),
209            total_samples: self.total_samples.get(),
210            counts: counts(self.told.get(), now.since(self.since)),
211        }
212    }
213
214    fn report(&self) {
215        if !self.subscribed() {
216            return;
217        }
218        let (told, summed) = (self.told.get(), self.summed.get());
219        let more = Told {
220            samples: told.samples.wrapping_sub(summed.samples),
221            computed_samples: (told.computed_samples).wrapping_sub(summed.computed_samples),
222            lookups: told.lookups.minus(summed.lookups),
223        };
224        self.summed.set(told);
225        self.reported.set(told.samples);
226        self.watch.tell(Some((self, more)));
227    }
228
229    pub fn phase(&self, phase: Phase) {
230        if self.phase.replace(phase) != phase {
231            self.report();
232        }
233    }
234
235    pub(crate) fn looked(&self, phase: Phase, lookups: impl FnOnce() -> Tally) {
236        self.phase.set(phase);
237        if self.subscribed() {
238            let told = self.told.get();
239            self.told.set(Told {
240                lookups: lookups(),
241                ..told
242            });
243            self.report();
244        }
245    }
246
247    /// Reported once `every` samples passed, at the total, and where the phase or total is new.
248    pub(crate) fn delivered(
249        &self,
250        (samples, total_samples, every): (u64, Option<u64>, u64),
251        work: impl FnOnce() -> (u64, Tally),
252    ) {
253        let entered = self.phase.replace(Phase::Delivering) != Phase::Delivering;
254        let ranged = self.total_samples.replace(total_samples) != total_samples;
255        if !self.subscribed() {
256            return;
257        }
258        let due = samples >= self.reported.get().saturating_add(every);
259        if entered || ranged || due || Some(samples) == total_samples {
260            let (computed_samples, lookups) = work();
261            self.told.set(Told {
262                samples,
263                computed_samples,
264                lookups,
265            });
266            self.report();
267        }
268    }
269
270    /// A stream once open: its reads occupy nothing, and let go from here on it is done.
271    pub(crate) fn rest(&self) {
272        if self.occupies.replace(false) {
273            self.watch.working.fetch_sub(1, Ordering::Relaxed);
274        }
275    }
276
277    pub(crate) fn end(&self, phase: Phase) {
278        if self.ended.replace(true) {
279            return;
280        }
281        if self.occupies.replace(false) {
282            self.watch.working.fetch_sub(1, Ordering::Relaxed);
283        }
284        self.phase.set(phase);
285        self.report();
286    }
287}
288
289impl Drop for Call {
290    fn drop(&mut self) {
291        self.end(match self.occupies.get() {
292            true => Phase::Failed,
293            false => Phase::Done,
294        });
295    }
296}
sva_engine/cache/watch.rs

sva_engine/cache/
watch.rs
