Skip to main content

shuttle_engine/runtime/
execution.rs

1use crate::config::DEFAULT_MAX_STEPS;
2use crate::runtime::failure::{init_panic_hook, persist_failure};
3use crate::runtime::storage::{StorageKey, StorageMap};
4use crate::runtime::task::clock::VectorClock;
5use crate::runtime::task::labels::Labels;
6use crate::runtime::task::{ChildLabelFn, ParkedDefault, Task, TaskId, TaskName, TaskSignature, DEFAULT_INLINE_TASKS};
7use crate::runtime::thread;
8use crate::runtime::thread::continuation::PooledContinuation;
9use crate::scheduler::{Schedule, Scheduler};
10use crate::sync_types::{ResourceSignature, ResourceType};
11use crate::thread_support::thread_fn;
12use crate::{backtrace_enabled, Config, MaxSteps, UNGRACEFUL_SHUTDOWN_CONFIG};
13use scoped_tls::scoped_thread_local;
14use smallvec::SmallVec;
15use std::any::Any;
16use std::cell::{Cell, RefCell};
17use std::collections::{HashMap, VecDeque};
18use std::fmt::Debug;
19use std::future::Future;
20use std::panic::{self, Location};
21use std::rc::Rc;
22use std::sync::Arc;
23use tracing::subscriber::NoSubscriber;
24use tracing::{trace, Span};
25
26#[allow(deprecated)]
27use super::task::Tag;
28
29// We use this scoped TLS to smuggle the ExecutionState, which is not 'static, across tasks that
30// need access to it (to spawn new tasks, interrogate task status, etc).
31scoped_thread_local! {
32    static EXECUTION_STATE: RefCell<ExecutionState>
33}
34
35thread_local! {
36    // Whether the panic hook stays silent, because execution teardown ignores the panics it catches
37    // now (see `ExecutionState::tear_down`). Outside `EXECUTION_STATE`, so that the hook can read it
38    // while the state is borrowed.
39    static TEARDOWN_IGNORES_PANICS: Cell<bool> = const { Cell::new(false) };
40    // How many panics there have been in the current step of execution teardown, for the panic
41    // hook (see `ExecutionState::teardown_ignores_panic`).
42    static TEARDOWN_STEP_PANICS: Cell<usize> = const { Cell::new(0) };
43}
44
45thread_local! {
46    // Whether this thread is emitting the `ExecutionState::try_with` event. A `tracing` subscriber
47    // may access the state while it handles an event (to call `current::clock()`, say), and that
48    // access must not emit the event again: `tracing` only refuses re-entrant calls for scoped
49    // subscribers, and with a global one this would recurse until the stack overflows.
50    static TRACING_TRY_WITH: Cell<bool> = const { Cell::new(false) };
51}
52
53// The reason this is separated out from `ExecutionState` is to ensure that we're always able to persist the schedule.
54// If we don't do this, then we may panic while borrowing `ExecutionState`, and then not be able to emit the schedule.
55// If we then panic again while trying to handle the panic, such that the panic becomes an abort, we will never log
56// the schedule.
57//
58// It is expected that if the `ExecutionState` exists, then this will exist, and any usage of this happens through the
59// `ExecutionState`, or at a point where it is known that the `ExecutionState` must exist (eg. when serializing on a panic).
60thread_local! {
61    static CURRENT_SCHEDULE: CurrentSchedule = CurrentSchedule::default();
62}
63
64#[derive(Debug, Default)]
65pub struct CurrentSchedule {
66    current_schedule: RefCell<Schedule>,
67}
68
69impl CurrentSchedule {
70    fn init(schedule: Schedule) {
71        CURRENT_SCHEDULE.with(|cs| *cs.current_schedule.borrow_mut() = schedule)
72    }
73
74    /// Add the given task ID as the next step of the schedule.
75    fn push_task(tid: TaskId) {
76        CURRENT_SCHEDULE.with(|cs| cs.current_schedule.borrow_mut().push_task(tid))
77    }
78
79    /// Add a choice of a random u64 value as the next step of the schedule
80    fn push_random() {
81        CURRENT_SCHEDULE.with(|cs| cs.current_schedule.borrow_mut().push_random())
82    }
83
84    /// Return the number of steps in the schedule
85    pub fn len() -> usize {
86        CURRENT_SCHEDULE.with(|cs| (*cs.current_schedule.borrow()).len())
87    }
88
89    /// Returns a clone of the inner schedule
90    pub fn get_schedule() -> Schedule {
91        CURRENT_SCHEDULE.with(|cs| (*cs.current_schedule.borrow()).clone())
92    }
93}
94
95thread_local! {
96    #[allow(clippy::complexity)]
97    #[allow(deprecated)]
98    pub static TASK_ID_TO_TAGS: RefCell<HashMap<TaskId, Arc<dyn Tag>>> = RefCell::new(HashMap::new());
99}
100
101thread_local! {
102    pub static LABELS: RefCell<HashMap<TaskId, Labels>> = RefCell::new(HashMap::new());
103}
104
105/// An `Execution` encapsulates a single run of a function under test against a chosen scheduler.
106/// Its only useful method is `Execution::run`, which executes the function to completion.
107///
108/// The key thing that an `Execution` manages is the `ExecutionState`, which contains all the
109/// mutable state a test's tasks might need access to during execution (to block/unblock tasks,
110/// spawn new tasks, etc). The `Execution` makes this state available through the `EXECUTION_STATE`
111/// static variable, but clients get access to it by calling `ExecutionState::with`.
112pub struct Execution {
113    scheduler: Rc<RefCell<dyn Scheduler>>,
114    initial_schedule: Schedule,
115}
116
117impl std::fmt::Debug for Execution {
118    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
119        f.debug_struct("Execution").finish_non_exhaustive()
120    }
121}
122
123impl Execution {
124    /// Construct a new execution that will use the given scheduler. The execution should then be
125    /// invoked via its `run` method, which takes as input the closure for task 0.
126    pub fn new(scheduler: Rc<RefCell<dyn Scheduler>>, initial_schedule: Schedule) -> Self {
127        Self {
128            scheduler,
129            initial_schedule,
130        }
131    }
132}
133
134#[derive(Debug)]
135enum StepError {
136    // Contains the panic payload of the task that failed.
137    TaskFailure(Box<dyn Any + Send>),
138    // The scheduler didn't make a decision. Indicates a scheduler error.
139    SchedulingError,
140    // Scheduling deetected a deadlock.
141    Deadlock,
142    // We exceeded the step bound of `MaxSteps::FailAfter`, which this is.
143    StepBoundExceeded(usize),
144    // Task panic and `config.immediately_return_on_panic` is set to `true`.
145    TaskPanicEarlyReturn,
146}
147
148/// How a failed execution fails the test, once it has been torn down.
149enum Failure {
150    /// Resume unwinding with the panic payload.
151    Unwind(Box<dyn Any + Send>),
152    /// Panic with the message.
153    Panic(String),
154    /// Panic with the message, as a `&'static str` payload.
155    StaticPanic(&'static str),
156}
157
158/// How tearing down an execution that had not failed fails the test (see `ExecutionState::tear_down`).
159enum TeardownFailure {
160    /// A destructor panicked: resume unwinding with its payload.
161    Unwind(Box<dyn Any + Send>),
162    /// A task that was unwinding a panic when the execution stopped has finished unwinding it:
163    /// resume unwinding with its payload.
164    LatePanic(Box<dyn Any + Send>),
165    /// Panic with the message: destructors blocked, and nothing could wake them, say.
166    Panic(String),
167}
168
169impl Execution {
170    /// Run a function to be tested, taking control of scheduling it and any tasks it might spawn.
171    /// This function runs until `f` and all tasks spawned by `f` have terminated, or until the
172    /// scheduler returns `None`, indicating the execution should not be explored any further.
173    pub fn run<F>(mut self, config: &Config, f: F, caller: &'static Location<'static>)
174    where
175        F: FnOnce() + Send + 'static,
176    {
177        let state = RefCell::new(ExecutionState::new(config.clone(), Rc::clone(&self.scheduler)));
178
179        init_panic_hook(config.clone());
180        CurrentSchedule::init(self.initial_schedule.clone());
181        UNGRACEFUL_SHUTDOWN_CONFIG.set(config.ungraceful_shutdown_config);
182
183        EXECUTION_STATE.set(&state, move || {
184            // Spawn `f` as the first task
185            ExecutionState::spawn_main_thread(
186                Box::new(move || thread_fn(f, true, Default::default())),
187                config.stack_size,
188                caller,
189            );
190
191            // Run the test to completion. A panic out of the executor (from a scheduler, say) fails
192            // the execution like a task's panic does, so that the execution is still torn down while
193            // Shuttle is there to call into.
194            let immediately_return_on_panic = UNGRACEFUL_SHUTDOWN_CONFIG.get().immediately_return_on_panic;
195            let failure = match panic::catch_unwind(panic::AssertUnwindSafe(|| {
196                self.run_to_completion(immediately_return_on_panic)
197            })) {
198                Ok(Ok(())) => None,
199                Ok(Err(e)) => Some(Self::failure(e, config)),
200                // The panic hook has reported it.
201                Err(payload) => Some(Failure::Unwind(payload)),
202            };
203
204            // Tear down the execution before it goes out of `EXECUTION_STATE` scope. A failed
205            // execution is torn down before its failure is raised, so that nothing that the failure
206            // leaves to unwind can call into Shuttle once `EXECUTION_STATE` is gone.
207            let kind = if failure.is_some() {
208                TeardownKind::Failed
209            } else if ExecutionState::execution_stopped() {
210                TeardownKind::Stopped
211            } else {
212                TeardownKind::Finished
213            };
214            let teardown_failure = ExecutionState::tear_down(kind);
215            match failure {
216                Some(Failure::Unwind(payload)) => panic::resume_unwind(payload),
217                Some(Failure::Panic(message)) => panic::panic_any(message),
218                Some(Failure::StaticPanic(message)) => panic::panic_any(message),
219                None => match teardown_failure {
220                    Some(TeardownFailure::Unwind(payload)) => {
221                        persist_failure(config);
222                        eprintln!("test panicked while dropping the tasks that were unfinished at the end of the execution");
223                        panic::resume_unwind(payload);
224                    }
225                    Some(TeardownFailure::LatePanic(payload)) => {
226                        persist_failure(config);
227                        eprintln!("test panicked in a task that was still unwinding the panic when the execution stopped");
228                        panic::resume_unwind(payload);
229                    }
230                    Some(TeardownFailure::Panic(message)) => panic::panic_any(message),
231                    None => {}
232                },
233            }
234        });
235    }
236
237    /// Report the failure of the execution, and return how to fail the test with it once the
238    /// execution has been torn down.
239    fn failure(e: StepError, config: &Config) -> Failure {
240        persist_failure(config);
241        // Teardown changes the current task, so remember which task failed for the panic hook.
242        ExecutionState::with(|state| state.failed_task = state.try_current().map(Task::display_name));
243
244        match e {
245            StepError::TaskFailure(payload) => {
246                eprintln!("test panicked in task '{}'", ExecutionState::failing_task());
247
248                Failure::Unwind(payload)
249            }
250            StepError::Deadlock => {
251                let blocked_tasks = ExecutionState::with(|state| {
252                    state
253                        .tasks
254                        .iter()
255                        .filter(|t| !t.finished())
256                        .map(|t| t.format_for_deadlock())
257                        .collect::<Vec<_>>()
258                });
259
260                // Collecting backtraces is expensive, so we only want to do it if the user opts in to collecting them.
261                if !backtrace_enabled() {
262                    eprintln!("Test deadlocked, and {} is not set. If either of those are set then the backtrace of each task will be collected and printed as part of the panic message.", crate::CAPTURE_BACKTRACE)
263                }
264
265                Failure::Panic(format!("deadlock! blocked tasks: [{}]", blocked_tasks.join(", ")))
266            }
267            StepError::SchedulingError => {
268                Failure::StaticPanic("no task was scheduled\nThis indicates an issue with the scheduler.")
269            }
270            StepError::StepBoundExceeded(max_steps) => Failure::Panic(format!(
271                "exceeded max_steps bound {max_steps}. this might be caused by an unfair schedule (e.g., a spin loop)?"
272            )),
273            StepError::TaskPanicEarlyReturn => Failure::Unwind(Box::new("Task panicked, and early return is enabled.")),
274        }
275    }
276
277    fn enter_task_span() {
278        // Enter the Task's span
279        // (Note that if any issues arise with spans and tracing, then
280        // 1) calling `exit` until `None` before entering the `Task`s `Span`,
281        // 2) storing the entirety of the `span_stack` when creating the `Task`, and
282        // 3) storing `top_level_span` as a stack
283        // should be tried.)
284        let parked = ExecutionState::with(|state| {
285            // Go through `Span::with_subscriber` rather than calling `Span::current()` inside
286            // `tracing::dispatcher::get_default`; see `exit_task_span` for why.
287            tracing::Span::current().with_subscriber(|(id, subscriber)| subscriber.exit(id));
288
289            // The `span_stack` stores `Span`s such that the top of the stack is the outermost `Span`,
290            // meaning that parents (left-most when printed) are entered first.
291            while let Some(span) = state.current_mut().span_stack.pop() {
292                span.with_subscriber(|(id, subscriber)| subscriber.enter(id));
293            }
294
295            if state.config.record_steps_in_span {
296                state.current().step_span.record("i", CurrentSchedule::len());
297            }
298
299            state.current_mut().parked_default.take()
300        });
301
302        // Reinstate the default dispatcher the task had when it last switched out. Outside
303        // `ExecutionState::with`, as that drops the dispatcher it displaces.
304        if let Some(parked) = parked {
305            parked.reinstate();
306        }
307    }
308
309    fn exit_task_span(yielded: bool) {
310        // Leave the Task's span and store the exited `Span` stack in order to restore it the next time the Task is run
311        ExecutionState::with(|state| {
312            // Before the task's spans are exited, so they are looked up in the dispatcher they were entered in.
313            if yielded {
314                Execution::park_task_default(state);
315            }
316
317            debug_assert!(state.current().span_stack.is_empty());
318            // Note that `Span::current()` must not be called from inside a
319            // `tracing::dispatcher::get_default` callback: `get_default` marks the thread's
320            // dispatcher state as in use for the duration of the callback, and while any thread in
321            // the process holds a scoped default subscriber, a nested `Span::current()` then
322            // returns `Span::none()`. This loop would then exit nothing while we still enter
323            // `top_level_span` below, leaking one entry per scheduling step until a later
324            // `Span::current()` resolved to a closed span and panicked. So each exit goes through
325            // `Span::with_subscriber`, which hands us the span's own dispatcher directly.
326            // `Span::current()` returns a disabled span once no span is entered, and a disabled span
327            // is exactly one `with_subscriber` does nothing for, so every iteration exits a span.
328            while let Some(current) = Some(tracing::Span::current()).filter(|span| !span.is_disabled()) {
329                current.with_subscriber(|(id, subscriber)| subscriber.exit(id));
330                state.current_mut().span_stack.push(current);
331            }
332
333            state
334                .top_level_span
335                .with_subscriber(|(id, subscriber)| subscriber.enter(id));
336        });
337    }
338
339    /// Saves the running task's default `tracing` dispatcher and reinstates the execution's.
340    ///
341    /// `tracing`'s default dispatcher is per OS thread, and every task runs on this one. Without
342    /// this, a task that yields inside `tracing::subscriber::with_default` would leave its dispatcher
343    /// installed for the scheduler and every other task, and `exit_task_span` would look for the
344    /// task's spans in that dispatcher instead of the one they were entered in.
345    fn park_task_default(state: &mut ExecutionState) {
346        // Installing a guard when there is nothing to park would put `tracing` on its slower scoped
347        // path for the whole process.
348        if !task_may_have_own_default(state) {
349            return;
350        }
351        // The guard remembers the task's current default, and `enter_task_span` drops it to
352        // reinstate exactly that.
353        let guard = tracing::dispatcher::set_default(&state.top_level_dispatch);
354        state.current_mut().parked_default = Some(ParkedDefault::new(guard));
355    }
356
357    /// Run the execution to completion.
358    #[inline]
359    fn run_to_completion(&mut self, immediately_return_on_panic: bool) -> Result<(), StepError> {
360        loop {
361            let next_step: Option<Rc<RefCell<PooledContinuation>>> = ExecutionState::with(|state| {
362                state.schedule()?;
363                state.advance_to_next_task();
364
365                match state.current_task {
366                    ScheduledTask::Some(tid) => {
367                        let task = state.get(tid);
368                        Ok(Some(
369                            task.continuation
370                                .clone()
371                                .expect("only execution teardown takes a task's continuation"),
372                        ))
373                    }
374                    ScheduledTask::Finished => {
375                        // The scheduler decided we're finished, so there are either no runnable tasks,
376                        // or all runnable tasks are detached and there are no unfinished attached
377                        // tasks. Therefore, it's a deadlock if there are unfinished attached tasks.
378                        if state.tasks.iter().any(|t| !t.finished() && !t.detached) {
379                            Err(StepError::Deadlock)
380                        } else {
381                            Ok(None)
382                        }
383                    }
384                    ScheduledTask::Stopped => Ok(None),
385                    ScheduledTask::None => Err(StepError::SchedulingError),
386                }
387            })?;
388
389            // Run a single step of the chosen task.
390            let ret = match next_step {
391                Some(continuation) => {
392                    Execution::enter_task_span();
393
394                    let result = panic::catch_unwind(panic::AssertUnwindSafe(|| continuation.borrow_mut().resume()));
395
396                    Execution::exit_task_span(matches!(result, Ok(false)));
397
398                    result
399                }
400                None => return Ok(()),
401            };
402
403            match ret {
404                // Task finished
405                Ok(true) => {
406                    crate::annotations::record_task_terminated();
407                    ExecutionState::with(|state| {
408                        state.finish_current_task();
409                        // The task may have caught the panic that was unwinding last, and finished
410                        // without another scheduling point. Here, between steps,
411                        // `std::thread::panicking()` says exactly whether a task that switched out is
412                        // unwinding a panic still (see `record_unwinding_task`).
413                        if state.switched_out_unwinding && !std::thread::panicking() {
414                            state.forget_unwinding_task();
415                        }
416                    });
417                }
418                // Task yielded
419                Ok(false) => {
420                    // We may have `switch`ed out of the task before we finished unwinding the stack (ie. a `drop` handler calls `switch`).
421                    // If `immediately_return_on_panic` is set, we will then return. If we don't do this, then we run the risk of panicking
422                    // again in some other task, which would result in the test aborting.
423                    if immediately_return_on_panic && std::thread::panicking() {
424                        ExecutionState::with(|state| state.current_task = ScheduledTask::Stopped);
425                        return Err(StepError::TaskPanicEarlyReturn);
426                    }
427                }
428                // Task failed
429                Err(e) => return Err(StepError::TaskFailure(e)),
430            }
431        }
432    }
433}
434
435/// `ExecutionState` contains the portion of a single execution's state that needs to be reachable
436/// from within a task's execution. It tracks which tasks exist and their states, as well as which
437/// tasks are pending spawn.
438pub struct ExecutionState {
439    pub config: Config,
440    // invariant: tasks are never removed from this list, until execution teardown drops them all
441    tasks: SmallVec<[Task; DEFAULT_INLINE_TASKS]>,
442    // invariant: if this transitions to Stopped or Finished, it can never change again, except that
443    // execution teardown makes a failed execution Stopped, and makes each task that it tears down the
444    // current task while it does (see `tear_down`)
445    current_task: ScheduledTask,
446    // the task the scheduler has chosen to run next
447    next_task: ScheduledTask,
448    // whether the current task has asked to yield
449    has_yielded: bool,
450    // the number of scheduling decisions made so far
451    context_switches: usize,
452    // the schedule length last time `reset_stop_bound()` was called
453    pub steps_reset_at: usize,
454
455    // static values for the current execution
456    storage: StorageMap,
457
458    scheduler: Rc<RefCell<dyn Scheduler>>,
459
460    // the state of execution teardown, while the execution is being torn down (see `tear_down`)
461    teardown: Option<Teardown>,
462
463    // the name of the task that failed the execution, if a task did, for the panic hook to report
464    // once the execution has been torn down (see `failing_task`)
465    failed_task: Option<String>,
466
467    // the task that switched out while unwinding a panic, if it may be unwinding it still, with
468    // whether it is detached otherwise: it is attached until it has finished unwinding (see
469    // `record_unwinding_task`)
470    unwinding_task: Option<(TaskId, bool)>,
471
472    // whether a task has switched out while unwinding a panic since `std::thread::panicking()` was
473    // last false: if not, no task but the current one can be unwinding a panic (see
474    // `record_unwinding_task`)
475    switched_out_unwinding: bool,
476
477    #[cfg(debug_assertions)]
478    has_cleaned_up: bool,
479
480    // The `Span` which the `ExecutionState` was created under. Will be the parent of all `Task` `Span`s
481    pub top_level_span: Span,
482
483    // The default `tracing` dispatcher the `ExecutionState` was created under. Tasks start out with it,
484    // and it is reinstated whenever a task with a different default switches out.
485    top_level_dispatch: tracing::Dispatch,
486
487    // If `top_level_dispatch` is the global default, where `get_default` hands it out from (see
488    // `task_may_have_own_default`).
489    top_level_dispatch_global_ptr: Option<*const tracing::Dispatch>,
490
491    // Persistent Vec used as a bump allocator for references to runnable tasks to avoid slow allocation
492    // on each scheduling decision. Should not be used outside of the `schedule` function
493    runnable_tasks: Vec<*const Task>,
494
495    // Ids of all tasks that have not yet finished, kept sorted in ascending order.
496    //
497    // `tasks` never shrinks, so it accumulates every task ever created by the execution. Scanning it
498    // on every scheduling decision therefore costs O(tasks ever created), even though only the
499    // unfinished ones can ever be scheduled. This set lets `schedule` iterate just the live tasks.
500    //
501    // invariant: contains exactly the ids of the tasks in `tasks` that are not `Finished`, in
502    // ascending order. Maintained by pushing on task creation (ids are handed out sequentially, so
503    // pushing keeps this sorted) and removing in `finish_current_task`. This is sound because
504    // `Finished` is a terminal state: it is only ever set by `Task::finish`, and `block`, `sleep`,
505    // and `unblock` all assert that they are never applied to a finished task. (Execution teardown
506    // schedules nothing, and doesn't maintain this: see `tear_down`. It also lets a finished task
507    // stand in for a while: see `Task::stand_in`.)
508    live_tasks: Vec<TaskId>,
509}
510
511impl std::fmt::Debug for ExecutionState {
512    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
513        f.debug_struct("ExecutionState").finish_non_exhaustive()
514    }
515}
516
517#[derive(Debug, PartialEq, Eq, Clone, Copy)]
518enum ScheduledTask {
519    None,         // no task has ever been scheduled
520    Some(TaskId), // this task is running
521    Stopped,      // the scheduler asked us to stop running
522    Finished,     // all tasks have finished running
523}
524
525impl ScheduledTask {
526    fn id(&self) -> Option<TaskId> {
527        match self {
528            ScheduledTask::Some(tid) => Some(*tid),
529            _ => None,
530        }
531    }
532
533    fn take(&mut self) -> Self {
534        std::mem::replace(self, ScheduledTask::None)
535    }
536}
537
538/// How an execution ended, which decides how it is torn down (see `ExecutionState::tear_down`).
539#[derive(Debug, PartialEq, Eq, Clone, Copy)]
540enum TeardownKind {
541    /// Every attached task finished.
542    Finished,
543    /// The scheduler stopped the execution before that.
544    Stopped,
545    /// The execution failed. Its failure is raised once it has been torn down.
546    Failed,
547}
548
549impl TeardownKind {
550    /// Whether the execution is abandoned, rather than torn down as if its tasks were cancelled.
551    fn abandons(self) -> bool {
552        self != Self::Finished
553    }
554
555    /// What `ExecutionState::current_task` is between the steps of teardown.
556    fn final_state(self) -> ScheduledTask {
557        match self {
558            Self::Finished => ScheduledTask::Finished,
559            Self::Stopped | Self::Failed => ScheduledTask::Stopped,
560        }
561    }
562}
563
564/// What execution teardown is doing (see `ExecutionState::tear_down`). In each step, the task named
565/// is the current task.
566#[derive(Debug, PartialEq, Eq, Clone, Copy)]
567enum TeardownStep {
568    /// Nothing, between two steps.
569    Idle,
570    /// The task runs on its own stack, cancelled: it drops its future, or the function it never ran.
571    Cancel(TaskId),
572    /// The task's stack is unwound.
573    Unwind(TaskId),
574    /// The task runs on its own stack to finish unwinding a panic (see
575    /// `ExecutionState::finish_unwinding`).
576    FinishUnwinding(TaskId),
577    /// What the task left behind, its task-local values say, is dropped on the executor's stack.
578    Leftovers(TaskId),
579    /// A static is dropped on the executor's stack, with the main thread standing in.
580    Static,
581}
582
583/// The state of execution teardown (see `ExecutionState::tear_down`).
584#[derive(Debug)]
585struct Teardown {
586    kind: TeardownKind,
587    step: TeardownStep,
588    /// The scheduling points that destructors have reached, which the step bound limits.
589    steps: usize,
590    /// The step bound, once a destructor has exceeded it. Then the rest of the execution is
591    /// abandoned (see `next_teardown_job`).
592    exceeded_step_bound: Option<usize>,
593    /// Whether the current task runs on its own stack now, so that it can switch out (see
594    /// `maybe_yield_in_teardown`).
595    on_own_stack: bool,
596    /// The state of the generator that destructors draw random numbers from, instead of the
597    /// scheduler, so that teardown doesn't extend the schedule but is the same when it is replayed.
598    rng: u64,
599}
600
601impl Teardown {
602    fn new(kind: TeardownKind, schedule_len: usize) -> Self {
603        Self {
604            kind,
605            step: TeardownStep::Idle,
606            steps: 0,
607            exceeded_step_bound: None,
608            on_own_stack: false,
609            rng: schedule_len as u64,
610        }
611    }
612
613    /// The next number from the generator (SplitMix64).
614    fn next_u64(&mut self) -> u64 {
615        self.rng = self.rng.wrapping_add(0x9E37_79B9_7F4A_7C15);
616        let mut z = self.rng;
617        z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
618        z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
619        z ^ (z >> 31)
620    }
621}
622
623/// Where execution teardown has got to (see `ExecutionState::next_teardown_job`).
624#[derive(Debug, Default)]
625struct TeardownPlan {
626    /// The task that switched out while it unwound a panic, and is unwinding it still (see
627    /// `ExecutionState::finish_unwinding`).
628    unwinding_task: Option<TaskId>,
629    /// The index in `ExecutionState::tasks` of the next task that teardown hasn't looked at.
630    next_task: usize,
631    /// The tasks that teardown cancelled and that switched out, in the order they did, each with
632    /// whether it blocked (rather than yielded).
633    suspended: Vec<(TaskId, bool)>,
634    /// The unfinished tasks in the middle of user code, whose stacks teardown unwinds or leaks.
635    stacks: VecDeque<TaskId>,
636    /// How many times in a row a task that yielded has been resumed without any task finishing or
637    /// blocking since, while stacks are waiting to be unwound.
638    idle_yields: usize,
639}
640
641/// How many turns each task that yields gets before a stack is unwound, if no task finishes or
642/// blocks meanwhile (see `ExecutionState::next_teardown_job`), unless that would take more than a
643/// quarter of the steps that the step bound leaves. The tasks that yield may wait for a task whose
644/// stack teardown is going to unwind, or that task's destructors may wait for a lock that they
645/// hold; unwinding a stack can't wait.
646const TEARDOWN_YIELD_ROUNDS: usize = 1000;
647
648/// How execution teardown resumes a task on the task's own stack (see
649/// `ExecutionState::run_on_own_stack`).
650#[derive(Debug, PartialEq, Eq, Clone, Copy)]
651enum Resumption {
652    /// Cancel the task (see `Continuation::cancel`).
653    Cancel,
654    /// Resume a cancelled task that switched out.
655    Resume,
656}
657
658/// What execution teardown does next (see `ExecutionState::next_teardown_job`).
659enum TeardownJob {
660    /// Let the task finish unwinding a panic.
661    FinishUnwinding(TaskId),
662    /// Resume the task on its own stack.
663    OnOwnStack(TaskId, Resumption),
664    /// Unwind the task's stack.
665    Unwind(TaskId),
666    /// Leak the task's stack.
667    Leak(TaskId),
668    /// Leak the function of a task that never ran.
669    LeakFunction(TaskId),
670    /// Drop task-local values that a finished task left.
671    TaskLocals(TaskId),
672    /// Drop a static.
673    Static(Box<dyn Any>),
674    /// Every cancelled task that hasn't finished blocked, and nothing can wake them, or teardown is
675    /// abandoning them.
676    Stuck,
677    /// A destructor exceeded the step bound, which this is: abandon the rest of the execution.
678    Abandon(usize),
679    Done,
680}
681
682/// Why execution teardown stops a destructor (see `ExecutionState::maybe_yield_in_teardown`).
683enum TeardownStall {
684    /// It blocked where it cannot wait.
685    Blocked,
686    /// It exceeded the step bound, which this is.
687    ExceededStepBound(usize),
688}
689
690/// How execution teardown is going to fail the test, if it does (see `ExecutionState::tear_down`).
691struct TeardownReport {
692    failure: Option<TeardownFailure>,
693    ignored_panics: usize,
694}
695
696impl TeardownReport {
697    fn new() -> Self {
698        Self {
699            failure: None,
700            ignored_panics: 0,
701        }
702    }
703
704    /// Run `f`, which drops something, and deal with a panic from it.
705    fn catch(&mut self, f: impl FnOnce()) {
706        TEARDOWN_STEP_PANICS.set(0);
707        if let Err(payload) = panic::catch_unwind(panic::AssertUnwindSafe(f)) {
708            self.panicked(payload);
709        }
710    }
711
712    /// A destructor panicked. The first panic while tearing down a finished execution fails the
713    /// test, and teardown ignores any other.
714    fn panicked(&mut self, payload: Box<dyn Any + Send>) {
715        let abandoned = ExecutionState::with(|state| state.teardown().kind.abandons());
716        if !abandoned && self.failure.is_none() {
717            self.failure = Some(TeardownFailure::Unwind(payload));
718            // The panic hook has reported this one, but stays silent for those ignored from now on.
719            TEARDOWN_IGNORES_PANICS.set(true);
720        } else {
721            self.ignored_panics += 1;
722            // Dropping the payload can panic too.
723            if let Err(payload) = panic::catch_unwind(panic::AssertUnwindSafe(move || drop(payload))) {
724                std::mem::forget(payload);
725            }
726        }
727    }
728
729    /// A task that was unwinding a panic when the execution stopped has finished unwinding it. That
730    /// panic fails the test, unless something else already does.
731    fn late_panic(&mut self, payload: Box<dyn Any + Send>) {
732        if self.failure.is_none() {
733            self.failure = Some(TeardownFailure::LatePanic(payload));
734        } else {
735            self.panicked(payload);
736        }
737    }
738
739    /// Fail the test with `message`, unless something else already does.
740    fn fail(&mut self, message: String) {
741        if self.failure.is_none() {
742            self.failure = Some(TeardownFailure::Panic(message));
743        }
744    }
745
746    /// The destructors that teardown of a finished execution has left all blocked, and nothing can
747    /// wake them.
748    fn deadlocked(&mut self, blocked_tasks: Vec<String>) {
749        self.fail(format!(
750            "deadlock while dropping the tasks that were unfinished at the end of the execution! blocked tasks: [{}]",
751            blocked_tasks.join(", ")
752        ));
753    }
754
755    fn finish(self) -> Option<TeardownFailure> {
756        if self.failure.is_some() && self.ignored_panics > 0 {
757            eprintln!(
758                "{} more panics while dropping the tasks that were unfinished at the end of the execution were ignored",
759                self.ignored_panics
760            );
761        }
762        self.failure
763    }
764}
765
766/// Error type for when an `ExecutionState::with` fails
767#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
768pub enum ExecutionStateBorrowError {
769    /// `ExecutionState` is currently not set
770    NotSet,
771    /// We are trying to borrow `ExecutionState` while it is already borrowed
772    AlreadyBorrowed,
773}
774
775impl ExecutionState {
776    fn new(config: Config, scheduler: Rc<RefCell<dyn Scheduler>>) -> Self {
777        let (top_level_dispatch, top_level_dispatch_global_ptr) = tracing::dispatcher::get_default(|dispatch| {
778            // While no thread has a scoped default, `get_default` hands out the global default, and
779            // nested calls get it too rather than `Dispatch::none()`.
780            let is_global = tracing::dispatcher::get_default(|nested| !nested.is::<NoSubscriber>());
781            (dispatch.clone(), is_global.then_some(dispatch as *const _))
782        });
783        Self {
784            config,
785            tasks: SmallVec::new(),
786            current_task: ScheduledTask::None,
787            next_task: ScheduledTask::None,
788            has_yielded: false,
789            context_switches: 0,
790            steps_reset_at: 0,
791            storage: StorageMap::new(),
792            scheduler,
793            teardown: None,
794            failed_task: None,
795            unwinding_task: None,
796            switched_out_unwinding: false,
797            #[cfg(debug_assertions)]
798            has_cleaned_up: false,
799            top_level_span: tracing::Span::current(),
800            top_level_dispatch,
801            top_level_dispatch_global_ptr,
802            runnable_tasks: Vec::with_capacity(DEFAULT_INLINE_TASKS),
803            live_tasks: Vec::with_capacity(DEFAULT_INLINE_TASKS),
804        }
805    }
806
807    /// Invoke a closure with access to the current execution state. Library code uses this to gain
808    /// access to the state of the execution to influence scheduling (e.g. to register a task as
809    /// blocked).
810    #[inline]
811    #[track_caller]
812    pub fn with<F, T>(f: F) -> T
813    where
814        F: FnOnce(&mut ExecutionState) -> T,
815    {
816        Self::try_with(f).unwrap_or_else(|e| {
817            eprintln!("`ExecutionState::try_with` failed with error: {e:?}");
818            eprintln!(
819                "Backtrace for `with`: {:#?}",
820                std::backtrace::Backtrace::force_capture()
821            );
822            match e {
823                ExecutionStateBorrowError::AlreadyBorrowed => panic!("`ExecutionState::with` panicked because `ExecutionState` is already borrowed."),
824                ExecutionStateBorrowError::NotSet => panic!("`ExecutionState::with` panicked because `ExecutionState` is not set. Are you accessing a Shuttle primitive outside of a Shuttle test?"),
825            }
826        })
827    }
828
829    /// Like `with`, but returns None instead of panicking if there is no current ExecutionState or
830    /// if the current ExecutionState is already borrowed.
831    #[inline]
832    #[track_caller]
833    pub fn try_with<F, T>(f: F) -> Result<T, ExecutionStateBorrowError>
834    where
835        F: FnOnce(&mut ExecutionState) -> T,
836    {
837        // Check that the event is enabled first, so that accesses don't pay for the flag when it isn't.
838        if tracing::enabled!(tracing::Level::TRACE) && !TRACING_TRY_WITH.with(|emitting| emitting.replace(true)) {
839            struct Reset;
840            impl Drop for Reset {
841                fn drop(&mut self) {
842                    TRACING_TRY_WITH.with(|emitting| emitting.set(false));
843                }
844            }
845            let _reset = Reset;
846            trace!(
847                "ExecutionState::try_with called from {:?}",
848                std::panic::Location::caller()
849            );
850        }
851        if EXECUTION_STATE.is_set() {
852            EXECUTION_STATE.with(|cell| {
853                if let Ok(mut state) = cell.try_borrow_mut() {
854                    Ok(f(&mut state))
855                } else {
856                    Err(ExecutionStateBorrowError::AlreadyBorrowed)
857                }
858            })
859        } else {
860            Err(ExecutionStateBorrowError::NotSet)
861        }
862    }
863
864    /// A shortcut to get the current task ID
865    pub fn me() -> TaskId {
866        Self::with(|s| s.current().id())
867    }
868
869    /// If there is only one attached, unfinished task and there is at least one detached, unfinished task
870    /// then exiting the attached task will cause the whole execution to exit. As a result, the unfinished
871    /// detached tasks are truncated -- their remaining events will not be executed because the program itself
872    /// has exited. This is relevant because it means that *exiting* a task can be a visible operation
873    /// in that it affects which events are executed.
874    pub fn exit_current_truncates_execution(&self) -> bool {
875        // Strictly speaking, this is only true if there are other runnable detached tasks, but always making the main thread
876        // exit a scheduling point is simpler conceptually
877        if self.current().id() == TaskId::from(0) {
878            return true;
879        }
880
881        // If the current task is detached, then it definitely doesn't truncate the execution
882        if self.current().is_detached() {
883            return false;
884        }
885
886        let mut single_unfinished_attached = false;
887        let mut has_unfinished_detached = false;
888        for t in self.tasks.iter() {
889            let unfinished_attached = !t.finished() && !t.detached;
890            if single_unfinished_attached && unfinished_attached {
891                // there are more than one unfinished attached tasks, so one exiting won't truncate
892                return false;
893            }
894
895            single_unfinished_attached |= unfinished_attached;
896            has_unfinished_detached |= !t.finished() && t.detached;
897        }
898        has_unfinished_detached && single_unfinished_attached
899    }
900
901    fn set_labels_for_new_task(state: &ExecutionState, task_id: TaskId, name: Option<String>) {
902        LABELS.with(|cell| {
903            let mut map = cell.borrow_mut();
904
905            // If parent has labels, inherit them
906            if let Some(parent_task_id) = state.try_current().map(|t| t.id()) {
907                let parent_map = map.get(&parent_task_id);
908                if let Some(parent_map) = parent_map {
909                    let mut child_map = parent_map.clone();
910
911                    // If the parent has a `ChildLabelFn` set, use that to update the child's Labels
912                    if let Some(gen) = parent_map.get::<ChildLabelFn>() {
913                        (gen.0)(task_id, &mut child_map);
914                    }
915
916                    map.insert(task_id, child_map);
917                }
918            }
919
920            // Add any name assigned to the task to its set of Labels
921            if let Some(name) = name {
922                let m = map.entry(task_id).or_default();
923                m.insert(TaskName::from(name));
924            }
925        });
926    }
927
928    // Note: `spawn_thread`, `spawn_main_thread`, and `spawn_future` share some similar logic.
929    // Changes to one of these functions likely need to be propagated to the other two as well.
930    pub fn spawn_main_thread(
931        f: Box<dyn FnOnce() + 'static>,
932        stack_size: usize,
933        caller: &'static Location<'static>,
934    ) -> TaskId {
935        let name = "main-thread".to_string();
936        let mut clock = VectorClock::new();
937
938        let task_id = Self::with(|state| {
939            let parent_span_id = state.top_level_span.id();
940            let task_id = TaskId(state.tasks.len());
941            let tag = state.get_tag_or_default_for_current_task();
942
943            Self::set_labels_for_new_task(state, task_id, Some(name.clone()));
944
945            clock.extend(task_id); // and extend it with an entry for the new thread
946
947            let schedule_len = CurrentSchedule::len();
948
949            let task = Task::from_closure(
950                f,
951                stack_size,
952                task_id,
953                Some(name),
954                clock,
955                parent_span_id,
956                schedule_len,
957                tag,
958                None,
959                TaskSignature::new_parentless(caller),
960            );
961            state.add_task(task);
962
963            task_id
964        });
965        crate::annotations::record_task_created(task_id, false);
966        task_id
967    }
968
969    // Note: `spawn_thread`, `spawn_main_thread`, and `spawn_future` share some similar logic.
970    // Changes to one of these functions likely need to be propagated to the other two as well.
971    /// Spawn a new task for a future. This doesn't create a yield point; the caller should do that
972    /// if it wants to give the new task a chance to run immediately.
973    pub fn spawn_future<F>(
974        future: F,
975        stack_size: usize,
976        name: Option<String>,
977        caller: &'static Location<'static>,
978    ) -> TaskId
979    where
980        F: Future<Output = ()> + 'static,
981    {
982        thread::switch();
983        let task_id = Self::with(|state| {
984            let schedule_len = CurrentSchedule::len();
985            let parent_span_id = state.top_level_span.id();
986
987            let task_id = TaskId(state.tasks.len());
988            let tag = state.get_tag_or_default_for_current_task();
989
990            Self::set_labels_for_new_task(state, task_id, name.clone());
991
992            let clock = state.increment_clock_mut(); // Increment the parent's clock
993            clock.extend(task_id); // and extend it with an entry for the new task
994
995            let task = Task::from_future(
996                future,
997                stack_size,
998                task_id,
999                name,
1000                clock.clone(),
1001                parent_span_id,
1002                schedule_len,
1003                tag,
1004                Some(state.current().id()),
1005                state.current_mut().signature.new_child(caller),
1006            );
1007
1008            state.add_task(task);
1009
1010            task_id
1011        });
1012        crate::annotations::record_task_created(task_id, true);
1013        task_id
1014    }
1015
1016    // Note: `spawn_thread`, `spawn_main_thread`, and `spawn_future` share some similar logic.
1017    // Changes to one of these functions likely need to be propagated to the other two as well.
1018    pub fn spawn_thread(
1019        f: Box<dyn FnOnce() + 'static>,
1020        stack_size: usize,
1021        name: Option<String>,
1022        mut initial_clock: Option<VectorClock>,
1023        caller: &'static Location<'static>,
1024    ) -> TaskId {
1025        thread::switch();
1026        let task_id = Self::with(|state| {
1027            let parent_span_id = state.top_level_span.id();
1028            let task_id = TaskId(state.tasks.len());
1029            let tag = state.get_tag_or_default_for_current_task();
1030
1031            Self::set_labels_for_new_task(state, task_id, name.clone());
1032
1033            let clock = if let Some(ref mut clock) = initial_clock {
1034                clock
1035            } else {
1036                // Inherit the clock of the parent thread (which spawned this task)
1037                state.increment_clock_mut()
1038            };
1039            clock.extend(task_id); // and extend it with an entry for the new thread
1040            let clock = clock.clone();
1041
1042            let task = Task::from_closure(
1043                f,
1044                stack_size,
1045                task_id,
1046                name,
1047                clock,
1048                parent_span_id,
1049                CurrentSchedule::len(),
1050                tag,
1051                Some(state.current().id()),
1052                state.current_mut().signature.new_child(caller),
1053            );
1054            state.add_task(task);
1055
1056            task_id
1057        });
1058        crate::annotations::record_task_created(task_id, false);
1059        task_id
1060    }
1061
1062    /// Tear down the execution, which ended as `kind` says, before it goes out of `EXECUTION_STATE`
1063    /// scope: drop what its unfinished tasks leave behind, the tasks' task-local values, and the
1064    /// execution's statics, while their destructors can still use Shuttle. Returns how that fails
1065    /// the test, if it does.
1066    ///
1067    /// A finished execution's unfinished tasks (detached future tasks, and tasks that destructors
1068    /// spawn during teardown) are torn down as if the execution cancelled them as it ended. While a
1069    /// task is torn down, it is the current task: its destructors can use Shuttle as the running
1070    /// task can, and what they do is attributed to it.
1071    ///
1072    /// * A task that never ran drops its function, and a future task that is parked between polls,
1073    ///   the usual state of an unfinished future task, drops its future, the way an async runtime
1074    ///   drops the futures of its tasks when it shuts down (see `Continuation::cancel`). They do
1075    ///   this on their own stacks, and then drop their task-local values. So a destructor that
1076    ///   blocks, on a lock that another unfinished task holds say, waits for teardown to drop that
1077    ///   task. A panic is an ordinary panic, which fails the test once teardown is done.
1078    /// * A future task that stopped in the middle of `poll` has user code on its stack, which only
1079    ///   unwinding the stack can drop. Teardown unwinds these stacks one at a time, once no other
1080    ///   task can make progress, so that the locks that the other tasks held are free. Their
1081    ///   destructors can use Shuttle too, but cannot wait: one that blocks, or panics, while a stack
1082    ///   is unwound aborts the process, as any panic during unwinding does. A task that catches the
1083    ///   unwind is unwound again at its next scheduling point.
1084    ///
1085    /// Nothing is scheduled. Teardown resumes the tasks it cancelled itself, in a fixed order that
1086    /// doesn't extend the schedule, so it is the same when the schedule is replayed (see
1087    /// `next_teardown_job`). A destructor that blocks waits until another destructor wakes it, and
1088    /// one that yields lets the others run; but if every destructor left blocks, nothing can wake
1089    /// them, and that fails the test like a deadlock. The step bound applies to destructors too (see
1090    /// `maybe_yield_in_teardown`). The execution's statics go last, as if the main thread dropped
1091    /// them, once nothing else is left that could still use them.
1092    ///
1093    /// A stopped or failed execution is abandoned instead. Shuttle's own destructors skip their
1094    /// bookkeeping (see `should_stop`), and panics are ignored, so that a failure is what gets
1095    /// reported. The stacks of its unfinished tasks are leaked. A failed execution's functions of
1096    /// tasks that never ran, and futures of parked future tasks, are leaked too, unless
1097    /// `UngracefulShutdownConfig::continuation_function_behavior` says to drop them, in which case
1098    /// they are cancelled as above. A stopped execution's functions of tasks that never ran are
1099    /// dropped that way. Every function that is dropped goes before any stack is freed, since a
1100    /// scoped thread's function borrows from its parent's stack. Task-local values and statics are
1101    /// dropped as in a finished execution.
1102    ///
1103    /// A task that was unwinding a panic when a stopped execution stopped finishes unwinding it first,
1104    /// and the panic fails the test (see `finish_unwinding`). A failed execution's is leaked like the
1105    /// rest, as running more of a failed execution risks a panic that aborts the process. If teardown
1106    /// can't tell which task it is (see `record_unwinding_task`), that fails the test too, and the
1107    /// rest of the execution is abandoned: unwinding that task's stack again would abort the process.
1108    fn tear_down(kind: TeardownKind) -> Option<TeardownFailure> {
1109        let mut plan = TeardownPlan::default();
1110        let mut report = TeardownReport::new();
1111        // Here, on the executor's stack, `std::thread::panicking()` says whether a task that switched
1112        // out is unwinding a panic still (see `record_unwinding_task`).
1113        let unwinding = std::thread::panicking();
1114        let lost_panic = Self::with(|state| {
1115            state.current_task = kind.final_state();
1116            state.next_task = ScheduledTask::None;
1117            state.teardown = Some(Teardown::new(kind, CurrentSchedule::len()));
1118            if kind != TeardownKind::Failed && unwinding {
1119                let unwinding_task = state.unwinding_task.take().map(|(id, _)| id);
1120                plan.unwinding_task = unwinding_task.filter(|&id| state.get(id).suspended());
1121            }
1122            let lost_panic = unwinding && plan.unwinding_task.is_none() && kind != TeardownKind::Failed;
1123            if lost_panic {
1124                // So that no stack is unwound.
1125                state.abandon_teardown();
1126            }
1127            lost_panic
1128        });
1129        if lost_panic {
1130            report.fail(
1131                "a task was still unwinding a panic when the execution ended, but it panicked while another task was \
1132                 unwinding a panic, so Shuttle can't tell which task it is, to let it finish unwinding"
1133                    .into(),
1134            );
1135        }
1136        // The panic hook goes back to reporting panics when teardown is done, however it ends.
1137        struct ReportPanicsAgain;
1138        impl Drop for ReportPanicsAgain {
1139            fn drop(&mut self) {
1140                TEARDOWN_IGNORES_PANICS.set(false);
1141            }
1142        }
1143        let report_panics_again = ReportPanicsAgain;
1144        TEARDOWN_IGNORES_PANICS.set(Self::with(|state| state.teardown().kind.abandons()));
1145
1146        loop {
1147            match Self::with(|state| state.next_teardown_job(&mut plan)) {
1148                TeardownJob::FinishUnwinding(id) => Self::finish_unwinding(id, &mut report),
1149                TeardownJob::OnOwnStack(id, resumption) => {
1150                    Self::run_on_own_stack(id, resumption, &mut plan, &mut report)
1151                }
1152                TeardownJob::Unwind(id) => Self::unwind_task(id, &mut report),
1153                TeardownJob::Leak(id) => Self::leak_task(id, &mut report),
1154                TeardownJob::LeakFunction(id) => {
1155                    Self::with(|state| {
1156                        let task = state.get(id);
1157                        let continuation = task
1158                            .continuation
1159                            .as_ref()
1160                            .expect("a task that never ran has a continuation");
1161                        continuation.borrow_mut().leak_function();
1162                    });
1163                    Self::finish_torn_down_task(id, &mut report);
1164                }
1165                TeardownJob::TaskLocals(id) => Self::drop_leftover_task_locals(id, &mut report),
1166                TeardownJob::Static(value) => Self::drop_static(value, &mut report),
1167                TeardownJob::Stuck => {
1168                    let (deadlocked, blocked_tasks) = Self::with(|state| {
1169                        let teardown = state.teardown();
1170                        let deadlocked = !teardown.kind.abandons() && teardown.exceeded_step_bound.is_none();
1171                        let blocked = plan
1172                            .suspended
1173                            .iter()
1174                            .map(|&(id, _)| state.get(id).format_for_deadlock());
1175                        let blocked = blocked.collect::<Vec<_>>();
1176                        state.abandon_teardown();
1177                        (deadlocked, blocked)
1178                    });
1179                    TEARDOWN_IGNORES_PANICS.set(true);
1180                    if deadlocked {
1181                        report.deadlocked(blocked_tasks);
1182                    }
1183                    for (id, _) in std::mem::take(&mut plan.suspended) {
1184                        Self::leak_task(id, &mut report);
1185                    }
1186                }
1187                TeardownJob::Abandon(max_steps) => {
1188                    // The panic that stopped the destructor fails the test, unless the destructor
1189                    // caught it.
1190                    report.fail(format!(
1191                        "a destructor exceeded the step bound ({max_steps}) while it was being dropped at the end of \
1192                         the execution"
1193                    ));
1194                    Self::with(|state| state.abandon_teardown());
1195                    TEARDOWN_IGNORES_PANICS.set(true);
1196                }
1197                TeardownJob::Done => {
1198                    // What the tasks still hold goes last, as each task (see `drop_task_leftovers`).
1199                    // Destructors can spawn tasks, which are torn down too.
1200                    if !Self::drop_task_leftovers(&mut report) {
1201                        break;
1202                    }
1203                }
1204            }
1205        }
1206
1207        // The tasks' labels, and tags that no task holds, while `EXECUTION_STATE` is set. Their
1208        // destructors may look up tasks: formatting a `TaskId` looks up its name, say. In the order
1209        // the tasks were created, so that the destructors run in the same order every time.
1210        let tags = TASK_ID_TO_TAGS.with(|cell| std::mem::take(&mut *cell.borrow_mut()));
1211        let mut tags = tags.into_iter().collect::<Vec<_>>();
1212        tags.sort_unstable_by_key(|&(id, _)| id);
1213        for (_, tag) in tags {
1214            report.catch(move || drop(tag));
1215        }
1216        let labels = LABELS.with(|cell| std::mem::take(&mut *cell.borrow_mut()));
1217        let mut labels = labels.into_iter().collect::<Vec<_>>();
1218        labels.sort_unstable_by_key(|&(id, _)| id);
1219        for (_, labels) in labels {
1220            report.catch(move || drop(labels));
1221        }
1222        let tasks = Self::with(|state| std::mem::take(&mut state.tasks));
1223        report.catch(move || drop(tasks));
1224
1225        Self::with(|state| {
1226            state.teardown = None;
1227            #[cfg(debug_assertions)]
1228            {
1229                state.has_cleaned_up = true;
1230            }
1231        });
1232        drop(report_panics_again);
1233        report.finish()
1234    }
1235
1236    /// Choose what execution teardown does next (see `tear_down`).
1237    fn next_teardown_job(&mut self, plan: &mut TeardownPlan) -> TeardownJob {
1238        let teardown = self.teardown();
1239        let kind = teardown.kind;
1240        // Once a destructor has exceeded the step bound, which fails the test, the rest of the
1241        // execution is abandoned: the tasks that are left are leaked, as destructors that spawn tasks
1242        // whose destructors spawn tasks, say, would go on forever.
1243        let exceeded_step_bound = teardown.exceeded_step_bound;
1244        if let Some(max_steps) = exceeded_step_bound.filter(|_| !kind.abandons()) {
1245            return TeardownJob::Abandon(max_steps);
1246        }
1247        let abandoned = exceeded_step_bound.is_some();
1248        if abandoned && !plan.suspended.is_empty() {
1249            return TeardownJob::Stuck;
1250        }
1251
1252        // A task that was unwinding a panic when the execution stopped finishes unwinding it first.
1253        if let Some(id) = plan.unwinding_task.take() {
1254            return TeardownJob::FinishUnwinding(id);
1255        }
1256
1257        // A cancelled task that blocked, and has been woken since, goes first.
1258        if let Some(i) = plan
1259            .suspended
1260            .iter()
1261            .position(|&(id, blocked)| blocked && self.get(id).runnable())
1262        {
1263            plan.idle_yields = 0;
1264            return TeardownJob::OnOwnStack(plan.suspended.remove(i).0, Resumption::Resume);
1265        }
1266
1267        // Then the tasks that teardown hasn't looked at yet, in the order they were created.
1268        let leaks = UNGRACEFUL_SHUTDOWN_CONFIG.get().continuation_function_behavior.leaks();
1269        let (drop_functions, drop_parked_futures) = match kind {
1270            _ if abandoned => (false, false),
1271            TeardownKind::Finished => (true, true),
1272            TeardownKind::Stopped => (true, false),
1273            TeardownKind::Failed => (!leaks, !leaks),
1274        };
1275        while let Some(task) = self.tasks.get(plan.next_task) {
1276            plan.next_task += 1;
1277            let id = task.id();
1278            if task.finished() {
1279                // A future task's future can set task-local values after the task dropped its own
1280                // ones (see `Wrapper::finish`).
1281                if task.has_locals() {
1282                    return TeardownJob::TaskLocals(id);
1283                }
1284            } else if task.never_ran() {
1285                plan.idle_yields = 0;
1286                return if drop_functions {
1287                    TeardownJob::OnOwnStack(id, Resumption::Cancel)
1288                } else {
1289                    TeardownJob::LeakFunction(id)
1290                };
1291            } else if drop_parked_futures && task.parked_between_polls() {
1292                plan.idle_yields = 0;
1293                return TeardownJob::OnOwnStack(id, Resumption::Cancel);
1294            } else {
1295                // The task is in the middle of user code.
1296                plan.stacks.push_back(id);
1297            }
1298        }
1299
1300        // A cancelled task that yielded runs again, in turn. While stacks are waiting to be unwound,
1301        // for a number of turns only, if no task finishes or blocks meanwhile (see
1302        // `TEARDOWN_YIELD_ROUNDS`).
1303        let unwinds_wait = kind == TeardownKind::Finished && !abandoned && !plan.stacks.is_empty();
1304        let yielded = |&(id, blocked): &(TaskId, bool)| !blocked && self.get(id).runnable();
1305        let num_yielded = plan.suspended.iter().filter(|task| yielded(task)).count();
1306        if num_yielded > 0 && (!unwinds_wait || plan.idle_yields < self.teardown_yield_budget(num_yielded)) {
1307            plan.idle_yields += 1;
1308            let i = plan.suspended.iter().position(yielded).unwrap();
1309            return TeardownJob::OnOwnStack(plan.suspended.remove(i).0, Resumption::Resume);
1310        }
1311
1312        // A finished execution's tasks in the middle of `poll` are unwound now, once no other task
1313        // can make progress.
1314        if unwinds_wait {
1315            plan.idle_yields = 0;
1316            return TeardownJob::Unwind(plan.stacks.pop_front().unwrap());
1317        }
1318
1319        // A static goes only once no cancelled task is left suspended, whose stack could still refer
1320        // to it.
1321        if !plan.suspended.is_empty() {
1322            return TeardownJob::Stuck;
1323        }
1324
1325        if let Some(value) = self.storage.pop() {
1326            return TeardownJob::Static(value);
1327        }
1328
1329        // An abandoned execution's stacks are leaked last, once the functions of the tasks that never
1330        // ran, which could borrow from them, are gone.
1331        if let Some(id) = plan.stacks.pop_front() {
1332            return TeardownJob::Leak(id);
1333        }
1334
1335        TeardownJob::Done
1336    }
1337
1338    /// How many turns in a row the tasks that yield get while stacks are waiting to be unwound (see
1339    /// `TEARDOWN_YIELD_ROUNDS`).
1340    fn teardown_yield_budget(&self, num_yielded: usize) -> usize {
1341        let rounds = TEARDOWN_YIELD_ROUNDS.saturating_mul(num_yielded);
1342        match self.teardown_step_bound() {
1343            Some(max_steps) => rounds.min(max_steps.saturating_sub(self.teardown().steps) / 4),
1344            None => rounds,
1345        }
1346    }
1347
1348    /// Abandon the rest of a finished execution's teardown, which has failed (see `tear_down`).
1349    fn abandon_teardown(&mut self) {
1350        let teardown = self.teardown_mut();
1351        if teardown.kind == TeardownKind::Finished {
1352            teardown.kind = TeardownKind::Failed;
1353            self.current_task = TeardownKind::Failed.final_state();
1354        }
1355    }
1356
1357    /// Resume a task on its own stack (see `Resumption`), until it finishes or switches out.
1358    fn run_on_own_stack(id: TaskId, resumption: Resumption, plan: &mut TeardownPlan, report: &mut TeardownReport) {
1359        let continuation = Self::with(|state| {
1360            state.begin_teardown_step(TeardownStep::Cancel(id));
1361            let task = state.get_mut(id);
1362            if resumption == Resumption::Cancel {
1363                // A parked future task is asleep, but it runs now, to drop its future.
1364                task.unblock();
1365            }
1366            task.continuation
1367                .clone()
1368                .expect("an unfinished task has a continuation")
1369        });
1370
1371        Execution::enter_task_span();
1372        let result = panic::catch_unwind(panic::AssertUnwindSafe(|| {
1373            let mut continuation = continuation.borrow_mut();
1374            // Only now can the task switch out (see `maybe_yield_in_teardown`).
1375            Self::with(|state| state.teardown_mut().on_own_stack = true);
1376            if resumption == Resumption::Cancel {
1377                continuation.cancel()
1378            } else {
1379                continuation.resume()
1380            }
1381        }));
1382        Self::with(|state| state.teardown_mut().on_own_stack = false);
1383        Execution::exit_task_span(matches!(result, Ok(false)));
1384        drop(continuation);
1385        // A panic's payload goes while the step is under way, so its destructor runs as the task.
1386        let finished = result.unwrap_or_else(|payload| {
1387            report.panicked(payload);
1388            true
1389        });
1390        let blocked = Self::with(|state| {
1391            state.end_teardown_step();
1392            !state.get(id).runnable()
1393        });
1394
1395        if finished {
1396            plan.idle_yields = 0;
1397            Self::finish_torn_down_task(id, report);
1398        } else {
1399            // A destructor blocked, or yielded (see `maybe_yield_in_teardown`).
1400            if blocked {
1401                plan.idle_yields = 0;
1402            }
1403            plan.suspended.push((id, blocked));
1404        }
1405    }
1406
1407    /// Let a task that switched out while it unwound a panic, and is unwinding it still when a
1408    /// stopped execution stopped, finish unwinding it, on its own stack (see `tear_down`). The panic
1409    /// then fails the test. The task runs until it has caught the panic, if it does, but no further.
1410    /// A task that blocks while the panic unwinds, or exceeds the step bound, waits for tasks that no
1411    /// longer run, but cannot be stopped with a panic, as that would abort the process. It switches
1412    /// out instead (see `maybe_yield_in_teardown`), and fails the test. Then its stack is leaked,
1413    /// which leaves the thread panicking.
1414    fn finish_unwinding(id: TaskId, report: &mut TeardownReport) {
1415        let (continuation, blocked) = Self::with(|state| {
1416            state.begin_teardown_step(TeardownStep::FinishUnwinding(id));
1417            let task = state.get(id);
1418            let continuation = task
1419                .continuation
1420                .clone()
1421                .expect("an unfinished task has a continuation");
1422            (continuation, !task.runnable())
1423        });
1424
1425        // A task that blocked already waits for a task that no longer runs.
1426        let result = if blocked {
1427            Ok(false)
1428        } else {
1429            Execution::enter_task_span();
1430            let result = panic::catch_unwind(panic::AssertUnwindSafe(|| {
1431                let mut continuation = continuation.borrow_mut();
1432                Self::with(|state| state.teardown_mut().on_own_stack = true);
1433                continuation.resume()
1434            }));
1435            Self::with(|state| state.teardown_mut().on_own_stack = false);
1436            Execution::exit_task_span(matches!(result, Ok(false)));
1437            result
1438        };
1439        drop(continuation);
1440
1441        // On the executor's stack, `std::thread::panicking()` says whether the task is unwinding the
1442        // panic still (see `record_unwinding_task`).
1443        let stuck = matches!(result, Ok(false)) && std::thread::panicking();
1444        // A panic's payload goes while the step is under way, so its destructor runs as the task.
1445        let finished = result.unwrap_or_else(|payload| {
1446            report.late_panic(payload);
1447            true
1448        });
1449        if stuck {
1450            let (name, exceeded_step_bound) =
1451                Self::with(|state| (state.get(id).display_name(), state.teardown().exceeded_step_bound));
1452            let why = match exceeded_step_bound {
1453                Some(max_steps) => format!("exceeded the step bound ({max_steps})"),
1454                None => "blocked".into(),
1455            };
1456            report.fail(format!(
1457                "{name} was unwinding a panic when the execution stopped, and {why} before it had finished unwinding it"
1458            ));
1459        }
1460        Self::with(|state| state.end_teardown_step());
1461
1462        if finished {
1463            Self::finish_torn_down_task(id, report);
1464        } else {
1465            // The task caught the panic, or is stuck.
1466            Self::leak_task(id, report);
1467        }
1468    }
1469
1470    /// Unwind the stack of a finished execution's task that is in the middle of `poll` (see
1471    /// `tear_down`).
1472    fn unwind_task(id: TaskId, report: &mut TeardownReport) {
1473        let continuation = Self::with(|state| {
1474            state.begin_teardown_step(TeardownStep::Unwind(id));
1475            let task = state.get_mut(id);
1476            // The task may have been blocked, but it runs now, to drop what is on its stack.
1477            task.unblock();
1478            task.take_continuation()
1479        });
1480        let continuation = continuation.expect("an unfinished task has a continuation");
1481
1482        // A panic while the stack unwinds aborts the process, so the panic hook reports every one.
1483        let ignoring_panics = TEARDOWN_IGNORES_PANICS.replace(false);
1484        // Among other things, this reinstates the task's default dispatcher, so that the guards on
1485        // its stack restore their priors in order as it unwinds (see `ParkedDefault`).
1486        Execution::enter_task_span();
1487        let result = panic::catch_unwind(panic::AssertUnwindSafe(move || {
1488            let continuation = Rc::try_unwrap(continuation).map_err(|_| ());
1489            let mut continuation = continuation.expect("teardown owns the continuation").into_inner();
1490            // Only now can the task switch, if it catches the unwind (see `maybe_yield_in_teardown`).
1491            Self::with(|state| state.teardown_mut().on_own_stack = true);
1492            continuation.unwind_stack();
1493        }));
1494        Self::with(|state| state.teardown_mut().on_own_stack = false);
1495        Execution::exit_task_span(false);
1496        TEARDOWN_IGNORES_PANICS.set(ignoring_panics);
1497        if let Err(payload) = result {
1498            report.panicked(payload);
1499        }
1500        Self::with(|state| {
1501            state.get_mut(id).lose_stack();
1502            state.end_teardown_step();
1503        });
1504
1505        Self::finish_torn_down_task(id, report);
1506    }
1507
1508    /// Leak the stack of an abandoned execution's task (see `tear_down`).
1509    fn leak_task(id: TaskId, report: &mut TeardownReport) {
1510        let (parked_default, continuation) = Self::with(|state| {
1511            let task = state.get_mut(id);
1512            task.lose_stack();
1513            (task.parked_default.take(), task.take_continuation())
1514        });
1515        // The parked default has to go before the task's stack does (see `ParkedDefault`).
1516        drop(parked_default);
1517        if let Some(continuation) = continuation {
1518            let continuation = Rc::try_unwrap(continuation).map_err(|_| ());
1519            let mut continuation = continuation.expect("teardown owns the continuation").into_inner();
1520            continuation.leak_stack();
1521        }
1522
1523        Self::finish_torn_down_task(id, report);
1524    }
1525
1526    /// Finish a task that execution teardown is done with, and drop the task-local values it left.
1527    fn finish_torn_down_task(id: TaskId, report: &mut TeardownReport) {
1528        crate::annotations::record_teardown_step(id);
1529        crate::annotations::record_task_terminated();
1530        Self::with(|state| {
1531            let task = state.get_mut(id);
1532            task.finish();
1533            // A task that joins this one can go on (see `JoinHandle::join`).
1534            if let Some(waiter) = task.take_waiter() {
1535                if !state.get(waiter).finished() {
1536                    state.get_mut(waiter).unblock();
1537                }
1538            }
1539        });
1540        Self::drop_leftover_task_locals(id, report);
1541    }
1542
1543    /// Drop the task-local values that a task left, as the task, once its stack is gone (see
1544    /// `tear_down` and `Task::stand_in`).
1545    fn drop_leftover_task_locals(id: TaskId, report: &mut TeardownReport) {
1546        let state_before = Self::with(|state| {
1547            if !state.get(id).has_locals() {
1548                return None;
1549            }
1550            state.begin_teardown_step(TeardownStep::Leftovers(id));
1551            Some(state.get_mut(id).stand_in())
1552        });
1553        let Some(state_before) = state_before else {
1554            return;
1555        };
1556
1557        // See `pop_local` for why this loop looks slightly funky.
1558        while let Some(local) = Self::with(|state| state.get_mut(id).pop_local()) {
1559            report.catch(move || drop(local));
1560        }
1561
1562        Self::with(|state| {
1563            state.get_mut(id).stop_standing_in(state_before);
1564            state.end_teardown_step();
1565        });
1566    }
1567
1568    /// Drop what the tasks still hold whose destructors may run user code, at the end of teardown:
1569    /// their spans, which a `tracing` subscriber sees close, and their tags. Each task's go as the
1570    /// task (see `Task::stand_in`). Returns whether there was anything to drop.
1571    fn drop_task_leftovers(report: &mut TeardownReport) -> bool {
1572        let ids = Self::with(|state| {
1573            TASK_ID_TO_TAGS.with(|tags| {
1574                let tags = tags.borrow();
1575                let has_tag = |id| !tags.is_empty() && tags.contains_key(&id);
1576                let tasks = state
1577                    .tasks
1578                    .iter()
1579                    .filter(|task| task.has_leftovers() || has_tag(task.id()));
1580                tasks.map(Task::id).collect::<Vec<_>>()
1581            })
1582        });
1583        for &id in &ids {
1584            let (state_before, leftovers) = Self::with(|state| {
1585                state.begin_teardown_step(TeardownStep::Leftovers(id));
1586                // `current::set_tag_for_task` keeps a reference to the tag too, which can be the last.
1587                let tag = TASK_ID_TO_TAGS.with(|tags| tags.borrow_mut().remove(&id));
1588                let task = state.get_mut(id);
1589                (task.stand_in(), (task.take_leftovers(), tag))
1590            });
1591            report.catch(move || drop(leftovers));
1592            Self::with(|state| {
1593                state.get_mut(id).stop_standing_in(state_before);
1594                state.end_teardown_step();
1595            });
1596        }
1597        !ids.is_empty()
1598    }
1599
1600    /// Drop one of the execution's statics, as the main thread (see `tear_down`).
1601    fn drop_static(value: Box<dyn Any>, report: &mut TeardownReport) {
1602        let main_thread = TaskId(0);
1603        let state_before = Self::with(|state| {
1604            state.begin_teardown_step(TeardownStep::Static);
1605            state.get_mut(main_thread).stand_in()
1606        });
1607
1608        report.catch(move || drop(value));
1609        // The static's destructor may have set task-local values of the main thread's, which go with
1610        // it.
1611        while let Some(local) = Self::with(|state| state.get_mut(main_thread).pop_local()) {
1612            report.catch(move || drop(local));
1613        }
1614
1615        Self::with(|state| {
1616            state.get_mut(main_thread).stop_standing_in(state_before);
1617            state.end_teardown_step();
1618        });
1619    }
1620
1621    /// Make `step` what execution teardown is doing, with its task as the current task (see
1622    /// `tear_down`).
1623    fn begin_teardown_step(&mut self, step: TeardownStep) {
1624        let id = match step {
1625            TeardownStep::Cancel(id)
1626            | TeardownStep::Unwind(id)
1627            | TeardownStep::FinishUnwinding(id)
1628            | TeardownStep::Leftovers(id) => id,
1629            // The main thread stands in for statics.
1630            TeardownStep::Static => TaskId(0),
1631            TeardownStep::Idle => unreachable!("teardown is always doing something during a step"),
1632        };
1633        self.current_task = ScheduledTask::Some(id);
1634        self.teardown_mut().step = step;
1635        TEARDOWN_STEP_PANICS.set(0);
1636        crate::annotations::record_teardown_step(id);
1637    }
1638
1639    /// End the step that `begin_teardown_step` began, if one is under way.
1640    fn end_teardown_step(&mut self) {
1641        let teardown = self.teardown_mut();
1642        teardown.step = TeardownStep::Idle;
1643        self.current_task = teardown.kind.final_state();
1644    }
1645
1646    fn teardown(&self) -> &Teardown {
1647        self.teardown.as_ref().expect("the execution is being torn down")
1648    }
1649
1650    fn teardown_mut(&mut self) -> &mut Teardown {
1651        self.teardown.as_mut().expect("the execution is being torn down")
1652    }
1653
1654    /// Whether the panic hook is to stay silent about a panic with `payload`, as execution teardown
1655    /// ignores the panics it catches now (see `tear_down`). A panic in a destructor while another
1656    /// panic unwinds aborts the process. So the hook reports the panic that says so, and any panic
1657    /// after the first in a step, or in what `TeardownReport::catch` drops, which may cause that.
1658    pub(crate) fn teardown_ignores_panic(payload: &dyn Any) -> bool {
1659        let panics = TEARDOWN_STEP_PANICS.get().saturating_add(1);
1660        TEARDOWN_STEP_PANICS.set(panics);
1661        let aborts = payload.downcast_ref::<&str>() == Some(&"panic in a destructor during cleanup");
1662        TEARDOWN_IGNORES_PANICS.get() && panics == 1 && !aborts
1663    }
1664
1665    /// Whether execution teardown is unwinding the current task's stack (see `tear_down`). That is
1666    /// not a panic, although `std::thread::panicking()` says otherwise, so it shouldn't poison a
1667    /// lock, say.
1668    pub fn unwinding_for_teardown() -> bool {
1669        Self::try_with(|state| {
1670            state
1671                .teardown
1672                .as_ref()
1673                .is_some_and(|teardown| matches!(teardown.step, TeardownStep::Unwind(_)))
1674        })
1675        .unwrap_or(false)
1676    }
1677
1678    /// Drop the current task's task-local values, as a task does when it finishes. A destructor can
1679    /// set another task-local value, which is dropped too (see `Task::pop_local`).
1680    pub fn drop_task_locals() {
1681        // See `pop_local` for why this loop looks slightly funky.
1682        while let Some(local) = Self::with(|state| state.current_mut().pop_local()) {
1683            tracing::trace!("dropping task-local value {:p}", local);
1684            drop(local);
1685        }
1686    }
1687
1688    /// Stop a destructor that execution teardown cannot let go on (see `maybe_yield_in_teardown`).
1689    #[cold]
1690    fn stop_destructor(stall: TeardownStall) -> ! {
1691        if std::thread::panicking() {
1692            // This aborts the process, so report it in any case.
1693            TEARDOWN_IGNORES_PANICS.set(false);
1694        }
1695        let (name, step) = Self::with(|state| (state.current().display_name(), state.teardown().step));
1696        let who = match step {
1697            TeardownStep::Static => format!("A static's destructor, dropped as {name},"),
1698            TeardownStep::Leftovers(_) => format!("A destructor of something that {name} left behind"),
1699            _ => name,
1700        };
1701        match stall {
1702            TeardownStall::Blocked => {
1703                let reason = if std::thread::panicking() || matches!(step, TeardownStep::Unwind(_)) {
1704                    "while unwinding a stack, which cannot be suspended"
1705                } else {
1706                    "outside any task's stack, where it cannot wait"
1707                };
1708                panic!(
1709                    "{who} blocked while it was being dropped at the end of the execution, {reason}. A destructor made a \
1710                     blocking call, such as locking a mutex that another task holds, receiving from an empty channel, \
1711                     joining a task, or `block_on` on a future that cannot complete."
1712                )
1713            }
1714            TeardownStall::ExceededStepBound(max_steps) => panic!(
1715                "{who} exceeded the step bound ({max_steps}) while it was being dropped at the end of the execution. A \
1716                 destructor may be spinning, waiting for something that no longer happens: once the execution is over, \
1717                 only destructors run."
1718            ),
1719        }
1720    }
1721
1722    /// Determine whether the execution has finished.
1723    pub fn is_finished(&self) -> bool {
1724        self.current_task == ScheduledTask::Stopped || self.current_task == ScheduledTask::Finished
1725    }
1726
1727    /// Invoke the scheduler to decide which task to schedule next. Returns true if the chosen task
1728    /// is different from the currently running task, indicating that the current task should yield
1729    /// its execution.
1730    pub fn maybe_yield() -> bool {
1731        let decision = Self::with(|state| {
1732            if state.teardown.is_some() {
1733                return state.maybe_yield_in_teardown();
1734            }
1735
1736            if std::thread::panicking() {
1737                if !state.switched_out_unwinding {
1738                    state.record_unwinding_task();
1739                }
1740                return Ok(true);
1741            }
1742            if state.switched_out_unwinding {
1743                // No task is unwinding a panic, as the tasks share the OS thread.
1744                state.forget_unwinding_task();
1745            }
1746
1747            debug_assert!(
1748                matches!(state.current_task, ScheduledTask::Some(_)) && state.next_task == ScheduledTask::None,
1749                "we're inside a task and scheduler should not yet have run"
1750            );
1751
1752            let result = state.schedule();
1753            // If scheduling failed, yield so that the outer scheduling loop can handle it.
1754            if result.is_err() {
1755                return Ok(true);
1756            }
1757
1758            // If the next task is the same as the current one, we can skip the context switch
1759            // and just advance to the next task immediately.
1760            if state.current_task == state.next_task {
1761                state.advance_to_next_task();
1762                Ok(false)
1763            } else {
1764                Ok(true)
1765            }
1766        });
1767        // Panic outside of `Self::with`, so the panic hook can name the task.
1768        decision.unwrap_or_else(|stall| Self::stop_destructor(stall))
1769    }
1770
1771    /// The current task switches out while it unwinds a panic, and no task was unwinding one when it
1772    /// was resumed (see `switched_out_unwinding`), so the panic is its own. (`std::thread::panicking()`
1773    /// can't tell which task panics: tasks share the OS thread. Between steps, on the executor's
1774    /// stack, it says exactly whether a task that switched out is unwinding a panic.) Until the task
1775    /// has finished unwinding the panic, the execution waits for it, even if it is detached.
1776    /// Otherwise the execution could end before the task resumes, and its panic would be lost. There
1777    /// is one such task at most: while it is unwinding, a task that is resumed can't tell a panic of
1778    /// its own from it.
1779    #[cold]
1780    fn record_unwinding_task(&mut self) {
1781        debug_assert!(self.unwinding_task.is_none());
1782        self.switched_out_unwinding = true;
1783        if let Some(me) = self.current_task.id() {
1784            let detached = std::mem::replace(&mut self.get_mut(me).detached, false);
1785            self.unwinding_task = Some((me, detached));
1786        }
1787    }
1788
1789    /// No task is unwinding a panic any more: the task that was recorded doing so (see
1790    /// `record_unwinding_task`) is detached again if it was before.
1791    #[cold]
1792    fn forget_unwinding_task(&mut self) {
1793        self.switched_out_unwinding = false;
1794        if let Some((task, detached)) = self.unwinding_task.take() {
1795            self.get_mut(task).detached = detached;
1796        }
1797    }
1798
1799    /// Detach a task, so that the execution doesn't wait for it to finish, as when its `JoinHandle`
1800    /// is dropped. A task that is unwinding a panic is detached once it has finished unwinding (see
1801    /// `record_unwinding_task`).
1802    pub fn detach(&mut self, id: TaskId) {
1803        match &mut self.unwinding_task {
1804            Some((task, detached)) if *task == id => *detached = true,
1805            _ => self.get_mut(id).detach(),
1806        }
1807    }
1808
1809    /// `maybe_yield` during execution teardown, which schedules nothing (see `tear_down`). Each
1810    /// scheduling point counts as a step, which the step bound limits as in a running execution.
1811    /// A task that teardown runs on its own stack switches out when it blocks or yields, so that
1812    /// teardown can run others. Anywhere else, or while a panic unwinds, nothing can be suspended,
1813    /// so a destructor that blocks is stopped with a panic. A task that finishes unwinding a panic
1814    /// switches out instead, and as soon as it has caught the panic (see `finish_unwinding`).
1815    #[cold]
1816    fn maybe_yield_in_teardown(&mut self) -> Result<bool, TeardownStall> {
1817        // Destructors do what tasks do, so `current::context_switches` counts their scheduling points
1818        // too.
1819        self.context_switches += 1;
1820        let panicking = std::thread::panicking();
1821        let yielded = std::mem::take(&mut self.has_yielded);
1822        let max_steps = self.teardown_step_bound();
1823
1824        let teardown = self.teardown_mut();
1825        teardown.steps += 1;
1826        let finishing_unwind = teardown.on_own_stack && matches!(teardown.step, TeardownStep::FinishUnwinding(_));
1827        if let Some(max_steps) = max_steps {
1828            // Stopping a destructor while a panic unwinds aborts the process, so give the unwind,
1829            // likely that of the panic that stopped the destructor, some more steps.
1830            let bound = if panicking && !finishing_unwind {
1831                max_steps.saturating_mul(2)
1832            } else {
1833                max_steps
1834            };
1835            if teardown.steps > bound {
1836                teardown.exceeded_step_bound = Some(max_steps);
1837                if finishing_unwind {
1838                    return Ok(true);
1839                }
1840                return Err(TeardownStall::ExceededStepBound(max_steps));
1841            }
1842        }
1843        let on_own_stack = teardown.on_own_stack && !panicking;
1844        if on_own_stack && matches!(teardown.step, TeardownStep::Unwind(_)) {
1845            // The task caught the unwind of its stack, which starts again once the task switches
1846            // (see `Coroutine::force_unwind`). It has to run for that.
1847            self.current_mut().unblock();
1848            return Ok(true);
1849        }
1850        if finishing_unwind {
1851            // Once no panic is unwinding, the task has caught its panic.
1852            return Ok(!panicking || !self.current().runnable());
1853        }
1854
1855        // Between steps (dropping the payload of a panic, say), no task runs that could switch out.
1856        let Some(runnable) = self.try_current().map(Task::runnable) else {
1857            return Ok(false);
1858        };
1859        if !runnable && !on_own_stack {
1860            // A destructor blocked the task where it cannot wait. Make it runnable again, so that the
1861            // unwind of the panic that stops the destructor doesn't stall too.
1862            self.current_mut().unblock();
1863            return Err(TeardownStall::Blocked);
1864        }
1865        Ok(on_own_stack && (!runnable || yielded))
1866    }
1867
1868    /// The step bound for the destructors that execution teardown runs, if there is one (see
1869    /// `maybe_yield_in_teardown`).
1870    fn teardown_step_bound(&self) -> Option<usize> {
1871        match self.config.max_steps {
1872            MaxSteps::FailAfter(max_steps) => Some(max_steps),
1873            // A small bound that stops an execution is not meant to fail a test, and teardown has no
1874            // way to stop. So destructors get at least as many steps as an execution does by default.
1875            MaxSteps::ContinueAfter(max_steps) => Some(max_steps.max(DEFAULT_MAX_STEPS)),
1876            MaxSteps::None => None,
1877        }
1878    }
1879
1880    /// Tell the scheduler that the next context switch is an explicit yield requested by the
1881    /// current task. Some schedulers use this as a hint to influence scheduling.
1882    pub fn request_yield() {
1883        Self::with(|state| {
1884            state.has_yielded = true;
1885        });
1886    }
1887
1888    /// Check whether Shuttle's own `Drop` handlers should skip their bookkeeping, and early exit,
1889    /// because the execution has stopped, and so is being abandoned (see `tear_down`).
1890    ///
1891    /// We also stop if we are currently panicking (e.g., perhaps we're unwinding the stack for a
1892    /// panic triggered while someone held a Mutex, and so are executing the Drop handler for
1893    /// MutexGuard). This avoids calling back into the scheduler during a panic, because the state
1894    /// may be poisoned or otherwise invalid.
1895    ///
1896    /// While a finished execution is torn down, though, destructors run as in a running execution,
1897    /// also while a panic unwinds: teardown doesn't call the scheduler. A failed execution is
1898    /// abandoned like a stopped one.
1899    pub fn should_stop() -> bool {
1900        if std::thread::panicking() {
1901            // The state may be borrowed, so don't insist on it.
1902            return Self::try_with(|s| s.stops(true)).unwrap_or(true);
1903        }
1904        Self::with(|s| s.stops(false))
1905    }
1906
1907    /// `should_stop`, given whether the current task is panicking.
1908    pub(crate) fn stops(&self, panicking: bool) -> bool {
1909        match &self.teardown {
1910            Some(teardown) => teardown.kind.abandons(),
1911            None => {
1912                panicking || {
1913                    assert_ne!(self.current_task, ScheduledTask::Finished);
1914                    self.current_task == ScheduledTask::Stopped
1915                }
1916            }
1917        }
1918    }
1919
1920    /// Whether the execution stopped or failed, so that the stacks of its unfinished tasks are
1921    /// leaked rather than unwound (see `tear_down`).
1922    pub(crate) fn execution_stopped() -> bool {
1923        Self::try_with(|state| match &state.teardown {
1924            Some(teardown) => teardown.kind.abandons(),
1925            None => state.current_task == ScheduledTask::Stopped,
1926        })
1927        .unwrap_or(false)
1928    }
1929
1930    /// Generate some diagnostic information used when persisting failures.
1931    ///
1932    /// Because this method may be called from a panic hook, it must not panic.
1933    pub fn failing_task() -> String {
1934        Self::try_with(|state| {
1935            if let Some(task) = state.try_current() {
1936                let name = task.display_name();
1937                match state.teardown.as_ref().map(|teardown| teardown.step) {
1938                    None | Some(TeardownStep::Idle) => name,
1939                    Some(TeardownStep::Static) => format!("{name} (dropping a static at the end of the execution)"),
1940                    Some(TeardownStep::FinishUnwinding(_)) => {
1941                        format!("{name} (unwinding a panic at the end of the execution)")
1942                    }
1943                    Some(_) => format!("{name} (being dropped at the end of the execution)"),
1944                }
1945            } else if let Some(name) = &state.failed_task {
1946                // The task that failed the execution, which has been torn down since.
1947                name.clone()
1948            } else {
1949                "<unknown>".into()
1950            }
1951        })
1952        .unwrap_or_else(|e| format!("Tried to get ExecutionState, but got the following error: {e:?}"))
1953    }
1954
1955    /// Generate a random u64 from the current scheduler and return it.
1956    #[inline]
1957    pub fn next_u64() -> u64 {
1958        Self::with(|state| {
1959            if let Some(teardown) = &mut state.teardown {
1960                // Teardown doesn't extend the schedule (see `Teardown::rng`).
1961                return teardown.next_u64();
1962            }
1963            CurrentSchedule::push_random();
1964            state.scheduler.borrow_mut().next_u64()
1965        })
1966    }
1967
1968    pub fn current(&self) -> &Task {
1969        self.get(self.current_task.id().expect("there is no current task"))
1970    }
1971
1972    pub fn current_mut(&mut self) -> &mut Task {
1973        self.get_mut(self.current_task.id().expect("there is no current task"))
1974    }
1975
1976    pub fn try_current(&self) -> Option<&Task> {
1977        self.try_get(self.current_task.id()?)
1978    }
1979
1980    pub fn get(&self, id: TaskId) -> &Task {
1981        self.try_get(id).unwrap()
1982    }
1983
1984    /// Register a newly created task. Task ids are handed out sequentially as `tasks.len()`, so the
1985    /// new id is always greater than every existing one and `live_tasks` stays sorted.
1986    fn add_task(&mut self, task: Task) {
1987        debug_assert!(self.live_tasks.last().is_none_or(|last| *last < task.id()));
1988        self.live_tasks.push(task.id());
1989        self.tasks.push(task);
1990    }
1991
1992    /// Mark the task as finished and drop it from the set of live tasks.
1993    fn finish_task(&mut self, task_id: TaskId) {
1994        self.get_mut(task_id).finish();
1995        let idx = self
1996            .live_tasks
1997            .binary_search(&task_id)
1998            .expect("finished task must be live");
1999        self.live_tasks.remove(idx);
2000        if self.unwinding_task.is_some_and(|(task, _)| task == task_id) {
2001            self.unwinding_task = None;
2002        }
2003    }
2004
2005    /// Mark the current task as finished and drop it from the set of live tasks.
2006    fn finish_current_task(&mut self) {
2007        self.finish_task(self.current_task.id().unwrap());
2008    }
2009
2010    pub fn get_mut(&mut self, id: TaskId) -> &mut Task {
2011        self.tasks.get_mut(id.0).unwrap()
2012    }
2013
2014    pub fn try_get(&self, id: TaskId) -> Option<&Task> {
2015        self.tasks.get(id.0)
2016    }
2017
2018    pub fn try_get_mut(&mut self, id: TaskId) -> Option<&mut Task> {
2019        self.tasks.get_mut(id.0)
2020    }
2021
2022    /// Whether the execution is being torn down (see `tear_down`).
2023    pub fn in_cleanup(&self) -> bool {
2024        self.teardown.is_some()
2025    }
2026
2027    pub fn context_switches() -> usize {
2028        Self::with(|state| state.context_switches)
2029    }
2030
2031    #[track_caller]
2032    pub fn new_resource_signature(resource_type: ResourceType) -> ResourceSignature {
2033        ExecutionState::with(|s| s.current_mut().signature.new_resource(resource_type))
2034    }
2035
2036    pub fn get_storage<K: Into<StorageKey>, T: 'static>(&self, key: K) -> Option<&T> {
2037        self.storage
2038            .get(key.into())
2039            .map(|result| result.expect("global storage is never destructed"))
2040    }
2041
2042    pub fn init_storage<K: Into<StorageKey>, T: 'static>(&mut self, key: K, value: T) {
2043        self.storage.init(key.into(), value);
2044    }
2045
2046    pub fn get_clock(&self, id: TaskId) -> &VectorClock {
2047        &self.tasks.get(id.0).unwrap().clock
2048    }
2049
2050    pub fn get_clock_mut(&mut self, id: TaskId) -> &mut VectorClock {
2051        &mut self.tasks.get_mut(id.0).unwrap().clock
2052    }
2053
2054    /// Increment the current thread's clock entry and update its clock with the one provided.
2055    pub fn update_clock(&mut self, clock: &VectorClock) {
2056        let task = self.current_mut();
2057        task.clock.increment(task.id);
2058        task.clock.update(clock);
2059    }
2060
2061    /// Increment the current thread's clock and return a shared reference to it
2062    pub fn increment_clock(&mut self) -> &VectorClock {
2063        let task = self.current_mut();
2064        task.clock.increment(task.id);
2065        &task.clock
2066    }
2067
2068    /// Increment the current thread's clock and return a mutable reference to it
2069    pub fn increment_clock_mut(&mut self) -> &mut VectorClock {
2070        let task = self.current_mut();
2071        task.clock.increment(task.id);
2072        &mut task.clock
2073    }
2074
2075    /// Set the number of steps used, with respect to the step bound, to 0 (see
2076    /// `current::reset_step_count`). During execution teardown, these are the steps of destructors
2077    /// (see `maybe_yield_in_teardown`).
2078    pub(crate) fn reset_step_count(&mut self) {
2079        match &mut self.teardown {
2080            Some(teardown) => teardown.steps = 0,
2081            None => self.steps_reset_at = CurrentSchedule::len(),
2082        }
2083    }
2084
2085    /// Returns `true` if the test has exceeded the step bound, and `false` otherwise.
2086    fn is_step_bound_exceeded(&self, max_steps: usize) -> bool {
2087        CurrentSchedule::len() - self.steps_reset_at >= max_steps
2088    }
2089
2090    /// Run the scheduler to choose the next task to run. `has_yielded` should be false if the
2091    /// scheduler is being invoked from within a running task. If scheduling fails, returns an Err
2092    /// with a String describing the failure.
2093    fn schedule(&mut self) -> Result<(), StepError> {
2094        // Don't schedule twice. If `maybe_yield` ran the scheduler, we don't want to run it
2095        // again at the top of `step`.
2096        if self.next_task != ScheduledTask::None {
2097            return Ok(());
2098        }
2099
2100        self.context_switches += 1;
2101
2102        match self.config.max_steps {
2103            MaxSteps::FailAfter(max_steps) => {
2104                if self.is_step_bound_exceeded(max_steps) {
2105                    return Err(StepError::StepBoundExceeded(max_steps));
2106                }
2107            }
2108            MaxSteps::ContinueAfter(max_steps) => {
2109                if self.is_step_bound_exceeded(max_steps) {
2110                    // TODO: We have to set `Stopped` and return `Ok` here, else assertions will fail. This should probably be cleaned up.
2111                    self.next_task = ScheduledTask::Stopped;
2112                    return Ok(());
2113                }
2114            }
2115            MaxSteps::None => {}
2116        }
2117
2118        let mut unfinished_attached = false;
2119        let mut all_runnable_detached = true;
2120        let mut any_runnable = false;
2121
2122        // The loop below only looks at `live_tasks`, so a task missing from that set would silently
2123        // never be scheduled. Check that direction of the invariant here; the loop itself checks the
2124        // other direction (that no finished task is still in the set).
2125        debug_assert!(
2126            self.tasks
2127                .iter()
2128                .filter(|task| !task.finished() && task.runnable())
2129                .all(|task| self.live_tasks.binary_search(&task.id()).is_ok()),
2130            "live_tasks is missing a runnable unfinished task"
2131        );
2132
2133        for &task_id in &self.live_tasks {
2134            let task = &self.tasks[task_id.0];
2135            debug_assert!(!task.finished());
2136            unfinished_attached |= !task.detached;
2137            let is_runnable = task.runnable();
2138            any_runnable |= is_runnable;
2139
2140            if is_runnable {
2141                all_runnable_detached &= task.detached;
2142                self.runnable_tasks.push(task as *const Task);
2143            } else if task.can_spuriously_wakeup() {
2144                // Some blocked tasks can be woken up spuriously, even though the condition the task is
2145                // blocked on hasn't happened yet. We'll add such tasks to the list of runnable tasks, but
2146                // they won't contribute to the check on `any_runnable`; if the only runnable tasks
2147                // are ones that are waiting for a potential spurious wakeup, it should still be treated as
2148                // a deadlock since there's no guarantee that spurious wakeups will ever occur.
2149                self.runnable_tasks.push(task as *const Task);
2150            }
2151        }
2152
2153        // We should finish execution when either
2154        // (1) There are no runnable tasks, or
2155        // (2) All runnable tasks have been detached AND there are no unfinished attached tasks
2156        // If there are some unfinished attached tasks and all runnable tasks are detached, we must
2157        // run some detached task to give them a chance to unblock some unfinished attached task.
2158        if !any_runnable || (!unfinished_attached && all_runnable_detached) {
2159            self.next_task = ScheduledTask::Finished;
2160            return Ok(());
2161        }
2162
2163        let is_yielding = std::mem::replace(&mut self.has_yielded, false);
2164
2165        // Cast the slice of raw pointers to a slice of references in place to provide schedulers with a safe API
2166        //
2167        // SAFETY: This is safe because the tasks themselves are only being accessed through this shared reference by the
2168        // schedulers, and all references are always cleared from the runnable_tasks Vec at the end of this function.
2169        // The transmute itself is safe because *const and & have the same layout, and the pointer is created from a
2170        // reference earlier in this function.
2171        let task_refs = unsafe { std::mem::transmute::<&[*const Task], &[&Task]>(&self.runnable_tasks) };
2172
2173        self.next_task = self
2174            .scheduler
2175            .borrow_mut()
2176            .next_task(task_refs, self.current_task.id(), is_yielding)
2177            .map(ScheduledTask::Some)
2178            .unwrap_or(ScheduledTask::Stopped);
2179
2180        // Tracing this `in_scope` is purely a matter of taste. We do it because
2181        // 1) It is an action taken by the scheduler, and should thus be traced under the scheduler's span
2182        // 2) It creates a visual separation of scheduling decisions and `Task`-induced tracing.
2183        // Note that there is a case to be made for not `in_scope`-ing it, as that makes seeing the context
2184        // of the context switch clearer.
2185        //
2186        // Note also that changing this trace! statement requires changing the test `basic::labels::test_tracing_with_label_fn`
2187        // which relies on this trace reporting the `runnable` tasks.
2188        self.top_level_span.in_scope(|| {
2189            trace!(
2190                i=CurrentSchedule::len(),
2191                next_task=?self.next_task,
2192                runnable=?task_refs.iter().map(|task| task.id()).collect::<SmallVec<[_; DEFAULT_INLINE_TASKS]>>(),
2193                "scheduling decision"
2194            );
2195        });
2196
2197        // If the task chosen by the scheduler is blocked, then it should be one that can be
2198        // spuriously woken up, and we need to unblock it here so that it can execute.
2199        if let Some(tid) = self.next_task.id() {
2200            let task = self.get_mut(tid);
2201            assert!(task.runnable() || task.blocked());
2202            if task.blocked() {
2203                assert!(task.can_spuriously_wakeup());
2204                task.unblock();
2205            }
2206        }
2207
2208        // Retains the capacity of `runnable_tasks` for future calls of `schedule`
2209        self.runnable_tasks.clear();
2210
2211        Ok(())
2212    }
2213
2214    /// Set the next task as the current task
2215    fn advance_to_next_task(&mut self) {
2216        debug_assert_ne!(self.next_task, ScheduledTask::None);
2217        self.current_task = self.next_task.take();
2218
2219        if let ScheduledTask::Some(tid) = self.current_task {
2220            CurrentSchedule::push_task(tid);
2221        }
2222    }
2223
2224    // Sets the `tag` field of the current task.
2225    // Returns the `tag` which was there previously.
2226    #[allow(deprecated)]
2227    pub fn set_tag_for_current_task(tag: Arc<dyn Tag>) -> Option<Arc<dyn Tag>> {
2228        ExecutionState::with(|s| s.current_mut().set_tag(tag))
2229    }
2230
2231    #[allow(deprecated)]
2232    fn get_tag_or_default_for_current_task(&self) -> Option<Arc<dyn Tag>> {
2233        self.try_current().and_then(|current| current.get_tag())
2234    }
2235
2236    #[allow(deprecated)]
2237    pub fn get_tag_for_current_task() -> Option<Arc<dyn Tag>> {
2238        ExecutionState::with(|s| s.get_tag_or_default_for_current_task())
2239    }
2240
2241    #[allow(deprecated)]
2242    pub fn set_tag_for_task(task: TaskId, tag: Arc<dyn Tag>) -> Option<Arc<dyn Tag>> {
2243        ExecutionState::with(|s| s.get_mut(task).set_tag(tag))
2244    }
2245}
2246
2247#[cfg(debug_assertions)]
2248impl Drop for ExecutionState {
2249    fn drop(&mut self) {
2250        assert!(self.has_cleaned_up || std::thread::panicking());
2251    }
2252}
2253
2254/// Whether the running task's default dispatcher might not be the execution's.
2255fn task_may_have_own_default(state: &ExecutionState) -> bool {
2256    use tracing::level_filters::LevelFilter;
2257
2258    // No dispatcher enables anything, so whichever one is the default makes no difference. (If that
2259    // changes while the task is switched out, other tasks see its default, as before defaults were
2260    // parked.)
2261    if LevelFilter::current() == LevelFilter::OFF {
2262        return false;
2263    }
2264    tracing::dispatcher::get_default(|current| {
2265        if let Some(global) = state.top_level_dispatch_global_ptr {
2266            // `get_default` hands out the global default from where it lives, and anything else from
2267            // this thread's scoped default. So if it hands out the global default, the task has no
2268            // default of its own. Parking otherwise is sometimes unnecessary but always safe,
2269            // including if `tracing` ever handed out the global default from elsewhere.
2270            !std::ptr::eq(current, global)
2271        } else if current.is::<NoSubscriber>() {
2272            // Nothing is recorded either way if the execution's default is `Dispatch::none()` too.
2273            !state.top_level_dispatch.is::<NoSubscriber>()
2274        } else {
2275            // A task can only have a default of its own through a scoped default. `get_default` hands
2276            // nested calls `Dispatch::none()` only while some thread has one; otherwise they get the
2277            // global default, which is then `current` and everyone's default.
2278            tracing::dispatcher::get_default(|nested| nested.is::<NoSubscriber>())
2279        }
2280    })
2281}