harn_vm/vm/
module_phase_timing.rs

1use std::sync::Arc;
2use std::time::{Duration, Instant};
3
4use parking_lot::Mutex;
5use serde::{Deserialize, Serialize};
6
7use super::Vm;
8
9type ModuleProgressObserver = Arc<dyn Fn(ModulePhaseStats) + Send + Sync>;
10
11/// VM-scoped cumulative work-time and cardinality for module preparation and loading.
12///
13/// Concurrent child-VM spans are additive, so durations can exceed enclosing
14/// wall time. The record is attribution, not another top-level phase clock.
15#[non_exhaustive]
16#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
17pub struct ModulePhaseStats {
18    /// Wall time spent compiling module sources after cache misses.
19    pub module_compile_ms: u64,
20    /// Wall time spent reading, hydrating, instantiating, and exporting modules.
21    pub module_load_ms: u64,
22    /// Module source compilations that completed successfully.
23    pub modules_compiled: u64,
24    /// Successful first loads across the VMs in this execution tree.
25    pub modules_loaded: u64,
26}
27
28impl ModulePhaseStats {
29    /// Add two snapshots without overflowing counters or durations.
30    pub fn saturating_add(self, other: Self) -> Self {
31        Self {
32            module_compile_ms: self
33                .module_compile_ms
34                .saturating_add(other.module_compile_ms),
35            module_load_ms: self.module_load_ms.saturating_add(other.module_load_ms),
36            modules_compiled: self.modules_compiled.saturating_add(other.modules_compiled),
37            modules_loaded: self.modules_loaded.saturating_add(other.modules_loaded),
38        }
39    }
40}
41
42#[derive(Debug, Default)]
43struct ModulePhaseAccumulator {
44    compile: Duration,
45    load: Duration,
46    modules_compiled: u64,
47    modules_loaded: u64,
48}
49
50/// Opt-in recorder shared by a root VM and the child VMs it creates.
51#[derive(Clone, Default)]
52pub struct ModulePhaseRecorder {
53    inner: Arc<Mutex<ModulePhaseAccumulator>>,
54    progress_observer: Arc<Mutex<Option<ModuleProgressObserver>>>,
55}
56
57impl std::fmt::Debug for ModulePhaseRecorder {
58    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
59        formatter
60            .debug_struct("ModulePhaseRecorder")
61            .field("stats", &self.snapshot())
62            .field("observed", &self.progress_observer.lock().is_some())
63            .finish()
64    }
65}
66
67impl ModulePhaseRecorder {
68    /// Create an empty recorder independent of any VM.
69    pub fn new() -> Self {
70        Self::default()
71    }
72
73    /// Return a stable value snapshot of the accumulated module work.
74    pub fn snapshot(&self) -> ModulePhaseStats {
75        let stats = self.inner.lock();
76        ModulePhaseStats {
77            module_compile_ms: duration_ms(stats.compile),
78            module_load_ms: duration_ms(stats.load),
79            modules_compiled: stats.modules_compiled,
80            modules_loaded: stats.modules_loaded,
81        }
82    }
83
84    /// Observe successful module preparation and load transitions.
85    ///
86    /// The observer is called outside recorder locks. Merely starting a phase,
87    /// or dropping an unsuccessful compilation span, never produces progress.
88    pub fn set_progress_observer(
89        &self,
90        observer: impl Fn(ModulePhaseStats) + Send + Sync + 'static,
91    ) {
92        *self.progress_observer.lock() = Some(Arc::new(observer));
93    }
94
95    fn notify_progress(&self) {
96        let observer = self.progress_observer.lock().clone();
97        if let Some(observer) = observer {
98            observer(self.snapshot());
99        }
100    }
101
102    pub(crate) fn compile_span(&self) -> ModulePhaseSpan {
103        ModulePhaseSpan::new(self.clone(), ModulePhase::Compile)
104    }
105
106    pub(crate) fn load_span(&self) -> ModulePhaseSpan {
107        ModulePhaseSpan::new(self.clone(), ModulePhase::Load)
108    }
109
110    pub(crate) fn record_module_loaded(&self) {
111        {
112            let mut stats = self.inner.lock();
113            stats.modules_loaded = stats.modules_loaded.saturating_add(1);
114        }
115        self.notify_progress();
116    }
117}
118
119#[derive(Clone, Copy)]
120enum ModulePhase {
121    Compile,
122    Load,
123}
124
125pub(crate) struct ModulePhaseSpan {
126    recorder: ModulePhaseRecorder,
127    phase: ModulePhase,
128    started: Instant,
129    successful_compile: bool,
130}
131
132impl ModulePhaseSpan {
133    fn new(recorder: ModulePhaseRecorder, phase: ModulePhase) -> Self {
134        Self {
135            recorder,
136            phase,
137            started: Instant::now(),
138            successful_compile: false,
139        }
140    }
141
142    pub(crate) fn mark_compile_succeeded(&mut self) {
143        debug_assert!(matches!(self.phase, ModulePhase::Compile));
144        self.successful_compile = true;
145    }
146}
147
148impl Drop for ModulePhaseSpan {
149    fn drop(&mut self) {
150        let elapsed = self.started.elapsed();
151        {
152            let mut stats = self.recorder.inner.lock();
153            match self.phase {
154                ModulePhase::Compile => {
155                    stats.compile = stats.compile.saturating_add(elapsed);
156                    if self.successful_compile {
157                        stats.modules_compiled = stats.modules_compiled.saturating_add(1);
158                    }
159                }
160                ModulePhase::Load => stats.load = stats.load.saturating_add(elapsed),
161            }
162        }
163        if matches!(self.phase, ModulePhase::Compile) && self.successful_compile {
164            self.recorder.notify_progress();
165        }
166    }
167}
168
169fn duration_ms(duration: Duration) -> u64 {
170    duration.as_millis().min(u128::from(u64::MAX)) as u64
171}
172
173impl Vm {
174    /// Enable module phase timing and return the recorder owned by this VM tree.
175    ///
176    /// Calling this more than once preserves the current recording session.
177    pub fn enable_module_phase_timing(&mut self) -> ModulePhaseRecorder {
178        self.module_phase_recorder
179            .get_or_insert_with(ModulePhaseRecorder::new)
180            .clone()
181    }
182
183    pub(crate) fn module_compile_span(&self) -> Option<ModulePhaseSpan> {
184        self.module_phase_recorder
185            .as_ref()
186            .map(ModulePhaseRecorder::compile_span)
187    }
188
189    pub(crate) fn module_load_span(&self) -> Option<ModulePhaseSpan> {
190        self.module_phase_recorder
191            .as_ref()
192            .map(ModulePhaseRecorder::load_span)
193    }
194
195    pub(crate) fn record_module_loaded(&mut self) {
196        if let Some(count) = &mut self.staged_module_load_count {
197            *count = count.saturating_add(1);
198            return;
199        }
200        if let Some(recorder) = &self.module_phase_recorder {
201            recorder.record_module_loaded();
202        }
203    }
204}
205
206#[cfg(test)]
207mod tests {
208    use super::*;
209
210    #[test]
211    fn stats_addition_saturates_each_field() {
212        let left = ModulePhaseStats {
213            module_compile_ms: u64::MAX,
214            module_load_ms: 2,
215            modules_compiled: 3,
216            modules_loaded: u64::MAX,
217        };
218        let right = ModulePhaseStats {
219            module_compile_ms: 1,
220            module_load_ms: 4,
221            modules_compiled: 5,
222            modules_loaded: 1,
223        };
224
225        assert_eq!(
226            left.saturating_add(right),
227            ModulePhaseStats {
228                module_compile_ms: u64::MAX,
229                module_load_ms: 6,
230                modules_compiled: 8,
231                modules_loaded: u64::MAX,
232            }
233        );
234    }
235
236    #[test]
237    fn child_vms_share_recorder_but_baselines_start_disabled() {
238        let mut vm = Vm::new();
239        assert!(vm.module_phase_recorder.is_none());
240
241        let recorder = vm.enable_module_phase_timing();
242        let mut child = vm.child_vm();
243        std::thread::spawn(move || child.record_module_loaded())
244            .join()
245            .expect("child records from another thread");
246
247        assert_eq!(recorder.snapshot().modules_loaded, 1);
248        assert!(vm.baseline().instantiate().module_phase_recorder.is_none());
249    }
250
251    #[test]
252    fn cancelled_span_releases_its_recorder_handle() {
253        let runtime = tokio::runtime::Builder::new_current_thread()
254            .enable_all()
255            .build()
256            .expect("runtime builds");
257        let recorder = ModulePhaseRecorder::new();
258        let task_recorder = recorder.clone();
259
260        runtime.block_on(async {
261            let (started_tx, started_rx) = tokio::sync::oneshot::channel();
262            let task = tokio::spawn(async move {
263                let _span = task_recorder.load_span();
264                let _ = started_tx.send(());
265                std::future::pending::<()>().await;
266            });
267            started_rx.await.expect("span starts");
268            task.abort();
269            assert!(task.await.expect_err("task is cancelled").is_cancelled());
270        });
271
272        assert_eq!(Arc::strong_count(&recorder.inner), 1);
273    }
274
275    #[test]
276    fn observer_only_sees_successful_module_transitions_and_can_reenter_snapshot() {
277        let recorder = ModulePhaseRecorder::new();
278        let observed = Arc::new(Mutex::new(Vec::new()));
279        let callback_observed = observed.clone();
280        let callback_recorder = recorder.clone();
281        recorder.set_progress_observer(move |stats| {
282            assert_eq!(stats, callback_recorder.snapshot());
283            callback_observed.lock().push(stats);
284        });
285
286        drop(recorder.compile_span());
287        assert!(
288            observed.lock().is_empty(),
289            "failed compilation is not progress"
290        );
291
292        let mut compile = recorder.compile_span();
293        compile.mark_compile_succeeded();
294        drop(compile);
295        recorder.record_module_loaded();
296
297        let observed = observed.lock();
298        assert_eq!(observed.len(), 2);
299        assert_eq!(observed[0].modules_compiled, 1);
300        assert_eq!(observed[0].modules_loaded, 0);
301        assert_eq!(observed[1].modules_loaded, 1);
302    }
303
304    #[test]
305    fn nested_module_loads_are_additive_across_real_child_vms() {
306        let runtime = tokio::runtime::Builder::new_current_thread()
307            .enable_all()
308            .build()
309            .expect("runtime builds");
310        let temp = tempfile::tempdir().expect("tempdir");
311        let dependency = temp.path().join("dependency.harn");
312        let importer = temp.path().join("importer.harn");
313        std::fs::write(&dependency, "pub fn value() { return 42 }\n").expect("write dependency");
314        std::fs::write(
315            &importer,
316            "import { value } from \"./dependency\"\npub fn answer() { return value() }\n",
317        )
318        .expect("write importer");
319
320        let mut vm = Vm::new();
321        let recorder = vm.enable_module_phase_timing();
322        let mut first = vm.child_vm();
323        let mut second = vm.child_vm();
324
325        runtime.block_on(async {
326            first
327                .load_module_exports(&importer)
328                .await
329                .expect("first nested load succeeds");
330            assert_eq!(recorder.snapshot().modules_loaded, 2);
331
332            second
333                .load_module_exports(&importer)
334                .await
335                .expect("second nested load succeeds");
336        });
337
338        assert_eq!(recorder.snapshot().modules_loaded, 4);
339    }
340}
harn_vm/vm/module_phase_timing.rs

harn_vm/vm/
module_phase_timing.rs
