detcore/preemptions/
callback_history.rs

1/*
2 * Copyright (c) Meta Platforms, Inc. and affiliates.
3 * All rights reserved.
4 *
5 * This source code is licensed under the BSD-style license found in the
6 * LICENSE file in the root directory of this source tree.
7 */
8
9//! Software callback expirations in an otherwise unchanged legacy event replay.
10//! These coordinates do not program the PMU or select a runnable owner.
11
12use std::collections::VecDeque;
13
14use serde::Deserialize;
15use serde::Serialize;
16
17use crate::ThreadState;
18use crate::slice_boundary::callbacks;
19use crate::slice_boundary::callbacks::Callback;
20use crate::types::LogicalTime;
21use crate::types::Op;
22use crate::types::SchedEvent;
23
24/// Recording must track actual events before its first optional checkpoint.
25/// Re-recording an old schedule does not manufacture replay authority.
26pub(crate) fn recording_enabled(cfg: &crate::Config) -> bool {
27    cfg.chaos
28        && cfg.record_preemptions
29        && !cfg.strict_schedule_boundaries
30        && cfg.replay_schedule_from.is_none()
31}
32
33#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash, Serialize, Deserialize)]
34/// The original callback location that ended the logical slice.
35pub enum Site {
36    /// Normal expiration before the handler's operation.
37    Prehook,
38    /// Normal expiration after the handler's operation.
39    Posthook,
40    /// Posthook expiration within one RCB of the physical maximum.
41    PosthookPmuEpsilon,
42}
43
44#[derive(Clone, Debug, Eq, PartialEq, Hash, Serialize, Deserialize)]
45#[serde(deny_unknown_fields)]
46/// Observed software-expiry coordinates, independent of the replay tape.
47pub struct Origin {
48    pub(crate) callback: Callback,
49    pub(crate) site: Site,
50    pub(crate) progress: u64,
51    pub(crate) last_op: Option<Op>,
52    pub(crate) time: LogicalTime,
53    pub(crate) rcbs: u64,
54}
55
56#[derive(Clone, Debug, Eq, PartialEq, Hash, Serialize, Deserialize)]
57#[serde(deny_unknown_fields)]
58/// One callback expiry attached to an existing priority-history row.
59pub struct Checkpoint {
60    pub(crate) history_index: usize,
61    pub(crate) origin: Origin,
62}
63
64/// An owned handler identity independent of strict boundary observations.
65/// Nesting keeps the outer callback intact; an early return drops only its token.
66#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
67pub(crate) struct Token {
68    callback: Option<Callback>,
69    boundary: Option<callbacks::Token>,
70    pre_cut: Option<Progress>,
71    post_cut: Option<Progress>,
72}
73
74impl Token {
75    /// Local phase evidence, sampled after the original event update/handoffs.
76    /// It is not a coordinate in the tape: split PMU callbacks may enter with
77    /// different progress before reaching the same positive-event endpoint.
78    pub(crate) fn note_cut(&mut self, progress: Progress, site: Site) {
79        match site {
80            Site::Prehook => self.pre_cut = Some(progress),
81            Site::Posthook => self.post_cut = Some(progress),
82            Site::PosthookPmuEpsilon => {}
83        }
84    }
85
86    fn proves_passed(&self, expected: &Origin, actual: &Origin) -> bool {
87        if self.callback != Some(expected.callback) || actual.callback != expected.callback {
88            return false;
89        }
90        let observed = match (expected.site, actual.site) {
91            (Site::Prehook, Site::Posthook | Site::PosthookPmuEpsilon) => &self.pre_cut,
92            (Site::Posthook, Site::PosthookPmuEpsilon) => &self.post_cut,
93            _ => return false,
94        };
95        observed.as_ref().is_some_and(|progress| {
96            progress.count == expected.progress
97                && progress.last_op == expected.last_op
98                && actual.time >= expected.time
99                && actual.rcbs >= expected.rcbs
100        })
101    }
102}
103
104pub(crate) fn begin<T>(
105    thread: &mut ThreadState<T>,
106    callback: Option<Callback>,
107) -> anyhow::Result<Option<Token>> {
108    let boundary = callbacks::begin(thread, callback)?;
109    Ok(Some(Token {
110        callback,
111        boundary,
112        pre_cut: None,
113        post_cut: None,
114    }))
115}
116
117pub(crate) fn pre<T>(thread: &mut ThreadState<T>, token: &mut Option<Token>) -> anyhow::Result<()> {
118    let mut boundary = token.as_mut().and_then(|token| token.boundary.take());
119    let result = callbacks::pre(thread, &mut boundary);
120    if let Some(token) = token {
121        token.boundary = boundary;
122    }
123    result
124}
125
126pub(crate) fn post<T>(thread: &mut ThreadState<T>, token: &Option<Token>) -> anyhow::Result<()> {
127    callbacks::post(
128        thread,
129        token.as_ref().and_then(|token| token.boundary.clone()),
130    )
131}
132
133pub(crate) fn delivered_signal(
134    token: &mut Option<Token>,
135    signal: crate::types::SigWrapper,
136    delivery: u64,
137) -> anyhow::Result<()> {
138    if let Some(token) = token {
139        callbacks::delivered_signal(&mut token.boundary, signal, delivery)?;
140    }
141    Ok(())
142}
143
144/// Passive progress from actual positive-count events, never the expected tape.
145#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
146pub struct Progress {
147    pub(crate) count: u64,
148    pub(crate) last_op: Option<Op>,
149}
150
151impl Progress {
152    pub(crate) fn observe_event(&mut self, event: &SchedEvent) {
153        if event.count > 0 {
154            self.count = self
155                .count
156                .checked_add(u64::from(event.count))
157                .expect("callback event progress overflow");
158            self.last_op = Some(event.op);
159        }
160    }
161}
162
163/// The existing startup grant restores actual progress after a Tool reload.
164#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
165pub struct Startup {
166    pub(crate) progress: Progress,
167    pub(crate) checkpoints: Vec<Checkpoint>,
168}
169
170#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
171pub(crate) struct State {
172    progress: Progress,
173    checkpoints: VecDeque<Checkpoint>,
174    active: bool,
175}
176
177#[derive(Debug, Eq, PartialEq)]
178pub(crate) enum Decision {
179    Wait,
180    Act(Origin),
181    Mismatch {
182        checkpoint: Checkpoint,
183        hard: bool,
184        actual: Origin,
185    },
186}
187
188/// The priority-history row whose end a thread's current slice waits for,
189/// under `--replay-preemptions-from` (`ThreadState::active_priority_row`).
190#[derive(Clone, Copy, Debug, Eq, PartialEq)]
191pub(crate) enum ActiveRow {
192    /// No recorded row is installed yet.
193    BeforeFirst,
194    /// The slice ends recorded row `n`.
195    Row(usize),
196    /// The final slice, after the last recorded row.
197    Exhausted,
198}
199
200/// Why priority-history replay refuses a recorded callback checkpoint. Each
201/// names the run's policy exit (`exit_on_scheduler_refusal`); none is a
202/// warning, and none falls back to the legacy RCB expiry.
203#[derive(Clone, Copy, Debug, Eq, PartialEq)]
204pub(crate) enum Refusal {
205    /// A checkpoint of a row that already ended was never consumed.
206    Skipped,
207    /// The thread went past the endpoint: more progress or more RCBs than the
208    /// recorded endpoint, or its own callback proves it left that site.
209    Overshot,
210    /// The endpoint's callback, site, progress and RCBs, but another time or
211    /// last operation.
212    Incompatible,
213    /// Checkpoints remain where no recorded row can consume them.
214    Incomplete,
215}
216
217impl Refusal {
218    pub(crate) fn name(self) -> &'static str {
219        match self {
220            Refusal::Skipped => "HERMIT_CALLBACK_CHECKPOINT_SKIPPED",
221            Refusal::Overshot => "HERMIT_CALLBACK_CHECKPOINT_OVERSHOT",
222            Refusal::Incompatible => "HERMIT_CALLBACK_CHECKPOINT_INCOMPATIBLE",
223            Refusal::Incomplete => "HERMIT_CALLBACK_CHECKPOINT_INCOMPLETE",
224        }
225    }
226}
227
228/// What priority-history replay does at one actual callback check-in of a
229/// thread whose record carries callback checkpoints.
230#[derive(Debug, Eq, PartialEq)]
231pub(crate) enum RowDecision {
232    /// The active row has no checkpoint: the legacy expiry decides.
233    NoCheckpoint,
234    /// The row's endpoint is still ahead: the slice does not end here.
235    Pending,
236    /// The actual endpoint of the row, popped: end the slice with it attached.
237    Match(Checkpoint),
238    /// Refuse, with the checkpoint popped from the thread's custody. `actual`
239    /// is `None` at a check-in without a callback identity.
240    Refuse {
241        refusal: Refusal,
242        checkpoint: Checkpoint,
243        actual: Option<Origin>,
244    },
245}
246
247impl State {
248    pub(crate) fn is_active(&self) -> bool {
249        self.active
250    }
251
252    pub(crate) fn progress(&self) -> Progress {
253        self.progress.clone()
254    }
255    pub fn observe_event(&mut self, event: &SchedEvent) {
256        self.progress.observe_event(event);
257    }
258
259    /// Drop the remaining checkpoints: the replay no longer follows the
260    /// schedule they belong to (`Replayer::release_schedule`). The scheduler
261    /// drops its copy at the reply that has the thread call this
262    /// (`Scheduler::schedule_replay_reply`).
263    pub(crate) fn release(&mut self) {
264        self.checkpoints.clear();
265    }
266
267    pub fn install(&mut self, startup: Startup) {
268        if self.active || self.progress != Progress::default() {
269            assert_eq!(
270                self.progress, startup.progress,
271                "preserved Tool progress matches actual coordinator observations"
272            );
273            assert_eq!(
274                self.checkpoints.iter().cloned().collect::<Vec<_>>(),
275                startup.checkpoints,
276                "preserved checkpoint custody matches coordinator"
277            );
278        }
279        self.progress = startup.progress;
280        self.checkpoints = startup.checkpoints.into();
281        // Envelope presence is activation, even at zero progress with no
282        // remaining checkpoints. Exhaustion and exec do not disable tracking.
283        self.active = true;
284    }
285
286    pub fn origin<T>(
287        &self,
288        thread: &ThreadState<T>,
289        token: &Option<Token>,
290        site: Site,
291    ) -> Option<Origin> {
292        Some(Origin {
293            callback: token.as_ref()?.callback?,
294            site,
295            progress: self.progress.count,
296            last_op: self.progress.last_op,
297            time: thread.thread_logical_time.as_nanos(),
298            rcbs: thread.committed_clock_value,
299        })
300    }
301
302    /// The actual coordinates of a check-in at `site` of the callback `token`
303    /// owns, or `None` when the check-in has no callback identity.
304    pub(crate) fn observed<T>(
305        &self,
306        thread: &ThreadState<T>,
307        token: Option<&Token>,
308        site: Site,
309    ) -> Option<Origin> {
310        Some(Origin {
311            callback: token?.callback?,
312            site,
313            progress: self.progress.count,
314            last_op: self.progress.last_op,
315            time: thread.thread_logical_time.as_nanos(),
316            rcbs: thread.committed_clock_value,
317        })
318    }
319
320    pub fn refresh<T>(&self, thread: &ThreadState<T>, origin: &Origin) -> Origin {
321        Origin {
322            callback: origin.callback,
323            site: origin.site,
324            progress: self.progress.count,
325            last_op: self.progress.last_op,
326            time: thread.thread_logical_time.as_nanos(),
327            rcbs: thread.committed_clock_value,
328        }
329    }
330
331    /// Only an independently reached callback endpoint may actuate a checkpoint.
332    /// Earlier/nested callbacks wait. Passing the endpoint reports desync; it
333    /// never authorizes a late administrative operation.
334    #[cfg(test)]
335    pub fn check(&mut self, actual: Origin) -> Decision {
336        self.check_at(actual, None)
337    }
338
339    pub(crate) fn check_at(&mut self, actual: Origin, token: Option<&Token>) -> Decision {
340        let Some(expected) = self.checkpoints.front() else {
341            return Decision::Wait;
342        };
343        if actual.progress < expected.origin.progress {
344            return Decision::Wait;
345        }
346        if actual.progress == expected.origin.progress {
347            if (actual.callback != expected.origin.callback || actual.site != expected.origin.site)
348                && !token.is_some_and(|token| token.proves_passed(&expected.origin, &actual))
349            {
350                return Decision::Wait;
351            }
352            if actual.time < expected.origin.time || actual.rcbs < expected.origin.rcbs {
353                return Decision::Wait;
354            }
355        }
356        let expected = self.checkpoints.pop_front().unwrap();
357        if actual == expected.origin {
358            return Decision::Act(actual);
359        }
360        let hard = actual.progress != expected.origin.progress
361            || actual.callback != expected.origin.callback
362            || actual.site != expected.origin.site
363            || actual.last_op != expected.origin.last_op;
364        Decision::Mismatch {
365            checkpoint: expected,
366            hard,
367            actual,
368        }
369    }
370
371    /// Decide one actual check-in of priority-history replay
372    /// (`--replay-preemptions-from` with callback checkpoints). Unlike
373    /// `check_at`, the global schedule replay's rule, this is row-aware and
374    /// strict:
375    ///
376    /// 1. The raw front first: a front for an earlier row than `active` was
377    ///    skipped; a front for a later row leaves `active` unannotated, as is
378    ///    any row before the first is installed. After the last row, any
379    ///    remaining checkpoint is incomplete.
380    /// 2. Passed evidence: more progress or more RCBs than the endpoint, or
381    ///    the thread's own callback proving it left the endpoint's site.
382    /// 3. At the endpoint's progress and RCBs, the same callback and site end
383    ///    the row when its time and last operation agree too, and refuse when
384    ///    they do not; another callback or site waits (a timer stop or a
385    ///    prehook before a posthook endpoint, or a nested callback).
386    /// 4. Below the endpoint, wait.
387    ///
388    /// A match or a refusal pops the front: the caller acknowledges it to the
389    /// scheduler, whose own copy pops in the same order. A check-in without a
390    /// callback identity (`actual` `None`) is never an endpoint: after step 1
391    /// it applies step 2's progress and RCB tests to the thread's own
392    /// progress and its committed RCBs `rcbs`, refusing as overshot what is
393    /// past the endpoint, and otherwise waits. Without an identity there is no
394    /// callback to prove a site was left. `rcbs` is read only then; an
395    /// identified check-in carries its RCBs in `actual`.
396    pub(crate) fn decide_row(
397        &mut self,
398        active: ActiveRow,
399        actual: Option<Origin>,
400        token: Option<&Token>,
401        rcbs: u64,
402    ) -> RowDecision {
403        let Some(front) = self.checkpoints.front() else {
404            return RowDecision::NoCheckpoint;
405        };
406        let row = match active {
407            ActiveRow::BeforeFirst => return RowDecision::NoCheckpoint,
408            ActiveRow::Exhausted => return self.refuse_front(Refusal::Incomplete, actual),
409            ActiveRow::Row(row) => row,
410        };
411        if front.history_index < row {
412            return self.refuse_front(Refusal::Skipped, actual);
413        }
414        if front.history_index > row {
415            return RowDecision::NoCheckpoint;
416        }
417        let expected = front.origin.clone();
418        let Some(actual) = actual else {
419            if self.progress.count > expected.progress || rcbs > expected.rcbs {
420                return self.refuse_front(Refusal::Overshot, None);
421            }
422            return RowDecision::Pending;
423        };
424        let passed = actual.progress > expected.progress
425            || actual.rcbs > expected.rcbs
426            || (actual.progress == expected.progress
427                && token.is_some_and(|token| token.proves_passed(&expected, &actual)));
428        if passed {
429            return self.refuse_front(Refusal::Overshot, Some(actual));
430        }
431        if actual.progress < expected.progress || actual.rcbs < expected.rcbs {
432            return RowDecision::Pending;
433        }
434        if actual.callback != expected.callback || actual.site != expected.site {
435            return RowDecision::Pending;
436        }
437        if actual.time != expected.time || actual.last_op != expected.last_op {
438            return self.refuse_front(Refusal::Incompatible, Some(actual));
439        }
440        RowDecision::Match(self.checkpoints.pop_front().unwrap())
441    }
442
443    fn refuse_front(&mut self, refusal: Refusal, actual: Option<Origin>) -> RowDecision {
444        RowDecision::Refuse {
445            refusal,
446            checkpoint: self.checkpoints.pop_front().unwrap(),
447            actual,
448        }
449    }
450
451    /// The checkpoints this thread has not consumed, in order.
452    pub(crate) fn remaining(&self) -> impl Iterator<Item = &Checkpoint> {
453        self.checkpoints.iter()
454    }
455
456    /// Whether the next unconsumed checkpoint is that of `row`.
457    pub(crate) fn annotates(&self, row: usize) -> bool {
458        self.checkpoints
459            .front()
460            .is_some_and(|front| front.history_index == row)
461    }
462}
463
464#[cfg(test)]
465mod tests {
466    use detcore_model::schedule::SyscallPhase;
467    use reverie::syscalls::Sysno;
468
469    use super::*;
470    use crate::config::Config;
471    use crate::types::DetTid;
472
473    fn thread() -> ThreadState<()> {
474        ThreadState::new(DetTid::from_raw(3), &Config::default(), ())
475    }
476    fn origin() -> Origin {
477        Origin {
478            callback: Callback::Syscall(Sysno::close),
479            site: Site::Posthook,
480            progress: 2,
481            last_op: Some(Op::Syscall(Sysno::close, SyscallPhase::Posthook)),
482            time: LogicalTime::from_nanos(100),
483            rcbs: 10,
484        }
485    }
486    fn replay(origin: Origin) -> State {
487        let mut state = State::default();
488        state.install(Startup {
489            checkpoints: vec![Checkpoint {
490                history_index: 0,
491                origin,
492            }],
493            ..Startup::default()
494        });
495        state
496    }
497
498    #[test]
499    fn a_released_replay_drops_the_remaining_checkpoints() {
500        let mut state = replay(origin());
501        state.release();
502        assert_eq!(state.check(origin()), Decision::Wait);
503        assert!(state.is_active());
504    }
505
506    #[test]
507    fn callback_checkpoint_requires_phase_progress_and_both_clocks() {
508        let expected = origin();
509        let mut state = replay(expected.clone());
510        for actual in [
511            Origin {
512                site: Site::Prehook,
513                ..expected.clone()
514            },
515            Origin {
516                time: LogicalTime::from_nanos(75),
517                ..expected.clone()
518            },
519            Origin {
520                rcbs: 9,
521                ..expected.clone()
522            },
523            Origin {
524                progress: 1,
525                ..expected.clone()
526            },
527            Origin {
528                callback: Callback::Cpuid,
529                ..expected.clone()
530            },
531        ] {
532            assert_eq!(state.check(actual), Decision::Wait);
533            assert_eq!(state.checkpoints.len(), 1);
534        }
535        assert_eq!(
536            state.check(expected.clone()),
537            Decision::Act(expected.clone())
538        );
539        assert_eq!(state.check(expected), Decision::Wait);
540    }
541
542    #[test]
543    fn callback_checkpoint_missed_endpoint_is_hard_but_clock_drift_is_soft() {
544        let expected = origin();
545        for (actual, hard) in [
546            (
547                Origin {
548                    progress: 3,
549                    ..expected.clone()
550                },
551                true,
552            ),
553            (
554                Origin {
555                    last_op: Some(Op::OtherInstructions),
556                    ..expected.clone()
557                },
558                true,
559            ),
560            (
561                Origin {
562                    time: LogicalTime::from_nanos(101),
563                    ..expected.clone()
564                },
565                false,
566            ),
567            (
568                Origin {
569                    rcbs: 11,
570                    ..expected.clone()
571                },
572                false,
573            ),
574        ] {
575            let mut state = replay(expected.clone());
576            assert!(
577                matches!(state.check(actual), Decision::Mismatch { hard: actual_hard, .. } if actual_hard == hard)
578            );
579            assert!(state.checkpoints.is_empty());
580        }
581    }
582
583    #[test]
584    fn callback_checkpoint_counts_positive_events_and_preserves_nested_handler_identity() {
585        let mut whole = State::default();
586        let mut split = State::default();
587        let tid = DetTid::from_raw(3);
588        whole.observe_event(&SchedEvent::branches(tid, 10));
589        for count in [0, 3, 0, 7, 0] {
590            split.observe_event(&SchedEvent::branches(tid, count));
591        }
592        assert_eq!(whole, split);
593        let mut thread = thread();
594        let mut outer = begin(&mut thread, Some(Callback::Syscall(Sysno::read))).unwrap();
595        pre(&mut thread, &mut outer).unwrap();
596        let mut inner = begin(&mut thread, Some(Callback::Syscall(Sysno::getpid))).unwrap();
597        pre(&mut thread, &mut inner).unwrap();
598        post(&mut thread, &inner).unwrap();
599        assert!(thread.slice_boundary.is_none());
600        assert_eq!(
601            whole
602                .origin(&thread, &outer, Site::Posthook)
603                .unwrap()
604                .callback,
605            Callback::Syscall(Sysno::read)
606        );
607        post(&mut thread, &outer).unwrap();
608        let _abandoned = begin(&mut thread, Some(Callback::Cpuid)).unwrap();
609        assert_eq!(
610            whole
611                .origin(&thread, &inner, Site::PosthookPmuEpsilon)
612                .unwrap()
613                .callback,
614            Callback::Syscall(Sysno::getpid)
615        );
616    }
617
618    #[test]
619    fn callback_checkpoint_passed_owned_phase_is_hard_without_fragment_authority() {
620        let expected = Origin {
621            site: Site::Prehook,
622            progress: 10,
623            last_op: Some(Op::Branch),
624            ..origin()
625        };
626        let tid = DetTid::from_raw(3);
627        let mut whole_thread = thread();
628        let mut split_thread = thread();
629        split_thread
630            .callback_history
631            .observe_event(&SchedEvent::branches(tid, 3));
632        let mut whole = begin(&mut whole_thread, Some(expected.callback))
633            .unwrap()
634            .unwrap();
635        let mut split = begin(&mut split_thread, Some(expected.callback))
636            .unwrap()
637            .unwrap();
638        whole_thread
639            .callback_history
640            .observe_event(&SchedEvent::branches(tid, 10));
641        split_thread
642            .callback_history
643            .observe_event(&SchedEvent::branches(tid, 7));
644        whole.note_cut(whole_thread.callback_history.progress(), Site::Prehook);
645        split.note_cut(split_thread.callback_history.progress(), Site::Prehook);
646        assert_eq!(
647            whole, split,
648            "only the actual cut endpoint is retained locally"
649        );
650        let later = Origin {
651            site: Site::Posthook,
652            ..expected.clone()
653        };
654        let nested = begin(&mut split_thread, Some(expected.callback))
655            .unwrap()
656            .unwrap();
657        let mut state = replay(expected.clone());
658        assert_eq!(state.check_at(later.clone(), Some(&nested)), Decision::Wait);
659        assert_eq!(
660            state.check_at(
661                Origin {
662                    time: LogicalTime::from_nanos(99),
663                    ..later.clone()
664                },
665                Some(&split)
666            ),
667            Decision::Wait
668        );
669        assert!(matches!(
670            state.check_at(later, Some(&split)),
671            Decision::Mismatch { hard: true, .. }
672        ));
673        assert!(state.checkpoints.is_empty());
674        let expected = Origin {
675            site: Site::Posthook,
676            ..expected
677        };
678        whole.note_cut(whole_thread.callback_history.progress(), Site::Posthook);
679        let mut state = replay(expected.clone());
680        assert!(matches!(
681            state.check_at(
682                Origin {
683                    site: Site::PosthookPmuEpsilon,
684                    ..expected
685                },
686                Some(&whole)
687            ),
688            Decision::Mismatch { hard: true, .. }
689        ));
690    }
691
692    #[test]
693    fn callback_checkpoint_wire_retains_full_positional_payload() {
694        let point = Checkpoint {
695            history_index: 7,
696            origin: origin(),
697        };
698        let response = crate::tool_global::GlobalResponse::WithCallbackHistory(
699            Startup {
700                checkpoints: vec![point.clone()],
701                ..Startup::default()
702            },
703            Box::new(crate::tool_global::GlobalResponse::StartNewThread(None)),
704        );
705        let bytes = bincode::serde::encode_to_vec(&response, bincode::config::standard()).unwrap();
706        let (decoded, used): (crate::tool_global::GlobalResponse, usize) =
707            bincode::serde::decode_from_slice(&bytes, bincode::config::standard()).unwrap();
708        assert_eq!(decoded, response);
709        assert_eq!(used, bytes.len());
710        let mut resources = crate::resources::Resources::new(DetTid::from_raw(3));
711        // Boxing the rare request payload must not change either wire format.
712        for inline in [None, Some(point.origin.clone())] {
713            let boxed = inline.clone().map(Box::new);
714            assert_eq!(
715                bincode::serde::encode_to_vec(&inline, bincode::config::standard()).unwrap(),
716                bincode::serde::encode_to_vec(&boxed, bincode::config::standard()).unwrap(),
717            );
718            assert_eq!(
719                serde_json::to_vec(&inline).unwrap(),
720                serde_json::to_vec(&boxed).unwrap(),
721            );
722        }
723        resources.callback_checkpoint = Some(Box::new(point.origin));
724        let bytes = bincode::serde::encode_to_vec(&resources, bincode::config::standard()).unwrap();
725        let (decoded, used): (crate::resources::Resources, usize) =
726            bincode::serde::decode_from_slice(&bytes, bincode::config::standard()).unwrap();
727        assert_eq!(decoded, resources);
728        assert_eq!(used, bytes.len());
729    }
730
731    #[test]
732    fn callback_history_activation_survives_empty_custody_and_refuses_disagreement() {
733        let expected = Origin {
734            progress: 0,
735            last_op: None,
736            ..origin()
737        };
738        let mut state = replay(expected.clone());
739        assert_eq!(state.check(expected.clone()), Decision::Act(expected));
740        assert!(state.checkpoints.is_empty());
741        assert!(state.is_active());
742        state.observe_event(&SchedEvent::branches(DetTid::from_raw(3), 7));
743        let startup = Startup {
744            progress: state.progress(),
745            checkpoints: Vec::new(),
746        };
747        let mut reloaded = State::default();
748        reloaded.install(startup.clone());
749        assert_eq!(reloaded, state);
750        reloaded.install(startup);
751        assert_eq!(reloaded, state);
752        assert!(
753            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
754                reloaded.install(Startup::default());
755            }))
756            .is_err()
757        );
758
759        // An active empty envelope is not an inactive startup. A preserved
760        // zero-progress Tool must refuse new, conflicting checkpoint custody.
761        let mut empty = State::default();
762        empty.install(Startup::default());
763        assert!(empty.is_active());
764        assert!(
765            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
766                empty.install(Startup {
767                    checkpoints: vec![Checkpoint {
768                        history_index: 0,
769                        origin: origin(),
770                    }],
771                    ..Startup::default()
772                });
773            }))
774            .is_err()
775        );
776        for original in [State::default(), state] {
777            let bytes =
778                bincode::serde::encode_to_vec(&original, bincode::config::legacy()).unwrap();
779            let (decoded, used): (State, usize) =
780                bincode::serde::decode_from_slice(&bytes, bincode::config::legacy()).unwrap();
781            assert_eq!(decoded, original);
782            assert_eq!(used, bytes.len());
783        }
784    }
785
786    /// Custody of priority-history checkpoints, each for its recorded row.
787    fn rows(points: &[(usize, Origin)]) -> State {
788        let mut state = State::default();
789        state.install(Startup {
790            checkpoints: points
791                .iter()
792                .map(|(history_index, origin)| Checkpoint {
793                    history_index: *history_index,
794                    origin: origin.clone(),
795                })
796                .collect(),
797            ..Startup::default()
798        });
799        state
800    }
801
802    /// Priority-history replay ends an annotated row only at its recorded
803    /// endpoint: not before the first row is installed, not for a row the
804    /// front does not name (a legal unannotated row), not below the
805    /// endpoint's progress or RCBs, not at another callback or site at the
806    /// endpoint's coordinates (a timer stop or a prehook before a posthook
807    /// endpoint), and not at a check-in without a callback identity. Only a
808    /// match consumes the checkpoint.
809    #[test]
810    fn a_priority_row_ends_only_at_its_recorded_endpoint() {
811        let expected = origin();
812        let mut state = rows(&[(1, expected.clone())]);
813        let endpoint_rcbs = expected.rcbs;
814        let decide = move |state: &mut State, active, actual: Option<Origin>| {
815            let rcbs = actual.as_ref().map_or(endpoint_rcbs, |actual| actual.rcbs);
816            state.decide_row(active, actual, None, rcbs)
817        };
818        assert_eq!(
819            decide(&mut state, ActiveRow::BeforeFirst, Some(expected.clone())),
820            RowDecision::NoCheckpoint
821        );
822        assert_eq!(
823            decide(&mut state, ActiveRow::Row(0), Some(expected.clone())),
824            RowDecision::NoCheckpoint,
825            "row 0 has no checkpoint: the legacy expiry decides it"
826        );
827        for actual in [
828            Origin {
829                progress: 1,
830                ..expected.clone()
831            },
832            Origin {
833                rcbs: 9,
834                ..expected.clone()
835            },
836            Origin {
837                site: Site::Prehook,
838                ..expected.clone()
839            },
840            Origin {
841                callback: Callback::PmuTimer,
842                site: Site::Prehook,
843                ..expected.clone()
844            },
845        ] {
846            assert_eq!(
847                decide(&mut state, ActiveRow::Row(1), Some(actual.clone())),
848                RowDecision::Pending,
849                "{actual:?}"
850            );
851        }
852        assert_eq!(
853            decide(&mut state, ActiveRow::Row(1), None),
854            RowDecision::Pending
855        );
856        assert!(state.annotates(1), "nothing was consumed");
857        assert_eq!(
858            decide(&mut state, ActiveRow::Row(1), Some(expected.clone())),
859            RowDecision::Match(Checkpoint {
860                history_index: 1,
861                origin: expected.clone(),
862            })
863        );
864        assert_eq!(state.remaining().count(), 0);
865        assert_eq!(
866            decide(&mut state, ActiveRow::Exhausted, Some(expected)),
867            RowDecision::NoCheckpoint,
868            "a consumed history ends in the legacy final slice"
869        );
870    }
871
872    /// The exact row of a retained Sapphire Rapids failure
873    /// (hermit_run_chaosreplay__rs_print_nanosleep_race, F1 063d4181, CI job
874    /// 45035997061922517; SPR-DIAGNOSTIC-REVIEW.md). The recording ended dtid
875    /// 7's row 1 at the `write` prehook, progress 10759483, raw RCBs
876    /// 10759329. The unrepaired replay ended it one event earlier, at the PMU
877    /// timer stop that landed exactly on those RCBs with the same time
878    /// (progress 10759482, last operation a branch), before `write` and with
879    /// no conditional branch between them. That stop must leave the row
880    /// pending; only the recorded `write` prehook ends it.
881    #[test]
882    fn the_retained_sapphire_rapids_row_waits_past_its_timer_stop_for_the_write_prehook() {
883        let time = LogicalTime::from_nanos(1_792_852_217_684_087_955);
884        let recorded = Origin {
885            callback: Callback::Syscall(Sysno::write),
886            site: Site::Prehook,
887            progress: 10_759_483,
888            last_op: Some(Op::OtherInstructions),
889            time,
890            rcbs: 10_759_329,
891        };
892        let timer_stop = Origin {
893            callback: Callback::PmuTimer,
894            site: Site::Prehook,
895            progress: 10_759_482,
896            last_op: Some(Op::Branch),
897            time,
898            rcbs: 10_759_329,
899        };
900        let row0_endpoint = Origin {
901            callback: Callback::PmuTimer,
902            site: Site::Prehook,
903            progress: 2_459_633,
904            last_op: Some(Op::Branch),
905            time: LogicalTime::from_nanos(1_792_852_217_600_760_455),
906            rcbs: 2_459_579,
907        };
908        let mut state = rows(&[(0, row0_endpoint.clone()), (1, recorded.clone())]);
909        let RowDecision::Match(row0) =
910            state.decide_row(ActiveRow::Row(0), Some(row0_endpoint), None, 2_459_579)
911        else {
912            panic!("row 0 ends at its recorded timer stop");
913        };
914        assert_eq!(row0.history_index, 0);
915        assert_eq!(
916            state.decide_row(ActiveRow::Row(1), Some(timer_stop), None, 10_759_329),
917            RowDecision::Pending,
918            "the timer stop at the endpoint's RCBs, one event early, must not end row 1"
919        );
920        assert!(state.annotates(1), "row 1's checkpoint was not consumed");
921        assert_eq!(
922            state.decide_row(
923                ActiveRow::Row(1),
924                Some(recorded.clone()),
925                None,
926                recorded.rcbs
927            ),
928            RowDecision::Match(Checkpoint {
929                history_index: 1,
930                origin: recorded,
931            })
932        );
933        assert_eq!(state.remaining().count(), 0);
934    }
935
936    /// Every way past or beside the recorded endpoint is a named refusal that
937    /// consumes the checkpoint, never a wait or a legacy fallback: more
938    /// progress or more RCBs (overshot), the endpoint's coordinates with
939    /// another time or last operation (incompatible), a front for a row that
940    /// already ended (skipped), and checkpoints left after the last row
941    /// (incomplete).
942    #[test]
943    fn a_priority_row_refuses_every_endpoint_it_cannot_honor() {
944        let expected = origin();
945        for (active, actual, refusal) in [
946            (
947                ActiveRow::Row(1),
948                Origin {
949                    progress: 3,
950                    ..expected.clone()
951                },
952                Refusal::Overshot,
953            ),
954            (
955                ActiveRow::Row(1),
956                Origin {
957                    rcbs: 11,
958                    site: Site::Prehook,
959                    ..expected.clone()
960                },
961                Refusal::Overshot,
962            ),
963            (
964                ActiveRow::Row(1),
965                Origin {
966                    time: LogicalTime::from_nanos(101),
967                    ..expected.clone()
968                },
969                Refusal::Incompatible,
970            ),
971            (
972                ActiveRow::Row(1),
973                Origin {
974                    last_op: Some(Op::Branch),
975                    ..expected.clone()
976                },
977                Refusal::Incompatible,
978            ),
979            (ActiveRow::Row(2), expected.clone(), Refusal::Skipped),
980            (ActiveRow::Exhausted, expected.clone(), Refusal::Incomplete),
981        ] {
982            let mut state = rows(&[(1, expected.clone()), (3, expected.clone())]);
983            assert_eq!(
984                state.decide_row(active, Some(actual.clone()), None, actual.rcbs),
985                RowDecision::Refuse {
986                    refusal,
987                    checkpoint: Checkpoint {
988                        history_index: 1,
989                        origin: expected.clone(),
990                    },
991                    actual: Some(actual.clone()),
992                },
993                "{active:?} {actual:?}"
994            );
995            assert!(
996                state.annotates(3),
997                "only the refused checkpoint is consumed"
998            );
999        }
1000        // A skipped front is refused even at a check-in without an identity.
1001        let mut state = rows(&[(1, expected.clone())]);
1002        assert!(matches!(
1003            state.decide_row(ActiveRow::Row(2), None, None, 0),
1004            RowDecision::Refuse {
1005                refusal: Refusal::Skipped,
1006                actual: None,
1007                ..
1008            }
1009        ));
1010        assert_eq!(
1011            Refusal::Skipped.name(),
1012            "HERMIT_CALLBACK_CHECKPOINT_SKIPPED"
1013        );
1014        assert_eq!(
1015            Refusal::Overshot.name(),
1016            "HERMIT_CALLBACK_CHECKPOINT_OVERSHOT"
1017        );
1018        assert_eq!(
1019            Refusal::Incompatible.name(),
1020            "HERMIT_CALLBACK_CHECKPOINT_INCOMPATIBLE"
1021        );
1022        assert_eq!(
1023            Refusal::Incomplete.name(),
1024            "HERMIT_CALLBACK_CHECKPOINT_INCOMPLETE"
1025        );
1026    }
1027
1028    /// A priority-history check-in without a callback identity (no token, or a
1029    /// token without a callback) is refused as overshot by name when the
1030    /// thread's own progress or committed RCBs are past the active row's
1031    /// endpoint, instead of waiting until the timer arm's assertion; at the
1032    /// endpoint it still waits, and the raw front is still judged first.
1033    #[test]
1034    fn a_check_in_without_identity_past_the_endpoint_is_refused_as_overshot() {
1035        let expected = origin();
1036        // A check-in without a callback identity is never the endpoint, but
1037        // the thread's own progress and committed RCBs still prove when it
1038        // went past one: that is refused as overshot by name, with only the
1039        // local custody consumed. Exactly at the endpoint's progress and RCBs
1040        // it waits; the boundary is the identified check-in's.
1041        let tid = DetTid::from_raw(3);
1042        let at_endpoint = |progress: u32| {
1043            let mut state = rows(&[(1, expected.clone()), (3, expected.clone())]);
1044            state.observe_event(&SchedEvent::branches(tid, progress));
1045            state
1046        };
1047        let mut state = at_endpoint(2);
1048        assert_eq!(
1049            state.decide_row(ActiveRow::Row(1), None, None, expected.rcbs),
1050            RowDecision::Pending,
1051            "at the endpoint's progress and RCBs"
1052        );
1053        assert!(state.annotates(1), "a pending check-in consumes nothing");
1054        for (progress, rcbs, why) in [
1055            (2, expected.rcbs + 1, "one RCB past the endpoint"),
1056            (3, expected.rcbs, "one progress unit past the endpoint"),
1057        ] {
1058            let mut state = at_endpoint(progress);
1059            assert_eq!(
1060                state.decide_row(ActiveRow::Row(1), None, None, rcbs),
1061                RowDecision::Refuse {
1062                    refusal: Refusal::Overshot,
1063                    checkpoint: Checkpoint {
1064                        history_index: 1,
1065                        origin: expected.clone(),
1066                    },
1067                    actual: None,
1068                },
1069                "{why}"
1070            );
1071            assert!(
1072                state.annotates(3),
1073                "{why}: only the refused checkpoint is consumed"
1074            );
1075        }
1076        // The raw front is still judged first: past the endpoint, a skipped
1077        // front is refused as skipped and an exhausted one as incomplete.
1078        for (active, refusal) in [
1079            (ActiveRow::Row(2), Refusal::Skipped),
1080            (ActiveRow::Exhausted, Refusal::Incomplete),
1081        ] {
1082            let mut state = at_endpoint(3);
1083            assert!(
1084                matches!(
1085                    state.decide_row(active, None, None, expected.rcbs + 1),
1086                    RowDecision::Refuse { refusal: r, actual: None, .. } if r == refusal
1087                ),
1088                "{active:?}"
1089            );
1090        }
1091    }
1092
1093    /// At the endpoint's progress, the thread's own callback proves it left a
1094    /// prehook endpoint for its posthook: overshot, not a wait. Without that
1095    /// evidence (another callback's token, or no cut at the endpoint) the
1096    /// later site waits.
1097    #[test]
1098    fn a_priority_row_refuses_a_prehook_endpoint_its_own_callback_left() {
1099        let tid = DetTid::from_raw(3);
1100        let expected = Origin {
1101            site: Site::Prehook,
1102            progress: 2,
1103            last_op: Some(Op::Branch),
1104            ..origin()
1105        };
1106        let later = Origin {
1107            site: Site::Posthook,
1108            ..expected.clone()
1109        };
1110        let mut thread = thread();
1111        let mut token = begin(&mut thread, Some(expected.callback))
1112            .unwrap()
1113            .unwrap();
1114        let unrelated = begin(&mut thread, Some(Callback::Cpuid)).unwrap().unwrap();
1115        let uncut = token.clone();
1116        thread
1117            .callback_history
1118            .observe_event(&SchedEvent::branches(tid, 2));
1119        token.note_cut(thread.callback_history.progress(), Site::Prehook);
1120        for witness in [&unrelated, &uncut] {
1121            let mut state = rows(&[(0, expected.clone())]);
1122            assert_eq!(
1123                state.decide_row(
1124                    ActiveRow::Row(0),
1125                    Some(later.clone()),
1126                    Some(witness),
1127                    later.rcbs
1128                ),
1129                RowDecision::Pending
1130            );
1131        }
1132        let mut state = rows(&[(0, expected.clone())]);
1133        assert!(matches!(
1134            state.decide_row(
1135                ActiveRow::Row(0),
1136                Some(later.clone()),
1137                Some(&token),
1138                later.rcbs
1139            ),
1140            RowDecision::Refuse {
1141                refusal: Refusal::Overshot,
1142                ..
1143            }
1144        ));
1145    }
1146
1147    /// A thread's active row follows its own history cursor: before the first
1148    /// row is installed, each installed row in turn, then the final slice. Its
1149    /// pending checkpoint is the front checkpoint only while that names the
1150    /// active row.
1151    #[test]
1152    fn the_active_priority_row_follows_the_history_cursor() {
1153        use crate::preemptions::ThreadHistory;
1154        let mut thread = thread();
1155        assert_eq!(thread.active_priority_row(), None, "no history replayed");
1156        thread.preemption_points = Some(
1157            ThreadHistory::new()
1158                .with_prio_changes(vec![
1159                    (LogicalTime::from_nanos(10), 7),
1160                    (LogicalTime::from_nanos(20), 8),
1161                ])
1162                .with_preemption_rcbs(vec![10, 20])
1163                .into_iter(),
1164        );
1165        thread.callback_history = rows(&[(1, origin())]);
1166        assert_eq!(thread.active_priority_row(), Some(ActiveRow::BeforeFirst));
1167        assert!(thread.pending_checkpoint().is_none());
1168        let history = thread.preemption_points.as_mut().unwrap();
1169        history.next_with_rcbs().unwrap();
1170        assert_eq!(thread.active_priority_row(), Some(ActiveRow::Row(0)));
1171        assert!(
1172            thread.pending_checkpoint().is_none(),
1173            "row 0 is unannotated"
1174        );
1175        thread
1176            .preemption_points
1177            .as_mut()
1178            .unwrap()
1179            .next_with_rcbs()
1180            .unwrap();
1181        assert_eq!(thread.active_priority_row(), Some(ActiveRow::Row(1)));
1182        assert_eq!(thread.pending_checkpoint().unwrap().origin, origin());
1183        assert!(
1184            thread
1185                .preemption_points
1186                .as_mut()
1187                .unwrap()
1188                .next_with_rcbs()
1189                .is_none()
1190        );
1191        thread.stats.last_recorded_slice = Some(thread.stats.timeslice_count);
1192        assert_eq!(thread.active_priority_row(), Some(ActiveRow::Exhausted));
1193        assert!(thread.pending_checkpoint().is_none());
1194    }
1195
1196    /// The observed origin of a check-in carries its callback and site with
1197    /// the thread's progress, time and RCBs now; none without an identity.
1198    #[test]
1199    fn an_observed_priority_checkin_reads_actual_coordinates() {
1200        let tid = DetTid::from_raw(3);
1201        let mut thread = thread();
1202        thread.committed_clock_value = 41;
1203        thread
1204            .callback_history
1205            .observe_event(&SchedEvent::branches(tid, 5));
1206        let token = begin(&mut thread, Some(Callback::Rdtsc)).unwrap().unwrap();
1207        let observed = thread
1208            .callback_history
1209            .observed(&thread, Some(&token), Site::Prehook)
1210            .unwrap();
1211        assert_eq!(observed.callback, Callback::Rdtsc);
1212        assert_eq!(observed.site, Site::Prehook);
1213        assert_eq!(observed.progress, 5);
1214        assert_eq!(observed.last_op, Some(Op::Branch));
1215        assert_eq!(observed.rcbs, 41);
1216        assert_eq!(observed.time, thread.thread_logical_time.as_nanos());
1217        assert_eq!(
1218            thread
1219                .callback_history
1220                .observed(&thread, None, Site::Prehook),
1221            None
1222        );
1223        let anonymous = begin(&mut thread, None).unwrap().unwrap();
1224        assert_eq!(
1225            thread
1226                .callback_history
1227                .observed(&thread, Some(&anonymous), Site::Prehook),
1228            None
1229        );
1230    }
1231}
detcore/preemptions/callback_history.rs

detcore/preemptions/
callback_history.rs
