ferrum_interfaces/vnext/completion/
checkpoint_timings.rs

1//! Host phases and optional device timing of the existing checkpoint path.
2//! Device samples come only from its terminal receipt, without an extra wait.
3
4use super::{CompletionReaper, StateTransferKind};
5use crate::vnext::{
6    DeviceExecutionTiming, DeviceRuntime, DeviceTimingMeasurement, DeviceTimingMode,
7    DeviceTimingUnavailableReason,
8};
9use serde::Serialize;
10use std::sync::{Arc, Mutex};
11use std::time::{Duration, Instant};
12
13#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize)]
14pub struct CheckpointTimingMeasurement {
15    pub samples: u64,
16    pub total_ns: u64,
17    pub max_ns: u64,
18}
19
20impl CheckpointTimingMeasurement {
21    fn record(&mut self, duration: Duration) {
22        let ns = u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX);
23        self.samples = self.samples.saturating_add(1);
24        self.total_ns = self.total_ns.saturating_add(ns);
25        self.max_ns = self.max_ns.max(ns);
26    }
27}
28
29/// Device elapsed time is separate from host phases, not additive with them.
30/// Samples cover successful native transfers only. Restore elapsed time also
31/// includes any required initialization commands in that same submission.
32#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize)]
33pub struct CheckpointDeviceTimings {
34    pub measured: CheckpointTimingMeasurement,
35    pub not_requested: u64,
36    /// Timing was requested but the backend returned NotRequested rather
37    /// than a measurement or an explicit unavailable reason.
38    pub not_reported: u64,
39    pub unavailable: u64,
40    pub last_unavailable_reason: Option<DeviceTimingUnavailableReason>,
41    pub failed_or_unproven: u64,
42}
43
44/// Physical checkpoint-copy ranges submitted to the backend. These are not
45/// allocator padding or restore initialization bytes, nor proof that every
46/// submitted byte was executed when the transfer fails.
47#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize)]
48pub struct CheckpointCopyMeasurements {
49    pub samples: u64,
50    pub total_bytes: u64,
51    pub max_bytes: u64,
52    pub total_commands: u64,
53    pub max_commands: u64,
54}
55
56#[derive(Clone, Copy)]
57pub(super) struct CheckpointCopyGeometry {
58    pub bytes: u64,
59    pub commands: u64,
60}
61
62#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize)]
63pub struct CheckpointOperationTimings {
64    pub device_execution: CheckpointDeviceTimings,
65    pub submitted_copies: CheckpointCopyMeasurements,
66    /// Submission panicked after device visibility became possible. Kept
67    /// separate from known-submitted copies and never counted as success.
68    pub indeterminate_copies: CheckpointCopyMeasurements,
69    /// Public facade validation and backing claim, including skipped and
70    /// deferred attempts. Private native submission tests bypass this phase.
71    pub prepare_claim: CheckpointTimingMeasurement,
72    /// Native lease/slot preparation, command encoding and submission,
73    /// including definitely-not-submitted and indeterminate outcomes.
74    pub encode_submit: CheckpointTimingMeasurement,
75    /// Existing fence queries, waits and recovery drains. Samples count calls,
76    /// not transfers. Time between observations is not included.
77    pub fence_recovery: CheckpointTimingMeasurement,
78    /// Terminal resource transitions and result outbox publication. Restore
79    /// still requires its consumer's separate conditional frontier commit.
80    pub publication: CheckpointTimingMeasurement,
81}
82
83/// Samples count calls, including empty replacement/eviction lookups, rather
84/// than the number of checkpoint owners released by those calls.
85#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize)]
86pub struct CheckpointCacheTimings {
87    pub maintenance: CheckpointTimingMeasurement,
88    pub replacement_drop: CheckpointTimingMeasurement,
89    pub eviction_drop: CheckpointTimingMeasurement,
90    pub index_publication: CheckpointTimingMeasurement,
91    /// May also record nested native fence/recovery/publication measurements.
92    pub abandoned_recovery: CheckpointTimingMeasurement,
93}
94
95/// Host phase totals may overlap (notably abandoned recovery and the native
96/// phases it invokes). They must not be summed with device execution time.
97#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize)]
98pub struct CheckpointTimingSnapshot {
99    pub capture: CheckpointOperationTimings,
100    pub restore: CheckpointOperationTimings,
101    pub cache: CheckpointCacheTimings,
102}
103
104#[derive(Debug, Clone, Copy)]
105pub enum CheckpointCacheTimingPhase {
106    Maintenance,
107    ReplacementDrop,
108    EvictionDrop,
109    IndexPublication,
110    AbandonedRecovery,
111}
112
113#[derive(Clone, Copy)]
114pub(super) enum CheckpointTimingPhase {
115    PrepareClaim,
116    EncodeSubmit,
117    FenceRecovery,
118    Publication,
119}
120
121#[derive(Default)]
122pub(super) struct CheckpointTimingCounters(Mutex<CheckpointTimingSnapshot>);
123
124impl CheckpointTimingCounters {
125    fn operation(
126        snapshot: &mut CheckpointTimingSnapshot,
127        kind: StateTransferKind,
128    ) -> &mut CheckpointOperationTimings {
129        match kind {
130            StateTransferKind::Capture => &mut snapshot.capture,
131            StateTransferKind::Restore => &mut snapshot.restore,
132        }
133    }
134
135    pub(super) fn record_copies(
136        &self,
137        kind: StateTransferKind,
138        geometry: CheckpointCopyGeometry,
139        indeterminate: bool,
140    ) {
141        let mut snapshot = self
142            .0
143            .lock()
144            .unwrap_or_else(std::sync::PoisonError::into_inner);
145        let operation = Self::operation(&mut snapshot, kind);
146        let copies = if indeterminate {
147            &mut operation.indeterminate_copies
148        } else {
149            &mut operation.submitted_copies
150        };
151        copies.samples = copies.samples.saturating_add(1);
152        copies.total_bytes = copies.total_bytes.saturating_add(geometry.bytes);
153        copies.max_bytes = copies.max_bytes.max(geometry.bytes);
154        copies.total_commands = copies.total_commands.saturating_add(geometry.commands);
155        copies.max_commands = copies.max_commands.max(geometry.commands);
156    }
157
158    pub(super) fn record_device_terminal(
159        &self,
160        kind: StateTransferKind,
161        mode: DeviceTimingMode,
162        succeeded: bool,
163        timing: Option<DeviceTimingMeasurement<DeviceExecutionTiming>>,
164    ) {
165        let mut snapshot = self
166            .0
167            .lock()
168            .unwrap_or_else(std::sync::PoisonError::into_inner);
169        let device = &mut Self::operation(&mut snapshot, kind).device_execution;
170        if !mode.completion_enabled() {
171            device.not_requested = device.not_requested.saturating_add(1);
172        } else if !succeeded {
173            device.failed_or_unproven = device.failed_or_unproven.saturating_add(1);
174        } else {
175            match timing.unwrap_or(DeviceTimingMeasurement::Unavailable(
176                DeviceTimingUnavailableReason::BackendMeasurementFailed,
177            )) {
178                DeviceTimingMeasurement::Measured(timing) => {
179                    device
180                        .measured
181                        .record(Duration::from_nanos(timing.elapsed_ns()));
182                }
183                DeviceTimingMeasurement::NotRequested => {
184                    device.not_reported = device.not_reported.saturating_add(1);
185                }
186                DeviceTimingMeasurement::Unavailable(reason) => {
187                    device.unavailable = device.unavailable.saturating_add(1);
188                    device.last_unavailable_reason = Some(reason);
189                }
190            }
191        }
192    }
193
194    fn record(&self, kind: StateTransferKind, phase: CheckpointTimingPhase, elapsed: Duration) {
195        let mut snapshot = self
196            .0
197            .lock()
198            .unwrap_or_else(std::sync::PoisonError::into_inner);
199        let operation = Self::operation(&mut snapshot, kind);
200        match phase {
201            CheckpointTimingPhase::PrepareClaim => &mut operation.prepare_claim,
202            CheckpointTimingPhase::EncodeSubmit => &mut operation.encode_submit,
203            CheckpointTimingPhase::FenceRecovery => &mut operation.fence_recovery,
204            CheckpointTimingPhase::Publication => &mut operation.publication,
205        }
206        .record(elapsed);
207    }
208
209    pub(super) fn start(
210        self: &Arc<Self>,
211        kind: StateTransferKind,
212        phase: CheckpointTimingPhase,
213    ) -> CheckpointPhaseTimer {
214        CheckpointPhaseTimer {
215            counters: Arc::clone(self),
216            kind,
217            phase,
218            started: Instant::now(),
219        }
220    }
221}
222
223/// Drop also accounts for early errors and unwinding, without retaining the
224/// reaper or touching its slot/lane/resource locks.
225pub(super) struct CheckpointPhaseTimer {
226    counters: Arc<CheckpointTimingCounters>,
227    kind: StateTransferKind,
228    phase: CheckpointTimingPhase,
229    started: Instant,
230}
231
232impl Drop for CheckpointPhaseTimer {
233    fn drop(&mut self) {
234        self.counters
235            .record(self.kind, self.phase, self.started.elapsed());
236    }
237}
238
239impl<R: DeviceRuntime> CompletionReaper<R> {
240    pub fn checkpoint_timing_snapshot(&self) -> CheckpointTimingSnapshot {
241        *self
242            .checkpoint_timings
243            .0
244            .lock()
245            .unwrap_or_else(std::sync::PoisonError::into_inner)
246    }
247
248    /// Resets only observations, never transfer state. For a clean measurement
249    /// interval callers should first quiesce workers: a phase spanning reset is
250    /// recorded in full when it ends, just like other host timing counters.
251    pub fn reset_checkpoint_timings(&self) {
252        *self
253            .checkpoint_timings
254            .0
255            .lock()
256            .unwrap_or_else(std::sync::PoisonError::into_inner) =
257            CheckpointTimingSnapshot::default();
258    }
259
260    pub fn record_checkpoint_cache_timing(
261        &self,
262        phase: CheckpointCacheTimingPhase,
263        elapsed: Duration,
264    ) {
265        let mut snapshot = self
266            .checkpoint_timings
267            .0
268            .lock()
269            .unwrap_or_else(std::sync::PoisonError::into_inner);
270        let cache = &mut snapshot.cache;
271        match phase {
272            CheckpointCacheTimingPhase::Maintenance => &mut cache.maintenance,
273            CheckpointCacheTimingPhase::ReplacementDrop => &mut cache.replacement_drop,
274            CheckpointCacheTimingPhase::EvictionDrop => &mut cache.eviction_drop,
275            CheckpointCacheTimingPhase::IndexPublication => &mut cache.index_publication,
276            CheckpointCacheTimingPhase::AbandonedRecovery => &mut cache.abandoned_recovery,
277        }
278        .record(elapsed);
279    }
280}
281
282#[cfg(test)]
283mod tests {
284    use super::*;
285
286    #[test]
287    fn checkpoint_timing_totals_saturate_without_wrapping() {
288        let mut timing = CheckpointTimingMeasurement::default();
289        timing.record(Duration::from_nanos(7));
290        timing.record(Duration::from_nanos(3));
291        assert_eq!(
292            timing,
293            CheckpointTimingMeasurement {
294                samples: 2,
295                total_ns: 10,
296                max_ns: 7
297            }
298        );
299        timing.record(Duration::MAX);
300        timing.record(Duration::from_nanos(1));
301        assert_eq!(timing.total_ns, u64::MAX);
302        assert_eq!(timing.max_ns, u64::MAX);
303    }
304}
ferrum_interfaces/vnext/completion/checkpoint_timings.rs

ferrum_interfaces/vnext/completion/
checkpoint_timings.rs
