ax_task/runtime/service/
reclaim.rs

1use alloc::string::String;
2use core::pin::Pin;
3
4use crate::{
5    executor::CoroutineHeader,
6    runtime::{
7        TaskSystem,
8        context::{runtime_task_system, validate_task_context},
9        delivery::inbox::PublishResult,
10        switch::dispatch::yield_current_cpu,
11        task_runtime,
12    },
13    sync::irq::IrqWaitToken,
14    thread::{TaskError, ThreadBuilder, current::current_thread_handle},
15};
16
17pub(crate) fn publish_deferred_coroutine_reclaim(header: Pin<&'static CoroutineHeader>) {
18    let system = runtime_task_system().unwrap_or_else(|_| {
19        task_runtime::fatal_invariant(0x4558_0008, header.address());
20    });
21    match system.publish_deferred_coroutine_reclaim(header) {
22        PublishResult::Published => {}
23        PublishResult::AlreadyPending | PublishResult::WrongKind => {
24            task_runtime::fatal_invariant(0x4558_0004, header.address());
25        }
26    }
27}
28
29/// Notifies the resource reaper after a runtime drops the last active-mm lease.
30///
31/// This allocation-free publication is valid from the IRQ-off switch tail and
32/// carries no OS callback or address-space pointer.
33pub fn notify_address_space_reclaim() {
34    if let Ok(system) = runtime_task_system() {
35        system.publish_resource_release_ready();
36    }
37}
38
39/// Creates the shutdown-lifetime service thread for callbacks and reclamation.
40///
41/// A runtime must call this once after publishing its primary scheduler CPU and
42/// before allowing ordinary application threads to exit. The service is the
43/// only consumer of deferred Deadline, exit, and destruction work.
44pub fn start_deferred_task_work_service() -> Result<(), TaskError> {
45    let system = runtime_task_system()?;
46    system.begin_task_work_worker_install()?;
47    let worker =
48        match ThreadBuilder::new(String::from("ax-task-reaper")).spawn(task_work_service_entry) {
49            Ok(worker) => worker,
50            Err(error) => {
51                system.cancel_task_work_worker_install();
52                return Err(error);
53            }
54        };
55    worker.detach();
56    Ok(())
57}
58
59fn task_work_service_entry() {
60    if task_work_service_loop().is_err() {
61        task_runtime::fatal_invariant(0x4558_0030, 0);
62    }
63}
64
65fn task_work_service_loop() -> Result<(), TaskError> {
66    const BATCH_LIMIT: usize = 64;
67
68    let system = runtime_task_system()?;
69    let doorbell = system.task_work_doorbell();
70    let wake_owner = current_thread_handle()?.wake_handle();
71    let waiter = crate::sync::irq::worker::IrqWorkerWaiter::new(wake_owner);
72    system.finish_task_work_worker_install();
73
74    loop {
75        if let Some(claim) = doorbell.claim_pending() {
76            debug_assert_ne!(claim.epoch(), 0);
77        }
78        let batch = service_task_work_pass(system, &doorbell, BATCH_LIMIT)?;
79        let pending_after_pass = doorbell.claim_pending().is_some();
80        match task_work_service_action(batch, pending_after_pass, BATCH_LIMIT) {
81            TaskWorkServiceAction::Yield => {
82                yield_current_cpu()?;
83                continue;
84            }
85            TaskWorkServiceAction::Wait => {
86                waiter.wait(doorbell.event())?;
87            }
88        }
89    }
90}
91
92#[derive(Clone, Copy, Debug, Eq, PartialEq)]
93pub(crate) enum TaskWorkServiceAction {
94    Yield,
95    Wait,
96}
97
98pub(crate) fn task_work_service_action(
99    batch: Option<crate::runtime::service::DeferredTaskWorkBatch>,
100    pending_after_pass: bool,
101    limit: usize,
102) -> TaskWorkServiceAction {
103    let action = match batch {
104        None => TaskWorkServiceAction::Yield,
105        Some(batch) if batch.saturated(limit) || pending_after_pass => TaskWorkServiceAction::Yield,
106        Some(_) => TaskWorkServiceAction::Wait,
107    };
108    #[cfg(feature = "qperf-metrics")]
109    match action {
110        TaskWorkServiceAction::Yield => {
111            crate::diagnostics::counters::record_task_work_worker_yield()
112        }
113        TaskWorkServiceAction::Wait => crate::diagnostics::counters::record_task_work_worker_wait(),
114    }
115    action
116}
117
118pub(crate) fn service_task_work_pass(
119    system: &TaskSystem,
120    doorbell: &crate::runtime::delivery::work::TaskWorkDoorbell,
121    limit: usize,
122) -> Result<Option<crate::runtime::service::DeferredTaskWorkBatch>, TaskError> {
123    match system.service_deferred_task_work(limit) {
124        Ok(batch) => {
125            #[cfg(feature = "qperf-metrics")]
126            crate::diagnostics::counters::record_task_work_worker_pass(batch.processed());
127            Ok(Some(batch))
128        }
129        Err(TaskError::ThreadBusy) => {
130            doorbell.reassert_pending();
131            Ok(None)
132        }
133        Err(error) => Err(error),
134    }
135}
136
137/// Detaches one IRQ waiter and waits out any in-flight notification claim.
138///
139/// Linux PREEMPT_RT owns hard irq-work completion through the BUSY claim: the
140/// executor clears it as its final access, and `irq_work_sync()` on the hard
141/// path busy-waits for that clear. The IRQ cell follows the same rule: the
142/// notifier's `Notifying` claim covers every access to the registration and
143/// its wake payload and publishes `Detached` last, so this quiesce step only
144/// waits for that publication before storage may be reused.
145///
146/// Callers must invoke this in task context before reusing or releasing
147/// storage reachable through the matching [`IrqWaitRegistration`]. Hard-IRQ
148/// teardown must instead move the token to a task-context worker.
149pub fn quiesce_irq_wait(token: IrqWaitToken<'_>) -> Result<(), TaskError> {
150    validate_task_context()?;
151    token.detach().finish();
152    Ok(())
153}
ax_task/runtime/service/reclaim.rs

ax_task/runtime/service/
reclaim.rs
