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(&self) {
196        if let Some(recorder) = &self.module_phase_recorder {
197            recorder.record_module_loaded();
198        }
199    }
200}
201
202#[cfg(test)]
203mod tests {
204    use super::*;
205
206    #[test]
207    fn stats_addition_saturates_each_field() {
208        let left = ModulePhaseStats {
209            module_compile_ms: u64::MAX,
210            module_load_ms: 2,
211            modules_compiled: 3,
212            modules_loaded: u64::MAX,
213        };
214        let right = ModulePhaseStats {
215            module_compile_ms: 1,
216            module_load_ms: 4,
217            modules_compiled: 5,
218            modules_loaded: 1,
219        };
220
221        assert_eq!(
222            left.saturating_add(right),
223            ModulePhaseStats {
224                module_compile_ms: u64::MAX,
225                module_load_ms: 6,
226                modules_compiled: 8,
227                modules_loaded: u64::MAX,
228            }
229        );
230    }
231
232    #[test]
233    fn child_vms_share_recorder_but_baselines_start_disabled() {
234        let mut vm = Vm::new();
235        assert!(vm.module_phase_recorder.is_none());
236
237        let recorder = vm.enable_module_phase_timing();
238        let child = vm.child_vm();
239        std::thread::spawn(move || child.record_module_loaded())
240            .join()
241            .expect("child records from another thread");
242
243        assert_eq!(recorder.snapshot().modules_loaded, 1);
244        assert!(vm.baseline().instantiate().module_phase_recorder.is_none());
245    }
246
247    #[test]
248    fn cancelled_span_releases_its_recorder_handle() {
249        let runtime = tokio::runtime::Builder::new_current_thread()
250            .enable_all()
251            .build()
252            .expect("runtime builds");
253        let recorder = ModulePhaseRecorder::new();
254        let task_recorder = recorder.clone();
255
256        runtime.block_on(async {
257            let (started_tx, started_rx) = tokio::sync::oneshot::channel();
258            let task = tokio::spawn(async move {
259                let _span = task_recorder.load_span();
260                let _ = started_tx.send(());
261                std::future::pending::<()>().await;
262            });
263            started_rx.await.expect("span starts");
264            task.abort();
265            assert!(task.await.expect_err("task is cancelled").is_cancelled());
266        });
267
268        assert_eq!(Arc::strong_count(&recorder.inner), 1);
269    }
270
271    #[test]
272    fn observer_only_sees_successful_module_transitions_and_can_reenter_snapshot() {
273        let recorder = ModulePhaseRecorder::new();
274        let observed = Arc::new(Mutex::new(Vec::new()));
275        let callback_observed = observed.clone();
276        let callback_recorder = recorder.clone();
277        recorder.set_progress_observer(move |stats| {
278            assert_eq!(stats, callback_recorder.snapshot());
279            callback_observed.lock().push(stats);
280        });
281
282        drop(recorder.compile_span());
283        assert!(
284            observed.lock().is_empty(),
285            "failed compilation is not progress"
286        );
287
288        let mut compile = recorder.compile_span();
289        compile.mark_compile_succeeded();
290        drop(compile);
291        recorder.record_module_loaded();
292
293        let observed = observed.lock();
294        assert_eq!(observed.len(), 2);
295        assert_eq!(observed[0].modules_compiled, 1);
296        assert_eq!(observed[0].modules_loaded, 0);
297        assert_eq!(observed[1].modules_loaded, 1);
298    }
299
300    #[test]
301    fn nested_module_loads_are_additive_across_real_child_vms() {
302        let runtime = tokio::runtime::Builder::new_current_thread()
303            .enable_all()
304            .build()
305            .expect("runtime builds");
306        let temp = tempfile::tempdir().expect("tempdir");
307        let dependency = temp.path().join("dependency.harn");
308        let importer = temp.path().join("importer.harn");
309        std::fs::write(&dependency, "pub fn value() { return 42 }\n").expect("write dependency");
310        std::fs::write(
311            &importer,
312            "import { value } from \"./dependency\"\npub fn answer() { return value() }\n",
313        )
314        .expect("write importer");
315
316        let mut vm = Vm::new();
317        let recorder = vm.enable_module_phase_timing();
318        let mut first = vm.child_vm();
319        let mut second = vm.child_vm();
320
321        runtime.block_on(async {
322            first
323                .load_module_exports(&importer)
324                .await
325                .expect("first nested load succeeds");
326            assert_eq!(recorder.snapshot().modules_loaded, 2);
327
328            second
329                .load_module_exports(&importer)
330                .await
331                .expect("second nested load succeeds");
332        });
333
334        assert_eq!(recorder.snapshot().modules_loaded, 4);
335    }
336}
harn_vm/vm/module_phase_timing.rs

harn_vm/vm/
module_phase_timing.rs
