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

starnix_core/task/
kernel_threads.rs
