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)]
115pub struct ThreadLockupInfo {
116    pub thread: zx::Unowned<'static, zx::Thread>,
117    pub koid: zx::Koid,
118}
119
120/// Global registry of all tracked threads.
121static REGISTRY: LazyLock<
122    LockDepRwLock<HashSet<RegisteredThread>, ThreadLockupDetectorRegistryLock>,
123> = LazyLock::new(|| Default::default());
124
125impl ThreadLockupDetector {
126    /// Starts an operation by storing the current time in the thread-local atomic.
127    fn start_operation() {
128        THREAD_STATE.with(|state| {
129            let mut state = state.borrow_mut();
130            let state = state.get_or_insert_with(|| ThreadState::new());
131            state.atomic.store(zx::MonotonicInstant::get().into_nanos() as u64, Ordering::Relaxed);
132        });
133    }
134
135    /// Stops an operation by storing 0 in the thread-local atomic.
136    fn stop_operation() {
137        THREAD_STATE.with(|state| {
138            if let Some(state) = state.borrow().as_ref() {
139                state.atomic.store(0, Ordering::Relaxed);
140            }
141        });
142    }
143
144    /// Iterates over the registry, finds threads that have been running longer than the threshold,
145    /// and returns their `ThreadLockupInfo`.
146    pub fn get_long_running_threads(threshold: zx::MonotonicDuration) -> Vec<ThreadLockupInfo> {
147        let now = zx::MonotonicInstant::get();
148        let registry = REGISTRY.read();
149        registry
150            .iter()
151            .filter_map(|registered| {
152                // SAFETY: We hold the read lock on REGISTRY. Any thread exiting must
153                // acquire the write lock to remove its pointer before freeing the memory.
154                // So the pointer is valid as long as we hold the read lock.
155                let atomic = unsafe { &*registered.atomic };
156                let start_nanos = atomic.load(Ordering::Relaxed);
157                if start_nanos == 0 {
158                    return None;
159                }
160                let start_time = zx::MonotonicInstant::from_nanos(start_nanos as i64);
161                if now - start_time > threshold {
162                    Some(ThreadLockupInfo {
163                        thread: registered.thread.clone(),
164                        koid: registered.koid,
165                    })
166                } else {
167                    None
168                }
169            })
170            .collect()
171    }
172
173    /// Starts tracking the current operation on the current thread.
174    /// Returns a guard that stops tracking when dropped.
175    pub fn track() -> LockupDetectorGuard {
176        LockupDetectorGuard::new()
177    }
178
179    /// Pauses tracking for the current operation on the current thread.
180    /// Returns a guard that resumes tracking when dropped.
181    pub fn pause_tracking() -> LockupDetectorWaitingGuard {
182        LockupDetectorWaitingGuard::new()
183    }
184
185    /// Wraps a future to track its execution when polled.
186    pub fn track_future<F>(inner: F) -> LockupDetectorFuture<F> {
187        LockupDetectorFuture::new(inner)
188    }
189
190    /// Creates a channel where the receiver pauses tracking while waiting for messages.
191    pub fn tracked_channel<T>() -> (std::sync::mpsc::Sender<T>, LockupDetectorReceiver<T>) {
192        let (sender, receiver) = std::sync::mpsc::channel();
193        (sender, LockupDetectorReceiver::new(receiver))
194    }
195
196    pub fn active_rcu_read_locks<F>(mut check: F)
197    where
198        F: FnMut(&zx::Thread, zx::Koid, u8),
199    {
200        let registry = REGISTRY.read();
201        for registered in registry.iter() {
202            if registered.rcu_nesting_level.is_null() || registered.rcu_counter_index.is_null() {
203                continue;
204            }
205            // SAFETY: The pointers point to thread-local storage of the registered thread.
206            // Before the thread exits, its `ThreadState` is dropped, which acquires a write
207            // lock on `REGISTRY` before the thread-local storage is destroyed. Since we hold the
208            // read lock on `REGISTRY` here, the thread cannot complete its cleanup and destroy the
209            // TLS until we release the lock, ensuring the pointers remain valid.
210            let (nesting_level, counter_index) = unsafe {
211                (
212                    (*registered.rcu_nesting_level).load(Ordering::Relaxed),
213                    (*registered.rcu_counter_index).load(Ordering::Relaxed),
214                )
215            };
216            if nesting_level > 0 {
217                check(&registered.thread, registered.koid, counter_index);
218            }
219        }
220    }
221}
222
223pub struct LockupDetectorGuard;
224
225impl LockupDetectorGuard {
226    fn new() -> Self {
227        ThreadLockupDetector::start_operation();
228        Self
229    }
230}
231
232impl Drop for LockupDetectorGuard {
233    fn drop(&mut self) {
234        ThreadLockupDetector::stop_operation();
235    }
236}
237
238pub struct LockupDetectorWaitingGuard;
239
240impl LockupDetectorWaitingGuard {
241    fn new() -> Self {
242        ThreadLockupDetector::stop_operation();
243        Self
244    }
245}
246
247impl Drop for LockupDetectorWaitingGuard {
248    fn drop(&mut self) {
249        ThreadLockupDetector::start_operation();
250    }
251}
252
253#[pin_project]
254pub struct LockupDetectorFuture<F> {
255    #[pin]
256    inner: F,
257}
258
259impl<F> LockupDetectorFuture<F> {
260    fn new(inner: F) -> Self {
261        Self { inner }
262    }
263}
264
265impl<F: std::future::Future> std::future::Future for LockupDetectorFuture<F> {
266    type Output = F::Output;
267
268    fn poll(
269        self: std::pin::Pin<&mut Self>,
270        cx: &mut std::task::Context<'_>,
271    ) -> std::task::Poll<Self::Output> {
272        let _guard = LockupDetectorGuard::new();
273        let this = self.project();
274        this.inner.poll(cx)
275    }
276}
277
278pub struct LockupDetectorReceiver<T> {
279    inner: std::sync::mpsc::Receiver<T>,
280}
281
282impl<T> LockupDetectorReceiver<T> {
283    fn new(inner: std::sync::mpsc::Receiver<T>) -> Self {
284        Self { inner }
285    }
286
287    pub fn recv(&self) -> Result<T, std::sync::mpsc::RecvError> {
288        let _guard = LockupDetectorWaitingGuard::new();
289        self.inner.recv()
290    }
291
292    pub fn try_iter(&self) -> std::sync::mpsc::TryIter<'_, T> {
293        self.inner.try_iter()
294    }
295}
296
297#[cfg(test)]
298mod tests {
299    use super::*;
300
301    fn get_long_running_koids() -> Vec<zx::Koid> {
302        ThreadLockupDetector::get_long_running_threads(zx::MonotonicDuration::from_nanos(0))
303            .iter()
304            .map(|r| r.koid)
305            .collect()
306    }
307
308    #[test]
309    fn test_lockup_detector() {
310        let koid = fuchsia_runtime::with_thread_self(|thread| thread.koid()).unwrap();
311
312        {
313            let _guard = ThreadLockupDetector::track();
314
315            // Exceed threshold immediately with zero duration.
316            assert!(get_long_running_koids().contains(&koid));
317
318            // After triggering, it still contains it (we don't reset).
319            assert!(get_long_running_koids().contains(&koid));
320        }
321
322        // Guard dropped.
323        assert!(get_long_running_koids().is_empty());
324    }
325
326    #[test]
327    fn test_guard() {
328        let koid = fuchsia_runtime::with_thread_self(|thread| thread.koid()).unwrap();
329
330        {
331            let _guard = ThreadLockupDetector::track();
332            assert!(get_long_running_koids().contains(&koid));
333        }
334
335        // Guard dropped.
336        assert!(get_long_running_koids().is_empty());
337    }
338
339    #[test]
340    fn test_waiting_guard() {
341        let koid = fuchsia_runtime::with_thread_self(|thread| thread.koid()).unwrap();
342
343        let _guard = ThreadLockupDetector::track();
344
345        {
346            let _waiting_guard = ThreadLockupDetector::pause_tracking();
347            // Operation stopped during wait.
348            assert!(get_long_running_koids().is_empty());
349        }
350
351        // Guard dropped, operation restarted.
352        assert!(get_long_running_koids().contains(&koid));
353    }
354
355    #[test]
356    fn test_track_future() {
357        let (koid_tx, koid_rx) = std::sync::mpsc::channel();
358        let (signal_tx, signal_rx) = futures::channel::oneshot::channel::<()>();
359
360        let t = std::thread::spawn(move || {
361            let koid = fuchsia_runtime::with_thread_self(|thread| thread.koid()).unwrap();
362            koid_tx.send(koid).unwrap();
363
364            let fut = ThreadLockupDetector::track_future(async move {
365                signal_rx.await.unwrap();
366            });
367
368            fuchsia_async::LocalExecutor::default().run_singlethreaded(fut);
369
370            koid
371        });
372
373        let spawned_koid = koid_rx.recv().unwrap();
374
375        // Wait a bit to ensure it entered the future and is waiting.
376        std::thread::sleep(std::time::Duration::from_millis(100));
377
378        // Check that spawned_koid is NOT in long running koids.
379        assert!(!get_long_running_koids().contains(&spawned_koid));
380
381        // Now signal to unblock it.
382        signal_tx.send(()).unwrap();
383
384        t.join().unwrap();
385    }
386
387    #[test]
388    fn test_track_future_polling() {
389        let koid = fuchsia_runtime::with_thread_self(|thread| thread.koid()).unwrap();
390
391        // Before polling, should not be found.
392        assert!(!get_long_running_koids().contains(&koid));
393
394        let fut = ThreadLockupDetector::track_future(async {
395            assert!(get_long_running_koids().contains(&koid));
396        });
397
398        fuchsia_async::LocalExecutor::default().run_singlethreaded(fut);
399
400        // After polling, should not be found.
401        assert!(!get_long_running_koids().contains(&koid));
402    }
403
404    #[test]
405    fn test_track_channel() {
406        let (koid_tx, koid_rx) = std::sync::mpsc::channel();
407        let (tx, rx) = ThreadLockupDetector::tracked_channel();
408
409        let t = std::thread::spawn(move || {
410            let koid = fuchsia_runtime::with_thread_self(|thread| thread.koid()).unwrap();
411            koid_tx.send(koid).unwrap();
412
413            let _guard = ThreadLockupDetector::track();
414
415            // This will block.
416            rx.recv().unwrap();
417
418            koid
419        });
420
421        let spawned_koid = koid_rx.recv().unwrap();
422
423        // Wait a bit to ensure it entered rx.recv()
424        std::thread::sleep(std::time::Duration::from_millis(100));
425
426        // Check that spawned_koid is NOT in long running koids.
427        assert!(!get_long_running_koids().contains(&spawned_koid));
428
429        // Now send data to unblock it.
430        tx.send(()).unwrap();
431
432        t.join().unwrap();
433    }
434}
starnix_core/task/thread_lockup_detector.rs

starnix_core/task/
thread_lockup_detector.rs
