ax_task/sched/system/task_system/delivery/
admission.rs

1//! Admission under the owning scheduler transaction.
2
3use super::*;
4
5impl TaskSystem {
6    /// Enqueues a ready thread on an affinity-compatible owner CPU.
7    pub fn enqueue(&self, mut cpu: Pin<&mut CpuLocal>, thread: ThreadId) -> Result<(), TaskError> {
8        self.ensure_owner_cpu_context(&cpu)?;
9        let core = {
10            let state = self.state.lock();
11            state.ensure_cpu_online(&cpu)?;
12            Arc::clone(&state.thread_record(thread)?.core)
13        };
14        self.enqueue_owner_thread(cpu.as_mut(), core, EnqueueReason::Wake)?;
15        self.program_local_timer(cpu.as_mut(), SchedulerDeadlineDerivationSource::Enqueue)
16    }
17
18    /// Admits a new thread and commits its placement on an allowed active CPU.
19    ///
20    /// Rejected admission does not change lifecycle or placement. Success
21    /// guarantees either local
22    /// runqueue admission or an owned remote activation delivery. There is no
23    /// public state-only runnable transition to complete in a second call.
24    ///
25    /// Ordinary fair work is placed on the least-loaded allowed CPU, including
26    /// its current non-idle dispatch and migrations not yet consumed by the
27    /// destination owner. Other classes preserve owner-local placement unless
28    /// affinity requires a transfer. Remote placement uses the owner-only
29    /// owner-control inbox and never mutates another CPU's runqueue.
30    ///
31    /// # Errors
32    ///
33    /// Returns an error when the source CPU is offline, the thread is not a
34    /// new unqueued thread, no allowed CPU is online, or remote delivery
35    /// cannot be reserved. Failures after admission are runtime invariants.
36    pub fn start_thread(
37        &self,
38        mut cpu: Pin<&mut CpuLocal>,
39        thread: ThreadId,
40    ) -> Result<(), TaskError> {
41        self.ensure_owner_cpu_context(&cpu)?;
42        let handle = {
43            let state = self.state.lock();
44            state.ensure_cpu_online(&cpu)?;
45            let record = state.thread_record(thread)?;
46            // Managed entries require their owning publication token. The raw
47            // integration primitive cannot bypass an OS identity transaction.
48            if record.core.execution.is_some() {
49                return Err(TaskError::NotReady);
50            }
51            ThreadHandle::from_core(Arc::clone(&record.core))
52        };
53        self.stage_new_thread(&handle)?;
54        self.activate_staged_thread(cpu.as_mut(), &handle);
55        Ok(())
56    }
57
58    /// Reserves the first owner delivery while the thread is still TASK_NEW.
59    pub(crate) fn stage_new_thread(&self, handle: &ThreadHandle) -> Result<(), TaskError> {
60        // SAFETY: task-context preparation runs on an installed runtime CPU.
61        let source = CpuId::new(unsafe { task_runtime::current_cpu_id() }.as_u32());
62        let mut state = self.state.lock();
63        let record = state.thread_record(handle.id())?;
64        let sched = record.sched.lock();
65        if sched.lifecycle.state() != ThreadState::New || record.activation.is_some() {
66            return Err(TaskError::NotReady);
67        }
68        let active = record.core.sched().active(&sched);
69        let target = if matches!(active.policy(), SchedulePolicy::Fair { .. }) {
70            state.select_initial_fair_cpu(&sched.affinity.affinity, Some(source))
71        } else {
72            self.select_priority_cpu(
73                active.policy(),
74                Some(active.entity()),
75                &sched.affinity.affinity,
76                Some(source),
77                None,
78            )
79        }
80        .ok_or(TaskError::InvalidConfiguration)?;
81        let delivery = self.prepare_owner_migration(&record.core, source, target)?;
82        drop(active);
83        drop(sched);
84        state.thread_record_mut(handle.id())?.activation = Some(delivery);
85        Ok(())
86    }
87
88    /// Removes a ready thread from its owner run queue for migration or update.
89    pub fn dequeue(&self, cpu: Pin<&mut CpuLocal>, thread: ThreadId) -> Result<(), TaskError> {
90        self.ensure_owner_cpu_context(&cpu)?;
91        let state = self.state.lock();
92        state.ensure_cpu_online(&cpu)?;
93        let record = state.thread_record(thread)?;
94        let mut sched = record.sched.lock();
95        let remote = Arc::clone(cpu.remote());
96        let mut transaction = OwnerRqTxn::begin(self, &remote);
97        if transaction.current_thread() == Some(thread)
98            || transaction.scheduling_entity(thread).is_none()
99        {
100            transaction.commit();
101            return Err(TaskError::NotReady);
102        }
103        let queued = transaction.deactivate_task(thread);
104        record
105            .core
106            .sched()
107            .install_active(&mut sched, queued.into_active());
108        sched.placement.deactivate(cpu.owner());
109        transaction.commit();
110        drop(sched);
111        drop(state);
112        Ok(())
113    }
114}
ax_task/sched/system/task_system/delivery/admission.rs

ax_task/sched/system/task_system/delivery/
admission.rs
