starnix_core/task/
thread_lockup_detector.rs

1// Copyright 2026 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
5//! This module implements a lockup detector for Starnix kernel threads.
6//! It tracks the start time of operations and reports threads that run for too long
7//! without pausing or stopping the operation.
8//!
9//! It uses a global registry to track active operations across all threads.
10
11use pin_project::pin_project;
12use starnix_sync::{LockDepRwLock, ThreadLockupDetectorRegistryLock};
13use std::borrow::Borrow;
14use std::cell::RefCell;
15use std::collections::HashSet;
16use std::sync::LazyLock;
17use std::sync::atomic::{AtomicU8, AtomicU64, AtomicUsize, Ordering};
18
19#[derive(Default)]
20pub struct ThreadLockupDetector;
21
22/// Thread-local state that registers the thread in the global registry on creation
23/// and removes it on drop.
24struct ThreadState {
25    /// Pointer to the atomic u64 used to store the start time of the current operation.
26    /// This is boxed to ensure its address remains stable while registered.
27    atomic: Box<AtomicU64>,
28    /// The KOID of the thread, used as the key for removal in `Drop`.
29    koid: zx::Koid,
30}
31
32impl ThreadState {
33    /// Creates a new `ThreadState`, registering the current thread in the global `REGISTRY`.
34    fn new() -> Self {
35        let handle = fuchsia_runtime::with_thread_self(|thread| thread.raw_handle());
36        let koid = fuchsia_runtime::with_thread_self(|thread| thread.koid()).unwrap();
37        let atomic = Box::new(AtomicU64::new(0));
38        let ptr = &*atomic as *const AtomicU64;
39
40        let mut rcu_nesting_level = std::ptr::null();
41        let mut rcu_counter_index = std::ptr::null();
42        fuchsia_rcu::with_thread_block_counters(|nesting_ptr, counter_ptr| {
43            rcu_nesting_level = nesting_ptr;
44            rcu_counter_index = counter_ptr;
45        });
46
47        let registered = RegisteredThread {
48            // SAFETY: The handle is valid as long as the thread is registered.
49            thread: unsafe { zx::Unowned::from_raw_handle(handle) },
50            koid,
51            atomic: ptr,
52            rcu_nesting_level,
53            rcu_counter_index,
54        };
55        REGISTRY.write().insert(registered);
56        Self { atomic, koid }
57    }
58}
59
60impl Drop for ThreadState {
61    /// Removes the thread from the global `REGISTRY` when the thread exits.
62    fn drop(&mut self) {
63        REGISTRY.write().remove(&self.koid);
64    }
65}
66
67thread_local! {
68    static THREAD_STATE: RefCell<Option<ThreadState>> = const { RefCell::new(None) };
69}
70
71/// The information stored in the global registry for each tracked thread.
72#[derive(Clone)]
73struct RegisteredThread {
74    /// An unowned handle to the thread, used for inspection.
75    thread: zx::Unowned<'static, zx::Thread>,
76    /// The KOID of the thread.
77    koid: zx::Koid,
78    /// Pointer to the atomic u64 in the thread's `ThreadState`.
79    atomic: *const AtomicU64,
80    /// Pointer to the RCU nesting level.
81    rcu_nesting_level: *const AtomicUsize,
82    /// Pointer to the RCU counter index.
83    rcu_counter_index: *const AtomicU8,
84}
85
86// We only hash and compare by `koid` to allow lookup and removal by `koid`
87// in the `HashSet`.
88impl std::hash::Hash for RegisteredThread {
89    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
90        self.koid.hash(state);
91    }
92}
93
94impl PartialEq for RegisteredThread {
95    fn eq(&self, other: &Self) -> bool {
96        self.koid == other.koid
97    }
98}
99
100impl Eq for RegisteredThread {}
101
102impl Borrow<zx::Koid> for RegisteredThread {
103    fn borrow(&self) -> &zx::Koid {
104        &self.koid
105    }
106}
107
108// SAFETY: Access to the pointers in the global REGISTRY is protected by a LockDepRwLock,
109// ensuring that a thread cannot free its data while another thread is reading it.
110unsafe impl Send for RegisteredThread {}
111// SAFETY: Same as above.
112unsafe impl Sync for RegisteredThread {}
113
114#[derive(Clone, Debug)]
115pub struct ThreadLockupInfo {
116    pub thread: zx::Unowned<'static, zx::Thread>,
117    pub koid: zx::Koid,
118    pub start_time: zx::MonotonicInstant,
119}
120
121/// Global registry of all tracked threads.
122static REGISTRY: LazyLock<
123    LockDepRwLock<HashSet<RegisteredThread>, ThreadLockupDetectorRegistryLock>,
124> = LazyLock::new(|| Default::default());
125
126impl ThreadLockupDetector {
127    /// Starts an operation by storing the current time in the thread-local atomic.
128    fn start_operation() {
129        THREAD_STATE.with(|state| {
130            let mut state = state.borrow_mut();
131            let state = state.get_or_insert_with(|| ThreadState::new());
132            state.atomic.store(zx::MonotonicInstant::get().into_nanos() as u64, Ordering::Relaxed);
133        });
134    }
135
136    /// Stops an operation by storing 0 in the thread-local atomic.
137    fn stop_operation() {
138        THREAD_STATE.with(|state| {
139            if let Some(state) = state.borrow().as_ref() {
140                state.atomic.store(0, Ordering::Relaxed);
141            }
142        });
143    }
144
145    /// Iterates over the registry, finds threads that have been running longer than the threshold,
146    /// and returns their `ThreadLockupInfo`.
147    pub fn get_long_running_threads(threshold: zx::MonotonicDuration) -> Vec<ThreadLockupInfo> {
148        let now = zx::MonotonicInstant::get();
149        let registry = REGISTRY.read();
150        registry
151            .iter()
152            .filter_map(|registered| {
153                // SAFETY: We hold the read lock on REGISTRY. Any thread exiting must
154                // acquire the write lock to remove its pointer before freeing the memory.
155                // So the pointer is valid as long as we hold the read lock.
156                let atomic = unsafe { &*registered.atomic };
157                let start_nanos = atomic.load(Ordering::Relaxed);
158                if start_nanos == 0 {
159                    return None;
160                }
161                let start_time = zx::MonotonicInstant::from_nanos(start_nanos as i64);
162                if now - start_time > threshold {
163                    Some(ThreadLockupInfo {
164                        thread: registered.thread.clone(),
165                        koid: registered.koid,
166                        start_time,
167                    })
168                } else {
169                    None
170                }
171            })
172            .collect()
173    }
174
175    /// Starts tracking the current operation on the current thread.
176    /// Returns a guard that stops tracking when dropped.
177    pub fn track() -> LockupDetectorGuard {
178        LockupDetectorGuard::new()
179    }
180
181    /// Pauses tracking for the current operation on the current thread.
182    /// Returns a guard that resumes tracking when dropped.
183    pub fn pause_tracking() -> LockupDetectorWaitingGuard {
184        LockupDetectorWaitingGuard::new()
185    }
186
187    /// Wraps a future to track its execution when polled.
188    pub fn track_future<F>(inner: F) -> LockupDetectorFuture<F> {
189        LockupDetectorFuture::new(inner)
190    }
191
192    /// Creates a channel where the receiver pauses tracking while waiting for messages.
193    pub fn tracked_channel<T>() -> (std::sync::mpsc::Sender<T>, LockupDetectorReceiver<T>) {
194        let (sender, receiver) = std::sync::mpsc::channel();
195        (sender, LockupDetectorReceiver::new(receiver))
196    }
197
198    pub fn active_rcu_read_locks<F>(mut check: F)
199    where
200        F: FnMut(&zx::Thread, zx::Koid, u8),
201    {
202        let registry = REGISTRY.read();
203        for registered in registry.iter() {
204            if registered.rcu_nesting_level.is_null() || registered.rcu_counter_index.is_null() {
205                continue;
206            }
207            // SAFETY: The pointers point to thread-local storage of the registered thread.
208            // Before the thread exits, its `ThreadState` is dropped, which acquires a write
209            // lock on `REGISTRY` before the thread-local storage is destroyed. Since we hold the
210            // read lock on `REGISTRY` here, the thread cannot complete its cleanup and destroy the
211            // TLS until we release the lock, ensuring the pointers remain valid.
212            let (nesting_level, counter_index) = unsafe {
213                (
214                    (*registered.rcu_nesting_level).load(Ordering::Relaxed),
215                    (*registered.rcu_counter_index).load(Ordering::Relaxed),
216                )
217            };
218            if nesting_level > 0 {
219                check(&registered.thread, registered.koid, counter_index);
220            }
221        }
222    }
223}
224
225pub struct LockupDetectorGuard;
226
227impl LockupDetectorGuard {
228    fn new() -> Self {
229        ThreadLockupDetector::start_operation();
230        Self
231    }
232}
233
234impl Drop for LockupDetectorGuard {
235    fn drop(&mut self) {
236        ThreadLockupDetector::stop_operation();
237    }
238}
239
240pub struct LockupDetectorWaitingGuard;
241
242impl LockupDetectorWaitingGuard {
243    fn new() -> Self {
244        ThreadLockupDetector::stop_operation();
245        Self
246    }
247}
248
249impl Drop for LockupDetectorWaitingGuard {
250    fn drop(&mut self) {
251        ThreadLockupDetector::start_operation();
252    }
253}
254
255#[pin_project]
256pub struct LockupDetectorFuture<F> {
257    #[pin]
258    inner: F,
259}
260
261impl<F> LockupDetectorFuture<F> {
262    fn new(inner: F) -> Self {
263        Self { inner }
264    }
265}
266
267impl<F: std::future::Future> std::future::Future for LockupDetectorFuture<F> {
268    type Output = F::Output;
269
270    fn poll(
271        self: std::pin::Pin<&mut Self>,
272        cx: &mut std::task::Context<'_>,
273    ) -> std::task::Poll<Self::Output> {
274        let _guard = LockupDetectorGuard::new();
275        let this = self.project();
276        this.inner.poll(cx)
277    }
278}
279
280pub struct LockupDetectorReceiver<T> {
281    inner: std::sync::mpsc::Receiver<T>,
282}
283
284impl<T> LockupDetectorReceiver<T> {
285    fn new(inner: std::sync::mpsc::Receiver<T>) -> Self {
286        Self { inner }
287    }
288
289    pub fn recv(&self) -> Result<T, std::sync::mpsc::RecvError> {
290        let _guard = LockupDetectorWaitingGuard::new();
291        self.inner.recv()
292    }
293
294    pub fn try_iter(&self) -> std::sync::mpsc::TryIter<'_, T> {
295        self.inner.try_iter()
296    }
297}
298
299#[cfg(test)]
300mod tests {
301    use super::*;
302
303    fn get_long_running_koids() -> Vec<zx::Koid> {
304        ThreadLockupDetector::get_long_running_threads(zx::MonotonicDuration::from_nanos(0))
305            .iter()
306            .map(|r| r.koid)
307            .collect()
308    }
309
310    #[test]
311    fn test_lockup_detector() {
312        let koid = fuchsia_runtime::with_thread_self(|thread| thread.koid()).unwrap();
313
314        {
315            let _guard = ThreadLockupDetector::track();
316
317            // Exceed threshold immediately with zero duration.
318            assert!(get_long_running_koids().contains(&koid));
319
320            // After triggering, it still contains it (we don't reset).
321            assert!(get_long_running_koids().contains(&koid));
322        }
323
324        // Guard dropped.
325        assert!(get_long_running_koids().is_empty());
326    }
327
328    #[test]
329    fn test_guard() {
330        let koid = fuchsia_runtime::with_thread_self(|thread| thread.koid()).unwrap();
331
332        {
333            let _guard = ThreadLockupDetector::track();
334            assert!(get_long_running_koids().contains(&koid));
335        }
336
337        // Guard dropped.
338        assert!(get_long_running_koids().is_empty());
339    }
340
341    #[test]
342    fn test_waiting_guard() {
343        let koid = fuchsia_runtime::with_thread_self(|thread| thread.koid()).unwrap();
344
345        let _guard = ThreadLockupDetector::track();
346
347        {
348            let _waiting_guard = ThreadLockupDetector::pause_tracking();
349            // Operation stopped during wait.
350            assert!(get_long_running_koids().is_empty());
351        }
352
353        // Guard dropped, operation restarted.
354        assert!(get_long_running_koids().contains(&koid));
355    }
356
357    #[test]
358    fn test_track_future() {
359        let (koid_tx, koid_rx) = std::sync::mpsc::channel();
360        let (signal_tx, signal_rx) = futures::channel::oneshot::channel::<()>();
361
362        let t = std::thread::spawn(move || {
363            let koid = fuchsia_runtime::with_thread_self(|thread| thread.koid()).unwrap();
364            koid_tx.send(koid).unwrap();
365
366            let fut = ThreadLockupDetector::track_future(async move {
367                signal_rx.await.unwrap();
368            });
369
370            fuchsia_async::LocalExecutor::default().run_singlethreaded(fut);
371
372            koid
373        });
374
375        let spawned_koid = koid_rx.recv().unwrap();
376
377        // Wait a bit to ensure it entered the future and is waiting.
378        std::thread::sleep(std::time::Duration::from_millis(100));
379
380        // Check that spawned_koid is NOT in long running koids.
381        assert!(!get_long_running_koids().contains(&spawned_koid));
382
383        // Now signal to unblock it.
384        signal_tx.send(()).unwrap();
385
386        t.join().unwrap();
387    }
388
389    #[test]
390    fn test_track_future_polling() {
391        let koid = fuchsia_runtime::with_thread_self(|thread| thread.koid()).unwrap();
392
393        // Before polling, should not be found.
394        assert!(!get_long_running_koids().contains(&koid));
395
396        let fut = ThreadLockupDetector::track_future(async {
397            assert!(get_long_running_koids().contains(&koid));
398        });
399
400        fuchsia_async::LocalExecutor::default().run_singlethreaded(fut);
401
402        // After polling, should not be found.
403        assert!(!get_long_running_koids().contains(&koid));
404    }
405
406    #[test]
407    fn test_track_channel() {
408        let (koid_tx, koid_rx) = std::sync::mpsc::channel();
409        let (tx, rx) = ThreadLockupDetector::tracked_channel();
410
411        let t = std::thread::spawn(move || {
412            let koid = fuchsia_runtime::with_thread_self(|thread| thread.koid()).unwrap();
413            koid_tx.send(koid).unwrap();
414
415            let _guard = ThreadLockupDetector::track();
416
417            // This will block.
418            rx.recv().unwrap();
419
420            koid
421        });
422
423        let spawned_koid = koid_rx.recv().unwrap();
424
425        // Wait a bit to ensure it entered rx.recv()
426        std::thread::sleep(std::time::Duration::from_millis(100));
427
428        // Check that spawned_koid is NOT in long running koids.
429        assert!(!get_long_running_koids().contains(&spawned_koid));
430
431        // Now send data to unblock it.
432        tx.send(()).unwrap();
433
434        t.join().unwrap();
435    }
436}
starnix_core/task/thread_lockup_detector.rs

starnix_core/task/
thread_lockup_detector.rs
