ax_task/sched/system/task_system/
model.rs

1//! Scheduler ownership model and bounded work accounting.
2
3use super::*;
4
5/// Failure returned by [`TaskSystem::reap_thread_handle`].
6///
7/// A failed registry transition returns ownership of the strong handle, keeping
8/// the registry generation pinned while the caller handles the error.
9#[derive(Debug, thiserror::Error)]
10#[error("{error}")]
11pub struct OwnedThreadReapError {
12    error: TaskError,
13    handle: ThreadHandle,
14}
15
16impl OwnedThreadReapError {
17    pub(super) const fn new(error: TaskError, handle: ThreadHandle) -> Self {
18        Self { error, handle }
19    }
20
21    /// Returns the underlying scheduler error.
22    pub const fn task_error(&self) -> TaskError {
23        self.error
24    }
25
26    /// Returns the still-valid handle.
27    pub fn into_retry_handle(self) -> ThreadHandle {
28        self.handle
29    }
30}
31
32/// One bounded pass performed by the dedicated task-work service thread.
33#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
34pub struct DeferredTaskWorkBatch {
35    pub(super) deadline_events: usize,
36    pub(super) deadline_callbacks: usize,
37    pub(super) scheduler_tick_events: usize,
38    pub(super) scheduler_tick_callbacks: usize,
39    pub(super) exit_callbacks: usize,
40    pub(super) cancellation_events: usize,
41    pub(super) reaped_threads: usize,
42    pub(super) execution_reclaims: usize,
43    pub(super) coroutine_reclaims: usize,
44    pub(super) address_space_reclaims: usize,
45}
46
47impl DeferredTaskWorkBatch {
48    /// Returns the number of queue entries or resources consumed by this pass.
49    pub const fn processed(self) -> usize {
50        self.deadline_events
51            + self.scheduler_tick_events
52            + self.exit_callbacks
53            + self.cancellation_events
54            + self.reaped_threads
55            + self.execution_reclaims
56            + self.coroutine_reclaims
57            + self.address_space_reclaims
58    }
59
60    /// Returns the number of Deadline extension callbacks invoked.
61    pub const fn deadline_callbacks(self) -> usize {
62        self.deadline_callbacks
63    }
64
65    /// Returns consumed creation-cancellation requests, including deferred retries.
66    pub const fn cancellation_events(self) -> usize {
67        self.cancellation_events
68    }
69
70    /// Returns the number of scheduler-tick extension callbacks invoked.
71    pub const fn scheduler_tick_callbacks(self) -> usize {
72        self.scheduler_tick_callbacks
73    }
74
75    /// Returns zero-reference coroutine allocations reclaimed after hard IRQ.
76    pub const fn coroutine_reclaims(self) -> usize {
77        self.coroutine_reclaims
78    }
79
80    /// Returns active-mm ownership tokens whose final CPU lease was released.
81    pub const fn address_space_reclaims(self) -> usize {
82        self.address_space_reclaims
83    }
84
85    /// Returns whether another pass should run before the worker parks.
86    pub const fn made_progress(self) -> bool {
87        self.processed() != 0
88    }
89
90    /// Returns whether this pass consumed the complete shared caller budget.
91    pub const fn saturated(self, limit: usize) -> bool {
92        let capped_limit = if limit < crate::runtime::config::DEFAULT_BATCH_LIMIT {
93            limit
94        } else {
95            crate::runtime::config::DEFAULT_BATCH_LIMIT
96        };
97        capped_limit != 0 && self.processed() == capped_limit
98    }
99}
100
101/// Complete OS-independent scheduler instance.
102///
103/// No instance is stored globally. A runtime owns one pinned `TaskSystem` and
104/// passes explicit object references to the scheduler or exposes them through its
105/// trait-FFI facade.
106///
107/// IRQ and remote producers wake through
108/// [`ThreadWakeHandle::wake`](crate::thread::ThreadWakeHandle::wake). The wake path
109/// serializes thread state, selects an online destination, and activates the
110/// thread under that destination's IRQ-safe runqueue lock. Like Linux
111/// PREEMPT_RT with `TTWU_QUEUE` disabled, a remote waker waits for switch tail's
112/// `on_cpu` release before completing that direct activation.
113#[derive(Debug)]
114pub struct TaskSystem {
115    pub(super) config: TaskSystemConfig,
116    pub(super) cpu_remotes: Vec<Arc<CpuRemote>>,
117    // Cold-path order is registry/PI/admission -> root domain -> thread cell.
118    // Wake and placement hot paths lock thread state before the target runqueue.
119    pub(super) state: PreemptTicketLock<TaskSystemState>,
120    pub(super) root_domain: RootDomain,
121    pub(super) deferred_coroutine_reclaims: SchedulerInbox,
122    pub(super) deferred_thread_cancellations: SchedulerInbox,
123    pub(super) deferred_deadline_callbacks: SchedulerInbox,
124    pub(super) deferred_scheduler_ticks: SchedulerInbox,
125    pub(super) task_work: Arc<TaskWorkDoorbell>,
126}
127
128#[derive(Clone, Copy, Debug, Eq, PartialEq)]
129pub(super) enum BalanceReason {
130    RtDeadlinePush,
131    IdlePull,
132    FairPeriodic,
133}
134
135/// Result of one opportunistic owner-to-owner balance attempt.
136///
137/// `Retry` means the transfer transaction observed a concurrent affinity,
138/// hotplug, or publication change and restored every local ownership record.
139/// It is not a failure of the already committed local scheduling decision.
140#[derive(Clone, Copy, Debug, Eq, PartialEq)]
141pub(super) enum BalanceTransferOutcome {
142    Migrated(ThreadId),
143    NoCandidate,
144    Retry,
145}
146
147impl BalanceTransferOutcome {
148    pub(super) const fn migrated(self) -> Option<ThreadId> {
149        match self {
150            Self::Migrated(thread) => Some(thread),
151            Self::NoCandidate | Self::Retry => None,
152        }
153    }
154}
155
156#[derive(Clone, Copy, Debug, Eq, PartialEq)]
157pub(super) enum FairBalanceResult {
158    Migrated(ThreadId),
159    Balanced,
160    Constrained,
161}
162
163impl FairBalanceResult {
164    pub(super) const fn migrated(self) -> Option<ThreadId> {
165        match self {
166            Self::Migrated(thread) => Some(thread),
167            Self::Balanced | Self::Constrained => None,
168        }
169    }
170}
171
172pub(super) const FAIR_BALANCE_BALANCED_BACKOFF_FACTOR: u64 = 2;
173pub(super) const FAIR_BALANCE_CONSTRAINED_BACKOFF_FACTOR: u64 = 64;
174
175#[derive(Clone, Copy, Debug, Eq, PartialEq)]
176pub(super) enum DeferredTaskWorkClass {
177    Cancellation,
178    Deadline,
179    SchedulerTick,
180    Exit,
181    Reap,
182    Reclaim,
183}
184
185impl DeferredTaskWorkClass {
186    pub(super) const COUNT: usize = 6;
187
188    pub(super) const fn next(self) -> Self {
189        match self {
190            Self::Cancellation => Self::Deadline,
191            Self::Deadline => Self::SchedulerTick,
192            Self::SchedulerTick => Self::Exit,
193            Self::Exit => Self::Reap,
194            Self::Reap => Self::Reclaim,
195            Self::Reclaim => Self::Cancellation,
196        }
197    }
198}
199
200/// Owns the unprocessed suffix of one already-detached owner inbox batch.
201///
202/// `SchedulerInbox::drain` releases every intrusive node before the caller
203/// interprets any message. If processing one message fails, this guard still
204/// consumes every later raw `Arc` payload and its scheduler-delivery lease.
205pub(super) struct DetachedOwnerMessageBatch<'batch> {
206    pub(super) messages: &'batch [InboxMessage],
207    pub(super) next: usize,
208}
209
210impl<'batch> DetachedOwnerMessageBatch<'batch> {
211    pub(super) const fn new(messages: &'batch [InboxMessage]) -> Self {
212        Self { messages, next: 0 }
213    }
214
215    pub(super) fn next(&mut self) -> Option<InboxMessage> {
216        let message = self.messages.get(self.next).copied()?;
217        self.next += 1;
218        Some(message)
219    }
220
221    pub(super) fn release(message: InboxMessage) {
222        if message.payload() == 0 {
223            return;
224        }
225        let core = unsafe {
226            // SAFETY: every non-zero owner message transfers exactly one
227            // `ThreadCore` Arc count into its payload. This detached batch owns
228            // that count even when normal message processing aborts early.
229            Arc::from_raw(ptr::with_exposed_provenance::<ThreadCore>(
230                message.payload(),
231            ))
232        };
233        let _delivery = core.accept_scheduler_inbox_delivery();
234    }
235}
236
237impl Drop for DetachedOwnerMessageBatch<'_> {
238    fn drop(&mut self) {
239        for &message in &self.messages[self.next..] {
240            Self::release(message);
241        }
242    }
243}
244
245#[derive(Clone, Copy, Debug, Eq, PartialEq)]
246pub(super) struct FairPolicyPlacement {
247    pub(super) source_virtual_time: u64,
248    pub(super) destination_virtual_time: u64,
249}
ax_task/sched/system/task_system/model.rs

ax_task/sched/system/task_system/
model.rs
