starnix_core/task/
kernel_threads.rs

1// Copyright 2023 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use crate::execution::create_kernel_thread;
6use crate::task::dynamic_thread_spawner::DynamicThreadSpawner;
7use crate::task::{CurrentTask, DelayedReleaser, Kernel, Task, ThreadGroup};
8use fragile::Fragile;
9use fuchsia_async as fasync;
10use fuchsia_sync::Completion;
11use pin_project::pin_project;
12use scopeguard::ScopeGuard;
13
14use starnix_task_command::TaskCommand;
15
16use starnix_uapi::errors::Errno;
17use starnix_uapi::{errno, error};
18use std::future::Future;
19use std::pin::Pin;
20use std::sync::{Arc, OnceLock, Weak};
21use std::task::{Context, Poll};
22
23/// The threads that the kernel runs internally.
24///
25/// These threads run in the main starnix process and outlive any specific userspace process.
26pub struct KernelThreads {
27    /// The main starnix process. This process is used to create new processes when using the
28    /// restricted executor.
29    pub starnix_process: zx::Process,
30
31    /// A handle to the async executor running in `starnix_process`.
32    ///
33    /// You can spawn tasks on this executor using `spawn_future`. However, those task must not
34    /// block. If you need to block, you can spawn a worker thread using `spawner`.
35    ehandle: fasync::EHandle,
36
37    /// The thread pool to spawn blocking calls to.
38    spawner: OnceLock<DynamicThreadSpawner>,
39
40    /// Information about the main system task that is bound to the kernel main thread.
41    system_task: OnceLock<SystemTask>,
42
43    /// A weak reference to the kernel owning this struct.
44    kernel: Weak<Kernel>,
45
46    /// The RCU advancer thread running callbacks in the background.
47    rcu_advancer: OnceLock<RcuAdvancer>,
48}
49
50/// The minimum time between running rcu callbacks. This allows us to limit wake ups and batch
51/// callbacks together.
52const RCU_RATE_LIMIT: std::time::Duration = std::time::Duration::from_millis(10);
53
54struct RcuAdvancer {
55    stop: Arc<Completion>,
56    thread: Option<std::thread::JoinHandle<()>>,
57}
58
59impl Drop for RcuAdvancer {
60    fn drop(&mut self) {
61        self.stop.signal();
62        fuchsia_rcu::rcu_advancer_wake();
63        if let Some(thread) = self.thread.take() {
64            let _ = thread.join();
65        }
66    }
67}
68
69impl KernelThreads {
70    /// Create a KernelThreads object for the given Kernel.
71    ///
72    /// Must be called in the initial Starnix process on a thread with an async executor. This
73    /// function captures the async executor for this thread for use with spawned futures.
74    ///
75    /// Used during kernel boot.
76    pub fn new(kernel: Weak<Kernel>) -> Self {
77        KernelThreads {
78            starnix_process: fuchsia_runtime::process_self()
79                .duplicate_handle(zx::Rights::SAME_RIGHTS)
80                .expect("Failed to duplicate process self"),
81            ehandle: fasync::EHandle::local(),
82            spawner: Default::default(),
83            system_task: Default::default(),
84            kernel,
85            rcu_advancer: Default::default(),
86        }
87    }
88
89    /// Initialize this object with the system task that will be used for spawned threads.
90    ///
91    /// This function must be called before this object is used to spawn threads.
92    pub fn init(&self, system_task: CurrentTask) -> Result<(), Errno> {
93        self.system_task.set(SystemTask::new(system_task)).map_err(|_| errno!(EEXIST))?;
94        self.spawner
95            .set(DynamicThreadSpawner::new(2, self.system_task().weak_task(), "kthreadd/init"))
96            .map_err(|_| errno!(EEXIST))?;
97
98        let stop = Arc::new(Completion::new());
99        let stop_clone = Arc::clone(&stop);
100        let thread = std::thread::Builder::new()
101            .name("starnix-rcu".to_string())
102            .spawn(move || {
103                let _rcu_registration = fuchsia_rcu::register_thread();
104
105                while !stop_clone.is_signaled() {
106                    fuchsia_rcu::rcu_advancer_wait_for_work();
107                    while !stop_clone.is_signaled() {
108                        let start = std::time::Instant::now();
109                        if !fuchsia_rcu::rcu_run_callbacks() {
110                            break;
111                        }
112                        let elapsed = start.elapsed();
113                        if elapsed < RCU_RATE_LIMIT {
114                            if stop_clone.wait_for(RCU_RATE_LIMIT - elapsed) {
115                                break;
116                            }
117                        }
118                    }
119                }
120                // Run remaining ready callbacks before exiting.
121                fuchsia_rcu::rcu_run_callbacks();
122            })
123            .expect("failed to spawn rcu advancer thread");
124
125        self.rcu_advancer
126            .set(RcuAdvancer { stop, thread: Some(thread) })
127            .map_err(|_| errno!(EEXIST))?;
128
129        Ok(())
130    }
131
132    /// Spawn an async task in the main async executor to await the given future.
133    ///
134    /// Use this function to run async tasks in the background. These tasks cannot block or else
135    /// they will starve the main async executor.
136    ///
137    /// Prefer this function to `spawn` for non-blocking work.
138    pub fn spawn_future(
139        &self,
140        future: impl AsyncFnOnce() -> () + Send + 'static,
141        name: &'static str,
142    ) {
143        self.ehandle.spawn_detached(WrappedMainFuture::new(
144            self.kernel.clone(),
145            async move { fasync::Task::local(future()).await },
146            name,
147        ));
148    }
149
150    /// The dynamic thread spawner used to spawn threads.
151    ///
152    /// To spawn a thread in this thread pool, use `spawn()`.
153    pub fn spawner(&self) -> &DynamicThreadSpawner {
154        self.spawner.get().as_ref().unwrap()
155    }
156
157    /// Access the `CurrentTask` for the kernel main thread.
158    ///
159    /// This function can only be called from the kernel main thread itself.
160    pub fn system_task(&self) -> &CurrentTask {
161        self.system_task.get().expect("KernelThreads::init must be called").system_task.get()
162    }
163
164    /// Access the `ThreadGroup` for the system tasks.
165    ///
166    /// This function can be safely called from anywhere as soon as `KernelThreads::init` has been
167    /// called.
168    pub fn system_thread_group(&self) -> Arc<ThreadGroup> {
169        self.system_task
170            .get()
171            .expect("KernelThreads::init must be called")
172            .system_thread_group
173            .upgrade()
174            .expect("System task must be still alive")
175    }
176}
177
178impl Drop for KernelThreads {
179    // TODO: Replace with .release. This is not actually safe, since locks
180    // may be held elsewhere on this thread.
181    fn drop(&mut self) {
182        if let Some(system_task) = self.system_task.take() {
183            system_task.system_task.into_inner().release(());
184        }
185    }
186}
187
188/// Create a new system task, register it on the thread and run the given closure with it.
189
190pub fn with_new_current_task<F, R>(
191    system_task: &Weak<Task>,
192    task_name: String,
193    f: F,
194) -> Result<R, Errno>
195where
196    F: FnOnce(&CurrentTask) -> R,
197{
198    let current_task = {
199        let Some(system_task) = system_task.upgrade() else {
200            return error!(ESRCH);
201        };
202        create_kernel_thread(&system_task, TaskCommand::new(task_name.as_bytes())).unwrap()
203    };
204    let result = f(&current_task);
205    current_task.release(());
206
207    // Ensure that no releasables are registered after this point as we unwind the stack.
208    DelayedReleaser::finalize();
209
210    Ok(result)
211}
212
213struct SystemTask {
214    /// The system task is bound to the kernel main thread. `Fragile` ensures a runtime crash if it
215    /// is accessed from any other thread.
216    system_task: Fragile<CurrentTask>,
217
218    /// The system `ThreadGroup` is accessible from everywhere.
219    system_thread_group: Weak<ThreadGroup>,
220}
221
222impl SystemTask {
223    fn new(system_task: CurrentTask) -> Self {
224        let system_thread_group = Arc::downgrade(&system_task.thread_group());
225        Self { system_task: system_task.into(), system_thread_group }
226    }
227}
228
229// The order is important here. Rust will drop fields in declaration order and we want
230// the future to be dropped before the ScopeGuard runs.
231#[pin_project]
232pub(crate) struct WrappedFuture<F, C: Clone> {
233    #[pin]
234    fut: F,
235    cleaner: fn(C),
236    context: ScopeGuard<C, fn(C)>,
237    name: &'static str,
238}
239
240impl<F, C: Clone> WrappedFuture<F, C> {
241    pub(crate) fn new_with_cleaner(context: C, cleaner: fn(C), fut: F, name: &'static str) -> Self {
242        // We need the ScopeGuard in case the future queues releasers when dropped.
243        Self { fut, cleaner, context: ScopeGuard::with_strategy(context, cleaner), name }
244    }
245}
246
247impl<F: Future, C: Clone> Future for WrappedFuture<F, C> {
248    type Output = F::Output;
249
250    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
251        let this = self.project();
252        fuchsia_trace::duration!(starnix_logging::CATEGORY_STARNIX, &*this.name);
253        let result = this.fut.poll(cx);
254
255        (this.cleaner)(this.context.clone());
256        result
257    }
258}
259
260type WrappedMainFuture<F> = WrappedFuture<F, Weak<Kernel>>;
261
262impl<F> WrappedMainFuture<F> {
263    fn new(kernel: Weak<Kernel>, fut: F, name: &'static str) -> Self {
264        Self::new_with_cleaner(kernel, trigger_delayed_releaser, fut, name)
265    }
266}
267
268fn trigger_delayed_releaser(kernel: Weak<Kernel>) {
269    if let Some(kernel) = kernel.upgrade() {
270        if let Some(system_task) = kernel.kthreads.system_task.get() {
271            system_task.system_task.get().trigger_delayed_releaser();
272        }
273    }
274}
starnix_core/task/kernel_threads.rs

starnix_core/task/
kernel_threads.rs
