starnix_core/mm/
futex_table.rs

1// Copyright 2021 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::mm::memory::MemoryObject;
6use crate::mm::{CompareExchangeResult, ProtectionFlags};
7use crate::task::{CurrentTask, EventHandler, SignalHandler, SignalHandlerInner, Task, Waiter};
8use futures::channel::oneshot;
9use starnix_sync::{FutexTableStateLock, InterruptibleEvent, LockDepMutex};
10use starnix_types::futex_address::FutexAddress;
11use starnix_uapi::errors::Errno;
12use starnix_uapi::user_address::UserAddress;
13use starnix_uapi::{FUTEX_BITSET_MATCH_ANY, FUTEX_TID_MASK, FUTEX_WAITERS, errno, error};
14use std::collections::hash_map::Entry;
15use std::collections::{HashMap, VecDeque};
16use std::hash::Hash;
17use std::sync::{Arc, Weak};
18
19/// A table of futexes.
20///
21/// Each 32-bit aligned address in an address space can potentially have an associated futex that
22/// userspace can wait upon. This table is a sparse representation that has an actual WaitQueue
23/// only for those addresses that have ever actually had a futex operation performed on them.
24pub struct FutexTable<Key: FutexKey> {
25    /// The futexes associated with each address in each VMO.
26    ///
27    /// This HashMap is populated on-demand when futexes are used.
28    state: LockDepMutex<FutexTableState<Key>, FutexTableStateLock>,
29}
30
31impl<Key: FutexKey> Default for FutexTable<Key> {
32    fn default() -> Self {
33        Self { state: LockDepMutex::new(FutexTableState::default()) }
34    }
35}
36
37impl<Key: FutexKey> FutexTable<Key> {
38    /// Resolves the outcome of a blocking wait on this table.
39    ///
40    /// On success, the waker has already removed the waiter from the queue and there is nothing
41    /// left to do. On error (e.g. ETIMEDOUT, EINTR), `remove` takes the waiter out of the queue
42    /// to prevent a memory leak.
43    ///
44    /// Finding nothing to remove means a concurrent wake dequeued the waiter while it was waking
45    /// up for another reason. That wake counted this waiter as woken, so report success rather
46    /// than dropping the wake on the floor.
47    fn resolve_wait(
48        &self,
49        result: Result<(), Errno>,
50        remove: impl FnOnce(&mut FutexTableState<Key>) -> bool,
51    ) -> Result<(), Errno> {
52        result.or_else(|e| if remove(&mut self.state.lock()) { Err(e) } else { Ok(()) })
53    }
54
55    /// Wait on the futex at the given address given a boot deadline.
56    ///
57    /// See FUTEX_WAIT when passed a deadline in CLOCK_REALTIME.
58    pub fn wait_boot(
59        &self,
60        current_task: &CurrentTask,
61        addr: UserAddress,
62        value: u32,
63        mask: u32,
64        deadline: zx::BootInstant,
65        timer_slack: zx::BootDuration,
66    ) -> Result<(), Errno> {
67        let addr = FutexAddress::try_from(addr)?;
68        // Pre-fault the address before acquiring the FutexTable lock to make page faults or
69        // memory reclamation under the lock highly unlikely.
70        // TODO(https://fxbug.dev/548025031): Implement a non-blocking try_load and retry loop once
71        // supported by Zircon.
72        let _ = current_task.mm()?.atomic_load_u32_acquire(addr)?;
73        let mut state = self.state.lock();
74        // As the state is locked, no wake can happen before the waiter is registered.
75        // If the addr is remapped, we will read stale data, but we will not miss a futex wake.
76        // Acquire ordering to synchronize with userspace modifications to the value on other
77        // threads.
78        let loaded_value = current_task.mm()?.atomic_load_u32_acquire(addr)?;
79        if value != loaded_value {
80            return error!(EAGAIN);
81        }
82
83        let key = Key::get(current_task, addr)?;
84        // Linux futex_wait(): an already-expired timer returns ETIMEDOUT without sleeping.
85        if deadline != zx::BootInstant::INFINITE && deadline <= zx::BootInstant::get() {
86            return error!(ETIMEDOUT);
87        }
88        let waiter = Arc::new(Waiter::new());
89        let timer = zx::BootTimer::create();
90        let signal_handler = SignalHandler {
91            inner: SignalHandlerInner::None,
92            event_handler: EventHandler::None,
93            err_code: Some(errno!(ETIMEDOUT)),
94        };
95        waiter
96            .wake_on_zircon_signals(&timer, zx::Signals::TIMER_SIGNALED, signal_handler)
97            .expect("wait can only fail in OOM conditions");
98        timer
99            .set(deadline, timer_slack)
100            .expect("timer set cannot fail with valid handles and slack");
101        state.get_waiters_or_default(key.clone()).add(FutexWaiter {
102            mask,
103            notifiable: FutexNotifiable::new_internal_boot(Arc::downgrade(&waiter)),
104        });
105        std::mem::drop(state);
106
107        let result = waiter.wait(current_task);
108        self.resolve_wait(result, |state| {
109            state.remove_waiter_from_queue(key, &WaiterMatcher::BootWaiter(&waiter))
110        })
111    }
112
113    /// Wait on the futex at the given address.
114    ///
115    /// See FUTEX_WAIT.
116    pub fn wait(
117        &self,
118        current_task: &CurrentTask,
119        addr: UserAddress,
120        value: u32,
121        mask: u32,
122        deadline: zx::MonotonicInstant,
123    ) -> Result<(), Errno> {
124        let addr = FutexAddress::try_from(addr)?;
125        // Pre-fault the address before acquiring the FutexTable lock to make page faults or
126        // memory reclamation under the lock highly unlikely.
127        // TODO(https://fxbug.dev/548025031): Implement a non-blocking try_load and retry loop once
128        // supported by Zircon.
129        let _ = current_task.mm()?.atomic_load_u32_acquire(addr)?;
130        let mut state = self.state.lock();
131        // As the state is locked, no wake can happen before the waiter is registered.
132        // If the addr is remapped, we will read stale data, but we will not miss a futex wake.
133        // Acquire ordering to synchronize with userspace modifications to the value on other
134        // threads.
135        let loaded_value = current_task.mm()?.atomic_load_u32_acquire(addr)?;
136        if value != loaded_value {
137            return error!(EAGAIN);
138        }
139
140        let key = Key::get(current_task, addr)?;
141        // Linux futex_wait(): an already-expired timer returns ETIMEDOUT without sleeping.
142        if deadline != zx::MonotonicInstant::INFINITE && deadline <= zx::MonotonicInstant::get() {
143            return error!(ETIMEDOUT);
144        }
145        let event = InterruptibleEvent::new();
146        let guard = event.begin_wait();
147        state.get_waiters_or_default(key.clone()).add(FutexWaiter {
148            mask,
149            notifiable: FutexNotifiable::new_internal(Arc::downgrade(&event)),
150        });
151        std::mem::drop(state);
152
153        let result = current_task.block_until(guard, deadline);
154        self.resolve_wait(result, |state| {
155            state.remove_waiter_from_queue(key, &WaiterMatcher::Event(&event))
156        })
157    }
158
159    /// Wake the given number of waiters on futex at the given address. Returns the number of
160    /// waiters actually woken.
161    ///
162    /// See FUTEX_WAKE.
163    pub fn wake(
164        &self,
165        task: &Task,
166        addr: UserAddress,
167        count: usize,
168        mask: u32,
169    ) -> Result<usize, Errno> {
170        let addr = FutexAddress::try_from(addr)?;
171        let key = Key::get(task, addr)?;
172        Ok(self.state.lock().wake(key, count, mask))
173    }
174
175    /// Requeue the waiters to another address.
176    ///
177    /// See FUTEX_CMP_REQUEUE
178    pub fn requeue(
179        &self,
180        current_task: &CurrentTask,
181        addr: UserAddress,
182        wake_count: usize,
183        requeue_count: usize,
184        new_addr: UserAddress,
185        expected_value: Option<u32>,
186    ) -> Result<usize, Errno> {
187        let addr = FutexAddress::try_from(addr)?;
188        let new_addr = FutexAddress::try_from(new_addr)?;
189        if expected_value.is_some() {
190            // Pre-fault the address before acquiring the FutexTable lock to make page faults or
191            // memory reclamation under the lock highly unlikely.
192            // TODO(https://fxbug.dev/548025031): Implement a non-blocking try_load and retry loop once
193            // supported by Zircon.
194            let _ = current_task.mm()?.atomic_load_u32_acquire(addr)?;
195        }
196        let key = Key::get(current_task, addr)?;
197        let new_key = Key::get(current_task, new_addr)?;
198        let mut state = self.state.lock();
199        if let Some(expected) = expected_value {
200            // Use acquire ordering here to synchronize with mutex impls that store w/ release
201            // ordering.
202            let value = current_task.mm()?.atomic_load_u32_acquire(addr)?;
203            if value != expected {
204                return error!(EAGAIN);
205            }
206        }
207
208        Ok(state.requeue(key, new_key, wake_count, requeue_count))
209    }
210
211    /// Lock the futex at the given address.
212    ///
213    /// See FUTEX_LOCK_PI.
214    pub fn lock_pi(
215        &self,
216        current_task: &CurrentTask,
217        addr: UserAddress,
218        deadline: zx::MonotonicInstant,
219    ) -> Result<(), Errno> {
220        let addr = FutexAddress::try_from(addr)?;
221        let mm = current_task.mm()?;
222        // Perform a dummy CAS to pre-fault the page with write permissions (e.g. for COW pages)
223        // before acquiring the FutexTable lock to make page faults or memory reclamation under
224        // the lock highly unlikely.
225        // TODO(https://fxbug.dev/548025031): Implement a non-blocking try_load and retry loop once
226        // supported by Zircon.
227        if let Ok(current) = mm.atomic_load_u32_relaxed(addr) {
228            let _ = mm.atomic_compare_exchange_u32_acq_rel(addr, current, current);
229        }
230        let mut state = self.state.lock();
231        // As the state is locked, no unlock can happen before the waiter is registered.
232        // If the addr is remapped, we will read stale data, but we will not miss a futex unlock.
233        let key = Key::get(current_task, addr)?;
234
235        let tid = current_task.get_tid() as u32;
236
237        // Use a relaxed ordering because the compare/exchange below creates a synchronization
238        // point with userspace threads in the success case. No synchronization is required in
239        // failure cases.
240        let mut current_value = mm.atomic_load_u32_relaxed(addr)?;
241        let new_owner_tid = loop {
242            let new_owner_tid = current_value & FUTEX_TID_MASK;
243            if new_owner_tid == tid {
244                // From <https://man7.org/linux/man-pages/man2/futex.2.html>:
245                //
246                //   EDEADLK
247                //          (FUTEX_LOCK_PI, FUTEX_LOCK_PI2, FUTEX_TRYLOCK_PI,
248                //          FUTEX_CMP_REQUEUE_PI) The futex word at uaddr is
249                //          already locked by the caller.
250                return error!(EDEADLOCK);
251            }
252
253            if current_value == 0 {
254                // Use acq/rel ordering to synchronize with acquire ordering on userspace lock ops
255                // and with the release ordering on userspace unlock ops.
256                match mm.atomic_compare_exchange_weak_u32_acq_rel(addr, current_value, tid) {
257                    CompareExchangeResult::Success => return Ok(()),
258                    CompareExchangeResult::Stale { observed } => {
259                        current_value = observed;
260                        continue;
261                    }
262                    CompareExchangeResult::Error(e) => return Err(e),
263                }
264            }
265
266            // Use acq/rel ordering to synchronize with acquire ordering on userspace lock ops and
267            // with the release ordering on userspace unlock ops.
268            let target_value = current_value | FUTEX_WAITERS;
269            match mm.atomic_compare_exchange_u32_acq_rel(addr, current_value, target_value) {
270                CompareExchangeResult::Success => (),
271                CompareExchangeResult::Stale { observed } => {
272                    current_value = observed;
273                    continue;
274                }
275                CompareExchangeResult::Error(e) => return Err(e),
276            }
277            break new_owner_tid;
278        };
279
280        let event = InterruptibleEvent::new();
281        let guard = event.begin_wait();
282        let notifiable = FutexNotifiable::new_internal(Arc::downgrade(&event));
283        state
284            .get_rt_mutex_waiters_or_default(key.clone())
285            .push_back(RtMutexWaiter { tid, notifiable });
286        std::mem::drop(state);
287
288        // ESRCH  (FUTEX_LOCK_PI, FUTEX_LOCK_PI2, FUTEX_TRYLOCK_PI,
289        //        FUTEX_CMP_REQUEUE_PI) The thread ID in the futex word at
290        //        uaddr does not exist.
291        let result = current_task
292            .get_task(new_owner_tid as i32)
293            .ok()
294            .and_then(|o| o.running_state().unwrap().thread.get().map(|t| Arc::clone(&t.thread)))
295            .map_or_else(
296                || error!(ESRCH),
297                |owner| current_task.block_with_owner_until(guard, &owner, deadline),
298            );
299
300        // Being gone from the queue means `unlock_pi` picked this waiter as the new owner and
301        // already published its tid in the futex word, so the mutex is held even though the wait
302        // itself failed. Reporting the error would leave the mutex locked by a thread that
303        // believes it does not own it.
304        self.resolve_wait(result, |state| {
305            state.remove_rt_mutex_waiter_from_queue(key, &WaiterMatcher::Event(&event))
306        })
307    }
308
309    /// Unlock the futex at the given address.
310    ///
311    /// See FUTEX_UNLOCK_PI.
312    pub fn unlock_pi(&self, current_task: &CurrentTask, addr: UserAddress) -> Result<(), Errno> {
313        let addr = FutexAddress::try_from(addr)?;
314        let mm = current_task.mm()?;
315        // Perform a placeholder CAS to pre-fault the page with write permissions (e.g. for COW pages)
316        // before acquiring the FutexTable lock to make page faults or memory reclamation under
317        // the lock highly unlikely.
318        // TODO(https://fxbug.dev/548025031): Implement a non-blocking try_load and retry loop once
319        // supported by Zircon.
320        if let Ok(current) = mm.atomic_load_u32_relaxed(addr) {
321            let _ = mm.atomic_compare_exchange_u32_acq_rel(addr, current, current);
322        }
323        let mut state = self.state.lock();
324        let tid = current_task.get_tid() as u32;
325
326        let key = Key::get(current_task, addr)?;
327
328        // Use a relaxed ordering because the compare/exchange below creates a synchronization
329        // point with userspace threads in the success case. No synchronization is required in
330        // failure cases.
331        let mut expected_value = mm.atomic_load_u32_relaxed(addr)?;
332        if expected_value & FUTEX_TID_MASK != tid {
333            // From <https://man7.org/linux/man-pages/man2/futex.2.html>:
334            //
335            //   EPERM  (FUTEX_UNLOCK_PI) The caller does not own the lock
336            //          represented by the futex word.
337            return error!(EPERM);
338        }
339
340        loop {
341            let maybe_waiter = state.pop_rt_mutex_waiter(key.clone());
342            let target_value = maybe_waiter.as_ref().map_or(0, |waiter| waiter.tid);
343
344            // Use acq/rel ordering to synchronize with acquire ordering on userspace lock ops and
345            // with the release ordering on userspace unlock ops.
346            let handoff =
347                mm.atomic_compare_exchange_u32_acq_rel(addr, expected_value, target_value);
348            if let Err(e) = handoff_result(handoff) {
349                if let Some(waiter) = maybe_waiter {
350                    state.unpop_rt_mutex_waiter(key, waiter);
351                }
352                return Err(e);
353            }
354            // The futex word now holds the value just written, which is what a further handoff
355            // has to compare against.
356            expected_value = target_value;
357
358            let Some(mut waiter) = maybe_waiter else {
359                // We can stop trying to notify a thread if there are no more waiters.
360                break;
361            };
362
363            if waiter.notifiable.notify() {
364                break;
365            }
366
367            // If we couldn't notify the waiter, then we need to pull the next thread off the
368            // waiter list.
369        }
370
371        Ok(())
372    }
373}
374
375/// Maps the outcome of the compare-exchange that hands a PI mutex over to its next owner.
376fn handoff_result(result: CompareExchangeResult<u32>) -> Result<(), Errno> {
377    match result {
378        CompareExchangeResult::Success => Ok(()),
379        // From <https://man7.org/linux/man-pages/man2/futex.2.html>:
380        //
381        //   EINVAL (FUTEX_LOCK_PI, FUTEX_LOCK_PI2, FUTEX_TRYLOCK_PI,
382        //       FUTEX_UNLOCK_PI) The kernel detected an inconsistency
383        //       between the user-space state at uaddr and the kernel
384        //       state.  This indicates either state corruption or that the
385        //       kernel found a waiter on uaddr which is waiting via
386        //       FUTEX_WAIT or FUTEX_WAIT_BITSET.
387        CompareExchangeResult::Stale { .. } => error!(EINVAL),
388        // From <https://man7.org/linux/man-pages/man2/futex.2.html>:
389        //
390        //   EACCES No read access to the memory of a futex word.
391        CompareExchangeResult::Error(_) => error!(EACCES),
392    }
393}
394
395impl FutexTable<SharedFutexKey> {
396    /// Wait on the futex at the given offset in the memory.
397    ///
398    /// Returns a receiver that will be signaled when the futex is woken, and an
399    /// `Arc<()>` token that must be kept alive by the caller for the duration of the
400    /// wait. If the caller drops the token (e.g., if the external client
401    /// disconnects), the waiter is marked as stale and will be garbage-collected by the
402    /// next futex operation on this table.
403    ///
404    /// See FUTEX_WAIT.
405    pub fn external_wait(
406        &self,
407        memory: MemoryObject,
408        offset: u64,
409        value: u32,
410        mask: u32,
411    ) -> Result<(Arc<()>, oneshot::Receiver<()>), Errno> {
412        let key = SharedFutexKey::new(&memory, offset);
413        let mut state = self.state.lock();
414        // As the state is locked, no wake can happen before the waiter is registered.
415        Self::external_check_futex_value(&memory, offset, value)?;
416
417        let token = Arc::new(());
418        let (sender, receiver) = oneshot::channel::<()>();
419        state.get_waiters_or_default(key).add(FutexWaiter {
420            mask,
421            notifiable: FutexNotifiable::new_external(Arc::downgrade(&token), sender),
422        });
423        Ok((token, receiver))
424    }
425
426    /// Wake the given number of waiters on futex at the given offset in the memory. Returns the
427    /// number of waiters actually woken.
428    ///
429    /// See FUTEX_WAKE.
430    pub fn external_wake(
431        &self,
432        memory: MemoryObject,
433        offset: u64,
434        count: usize,
435        mask: u32,
436    ) -> Result<usize, Errno> {
437        Ok(self.state.lock().wake(SharedFutexKey::new(&memory, offset), count, mask))
438    }
439
440    pub fn external_requeue(
441        &self,
442        first_memory: MemoryObject,
443        first_offset: u64,
444        second_memory: Option<MemoryObject>,
445        second_offset: u64,
446        wake_count: usize,
447        requeue_count: usize,
448        expected_value: Option<u32>,
449    ) -> Result<usize, Errno> {
450        let first_key = SharedFutexKey::new(&first_memory, first_offset);
451        let second_key = match second_memory.as_ref() {
452            Some(second_memory) => SharedFutexKey::new(second_memory, second_offset),
453            None => SharedFutexKey::new(&first_memory, second_offset),
454        };
455        // If/when we move from a single table mutex to a mutex per futex, we'll likely want to
456        // define a consistent SharedFutexKey sort order independent of which is "first" and which
457        // is "second" in this call. Then we can acquire each of the two mutexes corresponding to
458        // each of the two futexes per that sort order. This way, we can be holding both mutexes to
459        // make the requeue atomic despite each futex having its own mutex, while avoiding
460        // deadlocks. But for now we lock the whole FutexTable.
461        let mut state = self.state.lock();
462        if let Some(expected) = expected_value {
463            // The state being locked is how this is included in the set of atomic changes.
464            Self::external_check_futex_value(&first_memory, first_offset, expected)?;
465        }
466        Ok(state.requeue(first_key, second_key, wake_count, requeue_count))
467    }
468
469    fn external_check_futex_value(
470        memory: &MemoryObject,
471        offset: u64,
472        value: u32,
473    ) -> Result<(), Errno> {
474        let loaded_value = {
475            // TODO: This read should be atomic.
476            let mut buf = [0u8; 4];
477            memory.read(&mut buf, offset).map_err(|_| errno!(EINVAL))?;
478            u32::from_ne_bytes(buf)
479        };
480        if loaded_value != value {
481            return error!(EAGAIN);
482        }
483        Ok(())
484    }
485}
486
487pub trait FutexKey: Sized + Ord + Hash + Clone {
488    fn get(task: &Task, addr: FutexAddress) -> Result<Self, Errno>;
489    fn get_table_from_task(task: &Task) -> Result<Arc<FutexTable<Self>>, Errno>;
490}
491
492#[derive(Debug, Clone, Eq, Hash, PartialEq, Ord, PartialOrd)]
493pub struct PrivateFutexKey {
494    addr: FutexAddress,
495}
496
497impl FutexKey for PrivateFutexKey {
498    fn get(_task: &Task, addr: FutexAddress) -> Result<Self, Errno> {
499        Ok(PrivateFutexKey { addr })
500    }
501
502    fn get_table_from_task(task: &Task) -> Result<Arc<FutexTable<Self>>, Errno> {
503        Ok(task.mm()?.futex.clone())
504    }
505}
506
507#[derive(Debug, Clone, Eq, Hash, PartialEq, Ord, PartialOrd)]
508pub struct SharedFutexKey {
509    // No chance of collisions since koids are never reused:
510    // https://fuchsia.dev/fuchsia-src/concepts/kernel/concepts#kernel_object_ids
511    koid: zx::Koid,
512    offset: u64,
513}
514
515impl FutexKey for SharedFutexKey {
516    fn get(task: &Task, addr: FutexAddress) -> Result<Self, Errno> {
517        let (memory, offset) = task.mm()?.get_mapping_memory(addr.into(), ProtectionFlags::READ)?;
518        Ok(SharedFutexKey::new(&memory, offset))
519    }
520
521    fn get_table_from_task(task: &Task) -> Result<Arc<FutexTable<Self>>, Errno> {
522        Ok(task.kernel().shared_futexes.clone())
523    }
524}
525
526impl SharedFutexKey {
527    fn new(memory: &MemoryObject, offset: u64) -> Self {
528        Self { koid: memory.get_koid(), offset }
529    }
530}
531
532struct FutexTableState<Key: FutexKey> {
533    waiters: HashMap<Key, FutexWaiters>,
534    rt_mutex_waiters: HashMap<Key, VecDeque<RtMutexWaiter>>,
535}
536
537impl<Key: FutexKey> Default for FutexTableState<Key> {
538    fn default() -> Self {
539        Self { waiters: Default::default(), rt_mutex_waiters: Default::default() }
540    }
541}
542
543impl<Key: FutexKey> FutexTableState<Key> {
544    /// Returns the FutexWaiters for a given address, creating an empty one if none is registered.
545    fn get_waiters_or_default(&mut self, key: Key) -> &mut FutexWaiters {
546        self.waiters.entry(key).or_default()
547    }
548
549    fn wake(&mut self, key: Key, count: usize, mask: u32) -> usize {
550        let entry = self.waiters.entry(key);
551        match entry {
552            Entry::Vacant(_) => 0,
553            Entry::Occupied(mut entry) => {
554                let count = entry.get_mut().notify(mask, count);
555                if entry.get().is_empty() {
556                    entry.remove();
557                }
558                count
559            }
560        }
561    }
562
563    fn requeue(
564        &mut self,
565        key: Key,
566        new_key: Key,
567        wake_count: usize,
568        requeue_count: usize,
569    ) -> usize {
570        let woken;
571        let to_requeue;
572        match self.waiters.entry(key) {
573            Entry::Vacant(_) => return 0,
574            Entry::Occupied(mut entry) => {
575                // Wake up at most `wake_count` waiters.
576                woken = entry.get_mut().notify(FUTEX_BITSET_MATCH_ANY, wake_count);
577
578                // Dequeue up to `requeue_count` waiters to requeue below.
579                to_requeue = entry.get_mut().split_for_requeue(requeue_count);
580
581                if entry.get().is_empty() {
582                    entry.remove();
583                }
584            }
585        }
586
587        let requeued = to_requeue.0.len();
588        if !to_requeue.is_empty() {
589            self.get_waiters_or_default(new_key).transfer(to_requeue);
590        }
591
592        woken + requeued
593    }
594
595    /// Returns the RT-Mutex waiters queue for a given address, creating an empty queue if none is
596    /// registered.
597    fn get_rt_mutex_waiters_or_default(&mut self, key: Key) -> &mut VecDeque<RtMutexWaiter> {
598        self.rt_mutex_waiters.entry(key).or_default()
599    }
600
601    /// Pop the next RT-Mutex for the given address.
602    fn pop_rt_mutex_waiter(&mut self, key: Key) -> Option<RtMutexWaiter> {
603        let entry = self.rt_mutex_waiters.entry(key);
604        match entry {
605            Entry::Vacant(_) => None,
606            Entry::Occupied(mut entry) => {
607                let mut waiter = entry.get_mut().pop_front();
608                // Clean up the hash map entry if the queue is empty. We do this
609                // regardless of whether `pop_front` returned a waiter or `None`,
610                // effectively garbage collecting erroneously empty map entries.
611                if entry.get().is_empty() {
612                    entry.remove();
613                } else if let Some(waiter) = &mut waiter {
614                    waiter.tid |= FUTEX_WAITERS;
615                }
616                waiter
617            }
618        }
619    }
620
621    /// Puts a waiter popped for a handoff that did not happen back at the head of its queue.
622    ///
623    /// Clearing `FUTEX_WAITERS` undoes what `pop_rt_mutex_waiter` set, restoring the tid the
624    /// waiter was queued with. The waiter must go back, otherwise it would conclude from its
625    /// absence that it was made the new owner of a mutex it does not hold.
626    fn unpop_rt_mutex_waiter(&mut self, key: Key, mut waiter: RtMutexWaiter) {
627        waiter.tid &= FUTEX_TID_MASK;
628        self.get_rt_mutex_waiters_or_default(key).push_front(waiter);
629    }
630
631    /// Removes the waiter designated by `matcher` from the `FUTEX_WAIT` queue it is waiting on.
632    ///
633    /// Returns whether the waiter was still queued. A `false` return means the waiter was
634    /// already dequeued by a concurrent wake.
635    fn remove_waiter_from_queue(&mut self, key: Key, matcher: &WaiterMatcher<'_>) -> bool {
636        search_and_remove_waiter(&mut self.waiters, key, matcher)
637    }
638
639    /// Removes a PI-mutex (`FUTEX_LOCK_PI`) waiter.
640    ///
641    /// Returns whether the waiter was still queued. A `false` return means `unlock_pi` already
642    /// dequeued it to make it the new owner of the mutex.
643    fn remove_rt_mutex_waiter_from_queue(&mut self, key: Key, matcher: &WaiterMatcher<'_>) -> bool {
644        search_and_remove_waiter(&mut self.rt_mutex_waiters, key, matcher)
645    }
646}
647
648/// Searches `queues` for the waiter designated by `matcher` and removes it from the queue it is
649/// waiting on.
650///
651/// This uses a two-step approach:
652/// 1. O(1) Fast path: Check the `key` where the waiter originally went to sleep.
653/// 2. O(N) Fallback: If not found (e.g. moved via `FUTEX_REQUEUE`), scan all futexes.
654///
655/// Queues left empty along the way are dropped from `queues`, whether or not they held the
656/// waiter.
657///
658/// Returns whether the waiter was still queued. A `false` return means the waiter was already
659/// dequeued by a concurrent wake.
660fn search_and_remove_waiter<Key: FutexKey, Queue: WaiterQueue>(
661    queues: &mut HashMap<Key, Queue>,
662    key: Key,
663    matcher: &WaiterMatcher<'_>,
664) -> bool {
665    if let Entry::Occupied(mut entry) = queues.entry(key) {
666        let found = entry.get_mut().remove_waiter(matcher);
667        if entry.get().is_empty() {
668            entry.remove();
669        }
670        if found {
671            return true;
672        }
673    }
674
675    // The waiter sits in at most one queue, so stop looking once it turns up. The scan still
676    // visits the remaining queues to drop the ones left empty.
677    let mut found = false;
678    queues.retain(|_, waiters| {
679        if !found {
680            found = waiters.remove_waiter(matcher);
681        }
682        !waiters.is_empty()
683    });
684    found
685}
686
687/// A queue of waiters parked on a single futex key.
688trait WaiterQueue {
689    /// Removes the waiter designated by `matcher` from the queue, also garbage collecting stale
690    /// waiters.
691    ///
692    /// Returns whether that waiter was in the queue. Removing stale waiters alone does not count
693    /// as a match: a caller that concludes it was dequeued by a wake would otherwise report a
694    /// wake that never happened.
695    fn remove_waiter(&mut self, matcher: &WaiterMatcher<'_>) -> bool;
696
697    /// Returns whether the queue holds no waiter.
698    fn is_empty(&self) -> bool;
699}
700
701/// Designates the waiter a removal operation is looking for.
702///
703/// The queues mix the waiters of every flavor of `FUTEX_WAIT`, so a removal needs to describe
704/// which one of them belongs to the caller.
705enum WaiterMatcher<'a> {
706    /// The waiter blocked on the given `InterruptibleEvent`. See `FutexNotifiable::Internal`.
707    Event(&'a Arc<InterruptibleEvent>),
708
709    /// The waiter blocked on the given `Waiter`. See `FutexNotifiable::InternalBoot`.
710    BootWaiter(&'a Arc<Waiter>),
711}
712
713impl WaiterMatcher<'_> {
714    /// Returns whether `notifiable` designates the waiter this matcher is looking for.
715    fn matches(&self, notifiable: &FutexNotifiable) -> bool {
716        match (self, notifiable) {
717            (Self::Event(event), FutexNotifiable::Internal(weak)) => {
718                weak.upgrade().is_some_and(|strong| Arc::ptr_eq(&strong, event))
719            }
720            (Self::BootWaiter(waiter), FutexNotifiable::InternalBoot(weak)) => {
721                weak.upgrade().is_some_and(|strong| Arc::ptr_eq(&strong, waiter))
722            }
723            _ => false,
724        }
725    }
726}
727
728/// Abstraction over a process waiting on a Futex that can be notified.
729enum FutexNotifiable {
730    /// An internal process waiting on a Futex.
731    Internal(Weak<InterruptibleEvent>),
732    // An internal process waiting on a Futex with a boot deadline.
733    InternalBoot(Weak<Waiter>),
734    /// An external process waiting on a Futex.
735    // The sender needs to be an option so that one can send the notification while only holding a
736    // mut reference on the ExternalWaiter.
737    External(Weak<()>, Option<oneshot::Sender<()>>),
738}
739
740impl FutexNotifiable {
741    fn new_internal(event: Weak<InterruptibleEvent>) -> Self {
742        Self::Internal(event)
743    }
744
745    fn new_internal_boot(waiter: Weak<Waiter>) -> Self {
746        Self::InternalBoot(waiter)
747    }
748
749    fn new_external(token: Weak<()>, sender: oneshot::Sender<()>) -> Self {
750        Self::External(token, Some(sender))
751    }
752
753    /// Tries to notify the process. Returns `true` is the process have been notified. Returns
754    /// `false` otherwise. This means the process is stale and will never be available again.
755    fn notify(&mut self) -> bool {
756        match self {
757            Self::Internal(event) => {
758                if let Some(event) = event.upgrade() {
759                    event.notify();
760                    true
761                } else {
762                    false
763                }
764            }
765            Self::InternalBoot(waiter) => {
766                if let Some(waiter) = waiter.upgrade() {
767                    waiter.notify();
768                    true
769                } else {
770                    false
771                }
772            }
773            Self::External(_, sender) => {
774                if let Some(sender) = sender.take() {
775                    sender.send(()).is_ok()
776                } else {
777                    false
778                }
779            }
780        }
781    }
782
783    fn is_stale(&self) -> bool {
784        match self {
785            Self::Internal(weak) => weak.strong_count() == 0,
786            Self::External(weak, _) => weak.strong_count() == 0,
787            Self::InternalBoot(weak) => weak.strong_count() == 0,
788        }
789    }
790}
791
792struct FutexWaiter {
793    mask: u32,
794    notifiable: FutexNotifiable,
795}
796
797#[derive(Default)]
798struct FutexWaiters(VecDeque<FutexWaiter>);
799
800impl FutexWaiters {
801    fn add(&mut self, waiter: FutexWaiter) {
802        self.0.push_back(waiter);
803    }
804
805    fn notify(&mut self, mask: u32, count: usize) -> usize {
806        let mut woken = 0;
807        self.0.retain_mut(|waiter| {
808            if woken == count || waiter.mask & mask == 0 {
809                return true;
810            }
811            // The send will fail if the receiver is gone, which means nothing was actualling
812            // waiting on the futex.
813            if waiter.notifiable.notify() {
814                woken += 1;
815            }
816            false
817        });
818        woken
819    }
820
821    fn transfer(&mut self, mut other: Self) {
822        self.0.append(&mut other.0);
823    }
824
825    fn split_for_requeue(&mut self, count: usize) -> Self {
826        let count = std::cmp::min(count, self.0.len());
827        let tail = self.0.split_off(count);
828        let head = std::mem::replace(&mut self.0, tail);
829        FutexWaiters(head)
830    }
831}
832
833/// An entry of a waiter queue, which a `WaiterMatcher` can be tested against.
834trait QueuedWaiter {
835    fn notifiable(&self) -> &FutexNotifiable;
836}
837
838impl QueuedWaiter for FutexWaiter {
839    fn notifiable(&self) -> &FutexNotifiable {
840        &self.notifiable
841    }
842}
843
844impl QueuedWaiter for RtMutexWaiter {
845    fn notifiable(&self) -> &FutexNotifiable {
846        &self.notifiable
847    }
848}
849
850impl<W: QueuedWaiter> WaiterQueue for VecDeque<W> {
851    fn remove_waiter(&mut self, matcher: &WaiterMatcher<'_>) -> bool {
852        let mut found = false;
853        self.retain(|w| {
854            if matcher.matches(w.notifiable()) {
855                found = true;
856                return false;
857            }
858            !w.notifiable().is_stale()
859        });
860        found
861    }
862
863    fn is_empty(&self) -> bool {
864        VecDeque::is_empty(self)
865    }
866}
867
868impl WaiterQueue for FutexWaiters {
869    fn remove_waiter(&mut self, matcher: &WaiterMatcher<'_>) -> bool {
870        self.0.remove_waiter(matcher)
871    }
872
873    fn is_empty(&self) -> bool {
874        self.0.is_empty()
875    }
876}
877
878struct RtMutexWaiter {
879    /// The tid, possibly with the FUTEX_WAITERS bit set.
880    tid: u32,
881
882    notifiable: FutexNotifiable,
883}
884
885#[cfg(test)]
886mod tests {
887    use super::*;
888    use starnix_sync::InterruptibleEvent;
889    use starnix_uapi::restricted_aspace::RESTRICTED_ASPACE_BASE;
890    use starnix_uapi::user_address::UserAddress;
891
892    #[fuchsia::test]
893    fn test_remove_waiter_simple() {
894        let mut state = FutexTableState::<PrivateFutexKey>::default();
895        let key = PrivateFutexKey {
896            addr: FutexAddress::try_from(UserAddress::from(
897                (RESTRICTED_ASPACE_BASE + 0x1000) as u64,
898            ))
899            .unwrap(),
900        };
901        let event = Arc::new(InterruptibleEvent::new());
902
903        state.get_waiters_or_default(key.clone()).add(FutexWaiter {
904            mask: u32::MAX,
905            notifiable: FutexNotifiable::new_internal(Arc::downgrade(&event)),
906        });
907
908        assert_eq!(state.waiters.len(), 1);
909        assert!(state.remove_waiter_from_queue(key, &WaiterMatcher::Event(&event)));
910        assert_eq!(state.waiters.len(), 0);
911    }
912
913    #[fuchsia::test]
914    fn test_remove_waiter_requeued() {
915        let mut state = FutexTableState::<PrivateFutexKey>::default();
916        let key1 = PrivateFutexKey {
917            addr: FutexAddress::try_from(UserAddress::from(
918                (RESTRICTED_ASPACE_BASE + 0x1000) as u64,
919            ))
920            .unwrap(),
921        };
922        let key2 = PrivateFutexKey {
923            addr: FutexAddress::try_from(UserAddress::from(
924                (RESTRICTED_ASPACE_BASE + 0x2000) as u64,
925            ))
926            .unwrap(),
927        };
928        let event = Arc::new(InterruptibleEvent::new());
929
930        state.get_waiters_or_default(key2.clone()).add(FutexWaiter {
931            mask: u32::MAX,
932            notifiable: FutexNotifiable::new_internal(Arc::downgrade(&event)),
933        });
934
935        assert_eq!(state.waiters.len(), 1);
936        assert!(state.remove_waiter_from_queue(key1, &WaiterMatcher::Event(&event)));
937        assert_eq!(state.waiters.len(), 0);
938    }
939
940    /// A wake that races with an interruption must not be lost.
941    ///
942    /// When a signal interrupts a waiter, the waiter can still be sitting in the queue when
943    /// another thread issues a wake. That wake counts the waiter as woken, so the waiter must
944    /// notice that it has been dequeued and report the wake instead of the interruption.
945    #[fuchsia::test]
946    fn test_wake_racing_with_interruption_is_not_lost() {
947        let mut state = FutexTableState::<PrivateFutexKey>::default();
948        let key = PrivateFutexKey {
949            addr: FutexAddress::try_from(UserAddress::from(
950                (RESTRICTED_ASPACE_BASE + 0x1000) as u64,
951            ))
952            .unwrap(),
953        };
954        let event = InterruptibleEvent::new();
955        let _guard = event.begin_wait();
956
957        state.get_waiters_or_default(key.clone()).add(FutexWaiter {
958            mask: u32::MAX,
959            notifiable: FutexNotifiable::new_internal(Arc::downgrade(&event)),
960        });
961
962        // A signal interrupts the waiter before it is dequeued.
963        event.interrupt();
964
965        // The waker still sees the waiter in the queue and counts it as woken.
966        assert_eq!(state.wake(key.clone(), 1, FUTEX_BITSET_MATCH_ANY), 1);
967
968        // The interrupted waiter must therefore observe that it is no longer queued, which tells
969        // it to report the wake rather than the interruption.
970        assert!(!state.remove_waiter_from_queue(key, &WaiterMatcher::Event(&event)));
971    }
972
973    /// The boot-deadline flavor of `FUTEX_WAIT` must resolve the same race the same way.
974    #[fuchsia::test]
975    fn test_wake_racing_with_boot_waiter_removal_is_not_lost() {
976        let mut state = FutexTableState::<PrivateFutexKey>::default();
977        let key = PrivateFutexKey {
978            addr: FutexAddress::try_from(UserAddress::from(
979                (RESTRICTED_ASPACE_BASE + 0x1000) as u64,
980            ))
981            .unwrap(),
982        };
983        let waiter = Arc::new(Waiter::new());
984
985        state.get_waiters_or_default(key.clone()).add(FutexWaiter {
986            mask: u32::MAX,
987            notifiable: FutexNotifiable::new_internal_boot(Arc::downgrade(&waiter)),
988        });
989
990        // The waker dequeues the waiter and counts it as woken, whether or not the waiter was
991        // about to give up on its deadline.
992        assert_eq!(state.wake(key.clone(), 1, FUTEX_BITSET_MATCH_ANY), 1);
993
994        assert!(!state.remove_waiter_from_queue(key, &WaiterMatcher::BootWaiter(&waiter)));
995    }
996
997    /// A waiter must only be removed by a matcher of its own kind.
998    ///
999    /// The two kinds share the same queues, so a mismatched removal must neither report a match
1000    /// nor dequeue the waiter, which would turn the next wake into a lost one.
1001    #[fuchsia::test]
1002    fn test_remove_waiter_ignores_other_waiter_kinds() {
1003        let mut state = FutexTableState::<PrivateFutexKey>::default();
1004        let key = PrivateFutexKey {
1005            addr: FutexAddress::try_from(UserAddress::from(
1006                (RESTRICTED_ASPACE_BASE + 0x1000) as u64,
1007            ))
1008            .unwrap(),
1009        };
1010        let event = InterruptibleEvent::new();
1011        let waiter = Arc::new(Waiter::new());
1012
1013        state.get_waiters_or_default(key.clone()).add(FutexWaiter {
1014            mask: u32::MAX,
1015            notifiable: FutexNotifiable::new_internal_boot(Arc::downgrade(&waiter)),
1016        });
1017
1018        assert!(!state.remove_waiter_from_queue(key.clone(), &WaiterMatcher::Event(&event)));
1019        assert_eq!(state.waiters.len(), 1);
1020
1021        assert!(state.remove_waiter_from_queue(key, &WaiterMatcher::BootWaiter(&waiter)));
1022        assert_eq!(state.waiters.len(), 0);
1023    }
1024
1025    #[fuchsia::test]
1026    fn test_remove_rt_mutex_waiter() {
1027        let mut state = FutexTableState::<PrivateFutexKey>::default();
1028        let key = PrivateFutexKey {
1029            addr: FutexAddress::try_from(UserAddress::from(
1030                (RESTRICTED_ASPACE_BASE + 0x1000) as u64,
1031            ))
1032            .unwrap(),
1033        };
1034        let event = Arc::new(InterruptibleEvent::new());
1035
1036        state.get_rt_mutex_waiters_or_default(key.clone()).push_back(RtMutexWaiter {
1037            tid: 1,
1038            notifiable: FutexNotifiable::new_internal(Arc::downgrade(&event)),
1039        });
1040
1041        assert_eq!(state.rt_mutex_waiters.len(), 1);
1042        state.remove_rt_mutex_waiter_from_queue(key, &WaiterMatcher::Event(&event));
1043        assert_eq!(state.rt_mutex_waiters.len(), 0);
1044    }
1045
1046    /// A PI-mutex waiter must learn whether it was still queued.
1047    ///
1048    /// `lock_pi` relies on that answer: a waiter that `unlock_pi` dequeued has been made the
1049    /// owner of the mutex, and must report success even if its wait failed.
1050    #[fuchsia::test]
1051    fn test_remove_rt_mutex_waiter_reports_whether_it_was_queued() {
1052        let mut state = FutexTableState::<PrivateFutexKey>::default();
1053        let key = PrivateFutexKey {
1054            addr: FutexAddress::try_from(UserAddress::from(
1055                (RESTRICTED_ASPACE_BASE + 0x1000) as u64,
1056            ))
1057            .unwrap(),
1058        };
1059        let event = InterruptibleEvent::new();
1060        let queue_waiter = |state: &mut FutexTableState<PrivateFutexKey>| {
1061            state.get_rt_mutex_waiters_or_default(key.clone()).push_back(RtMutexWaiter {
1062                tid: 1,
1063                notifiable: FutexNotifiable::new_internal(Arc::downgrade(&event)),
1064            });
1065        };
1066
1067        queue_waiter(&mut state);
1068        assert!(
1069            state.remove_rt_mutex_waiter_from_queue(key.clone(), &WaiterMatcher::Event(&event))
1070        );
1071
1072        // The handoff performed by `unlock_pi` takes the waiter out of the queue.
1073        queue_waiter(&mut state);
1074        assert!(state.pop_rt_mutex_waiter(key.clone()).is_some());
1075
1076        assert!(!state.remove_rt_mutex_waiter_from_queue(key, &WaiterMatcher::Event(&event)));
1077    }
1078
1079    /// Collecting a dead PI-mutex waiter is not a match.
1080    ///
1081    /// Reporting a match would tell the caller that its own waiter was still queued, while
1082    /// leaving the queue in place would leak the dead entry.
1083    #[fuchsia::test]
1084    fn test_remove_rt_mutex_waiter_stale_cleanup_is_not_a_match() {
1085        let mut state = FutexTableState::<PrivateFutexKey>::default();
1086        let key = PrivateFutexKey {
1087            addr: FutexAddress::try_from(UserAddress::from(
1088                (RESTRICTED_ASPACE_BASE + 0x1000) as u64,
1089            ))
1090            .unwrap(),
1091        };
1092
1093        {
1094            let dead_event = InterruptibleEvent::new();
1095            state.get_rt_mutex_waiters_or_default(key.clone()).push_back(RtMutexWaiter {
1096                tid: 1,
1097                notifiable: FutexNotifiable::new_internal(Arc::downgrade(&dead_event)),
1098            });
1099        } // dead_event is dropped here, so its waiter becomes stale.
1100
1101        let event = InterruptibleEvent::new();
1102        assert!(!state.remove_rt_mutex_waiter_from_queue(key, &WaiterMatcher::Event(&event)));
1103        assert_eq!(state.rt_mutex_waiters.len(), 0, "the stale waiter should be collected");
1104    }
1105
1106    /// Queues emptied by the fallback scan are collected too, not just the one holding the
1107    /// waiter.
1108    #[fuchsia::test]
1109    fn test_remove_waiter_collects_queues_emptied_by_the_scan() {
1110        let mut state = FutexTableState::<PrivateFutexKey>::default();
1111        let stale_key = PrivateFutexKey {
1112            addr: FutexAddress::try_from(UserAddress::from(
1113                (RESTRICTED_ASPACE_BASE + 0x1000) as u64,
1114            ))
1115            .unwrap(),
1116        };
1117        let key = PrivateFutexKey {
1118            addr: FutexAddress::try_from(UserAddress::from(
1119                (RESTRICTED_ASPACE_BASE + 0x2000) as u64,
1120            ))
1121            .unwrap(),
1122        };
1123
1124        {
1125            let dead_event = InterruptibleEvent::new();
1126            state.get_waiters_or_default(stale_key).add(FutexWaiter {
1127                mask: u32::MAX,
1128                notifiable: FutexNotifiable::new_internal(Arc::downgrade(&dead_event)),
1129            });
1130        } // dead_event is dropped here, so its waiter becomes stale.
1131
1132        // Looking for a waiter that is not there scans every queue, emptying the stale one.
1133        let event = InterruptibleEvent::new();
1134        assert!(!state.remove_waiter_from_queue(key, &WaiterMatcher::Event(&event)));
1135        assert_eq!(state.waiters.len(), 0, "the emptied queue should be collected");
1136    }
1137
1138    #[fuchsia::test]
1139    fn test_split_for_requeue_fairness() {
1140        let mut waiters = FutexWaiters::default();
1141        let e1 = Arc::new(InterruptibleEvent::new());
1142        let e2 = Arc::new(InterruptibleEvent::new());
1143        let e3 = Arc::new(InterruptibleEvent::new());
1144
1145        waiters.add(FutexWaiter {
1146            mask: 1,
1147            notifiable: FutexNotifiable::new_internal(Arc::downgrade(&e1)),
1148        });
1149        waiters.add(FutexWaiter {
1150            mask: 2,
1151            notifiable: FutexNotifiable::new_internal(Arc::downgrade(&e2)),
1152        });
1153        waiters.add(FutexWaiter {
1154            mask: 3,
1155            notifiable: FutexNotifiable::new_internal(Arc::downgrade(&e3)),
1156        });
1157
1158        let split = waiters.split_for_requeue(2);
1159
1160        assert_eq!(split.0.len(), 2);
1161        assert_eq!(split.0[0].mask, 1);
1162        assert_eq!(split.0[1].mask, 2);
1163
1164        assert_eq!(waiters.0.len(), 1);
1165        assert_eq!(waiters.0[0].mask, 3);
1166    }
1167
1168    #[fuchsia::test]
1169    fn test_stale_external_waiter_cleanup() {
1170        let mut state = FutexTableState::<PrivateFutexKey>::default();
1171        let key = PrivateFutexKey {
1172            addr: FutexAddress::try_from(UserAddress::from(
1173                (RESTRICTED_ASPACE_BASE + 0x1000) as u64,
1174            ))
1175            .unwrap(),
1176        };
1177
1178        {
1179            let token = Arc::new(());
1180            let (sender, _receiver) = oneshot::channel::<()>();
1181            state.get_waiters_or_default(key.clone()).add(FutexWaiter {
1182                mask: u32::MAX,
1183                notifiable: FutexNotifiable::new_external(Arc::downgrade(&token), sender),
1184            });
1185        } // token is dropped here, so it becomes stale
1186
1187        assert_eq!(state.waiters.len(), 1);
1188
1189        // Trigger a cleanup with a placeholder event
1190        let dummy_event = InterruptibleEvent::new();
1191        state.remove_waiter_from_queue(key, &WaiterMatcher::Event(&dummy_event));
1192
1193        assert_eq!(state.waiters.len(), 0, "Stale external waiter should be removed");
1194    }
1195
1196    /// `unlock_pi` must keep handing the mutex over after skipping a dead waiter.
1197    ///
1198    /// Each handoff rewrites the futex word, so a second attempt has to compare against what the
1199    /// first one wrote. Comparing against the value read on entry made the retry fail with
1200    /// EINVAL, leaving the mutex owned by a thread that no longer exists and dropping the waiter
1201    /// that should have got it.
1202    #[::fuchsia::test]
1203    async fn test_unlock_pi_hands_over_to_next_waiter_after_a_stale_one() {
1204        use crate::mm::memory::MemoryObject;
1205        use crate::mm::{DesiredAddress, MappingName, MappingOptions, PAGE_SIZE, ProtectionFlags};
1206        use crate::testing::spawn_kernel_and_run;
1207
1208        spawn_kernel_and_run(async move |current_task| {
1209            let mm = current_task.mm().unwrap();
1210            let addr = mm
1211                .map_memory(
1212                    DesiredAddress::Any,
1213                    Arc::new(MemoryObject::from(zx::Vmo::create(*PAGE_SIZE).unwrap())),
1214                    0,
1215                    *PAGE_SIZE as usize,
1216                    ProtectionFlags::READ | ProtectionFlags::WRITE,
1217                    MappingOptions::empty(),
1218                    MappingName::None,
1219                )
1220                .expect("map failed");
1221            let futex_addr = FutexAddress::try_from(addr).unwrap();
1222
1223            // The current task owns the mutex and has waiters queued behind it.
1224            let owner_tid = current_task.get_tid() as u32;
1225            assert!(matches!(
1226                mm.atomic_compare_exchange_u32_acq_rel(futex_addr, 0, owner_tid | FUTEX_WAITERS),
1227                CompareExchangeResult::Success
1228            ));
1229
1230            const STALE_TID: u32 = 0x111;
1231            const NEXT_OWNER_TID: u32 = 0x222;
1232            let futex_table = FutexTable::<PrivateFutexKey>::default();
1233            let key = PrivateFutexKey::get(current_task, futex_addr).unwrap();
1234            let next_owner_event = InterruptibleEvent::new();
1235            {
1236                let mut state = futex_table.state.lock();
1237                let queue = state.get_rt_mutex_waiters_or_default(key);
1238                {
1239                    let dead_event = InterruptibleEvent::new();
1240                    queue.push_back(RtMutexWaiter {
1241                        tid: STALE_TID,
1242                        notifiable: FutexNotifiable::new_internal(Arc::downgrade(&dead_event)),
1243                    });
1244                } // dead_event is dropped here, so its waiter can never be notified.
1245                queue.push_back(RtMutexWaiter {
1246                    tid: NEXT_OWNER_TID,
1247                    notifiable: FutexNotifiable::new_internal(Arc::downgrade(&next_owner_event)),
1248                });
1249            }
1250
1251            futex_table.unlock_pi(current_task, addr).expect("unlock_pi failed");
1252
1253            assert_eq!(
1254                mm.atomic_load_u32_relaxed(futex_addr).unwrap(),
1255                NEXT_OWNER_TID,
1256                "the mutex should have been handed to the waiter behind the stale one"
1257            );
1258            assert!(futex_table.state.lock().rt_mutex_waiters.is_empty());
1259        })
1260        .await;
1261    }
1262
1263    #[::fuchsia::test]
1264    async fn test_futex_deadlock_with_pager() {
1265        use crate::mm::memory::MemoryObject;
1266        use crate::mm::{DesiredAddress, MappingName, MappingOptions, PAGE_SIZE, ProtectionFlags};
1267        use crate::testing::spawn_kernel_and_run;
1268        use std::sync::atomic::{AtomicBool, Ordering};
1269        use zx::sys::zx_page_request_command_t::ZX_PAGER_VMO_READ;
1270
1271        spawn_kernel_and_run(async move |current_task| {
1272            let mm = current_task.mm().unwrap();
1273
1274            let port = Arc::new(zx::Port::create());
1275            let port_clone = port.clone();
1276            let pager =
1277                Arc::new(zx::Pager::create(zx::PagerOptions::empty()).expect("create failed"));
1278            let pager_clone = pager.clone();
1279
1280            let vmo = Arc::new(
1281                pager
1282                    .create_vmo(zx::VmoOptions::RESIZABLE, &port, 1, *PAGE_SIZE)
1283                    .expect("create_vmo failed"),
1284            );
1285            let vmo_clone = vmo.clone();
1286
1287            let mapped_addr = mm
1288                .map_memory(
1289                    DesiredAddress::Any,
1290                    Arc::new(MemoryObject::from(
1291                        (*vmo).duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap(),
1292                    )),
1293                    0,
1294                    *PAGE_SIZE as usize,
1295                    ProtectionFlags::READ | ProtectionFlags::WRITE,
1296                    MappingOptions::empty(),
1297                    MappingName::None,
1298                )
1299                .expect("map failed");
1300
1301            let futex_table = Arc::new(FutexTable::<PrivateFutexKey>::default());
1302            let futex_table_clone = futex_table.clone();
1303            let task_clone = current_task.task.clone();
1304            let dummy_addr = UserAddress::from((RESTRICTED_ASPACE_BASE + 0x2000) as u64);
1305
1306            let page_requested = Arc::new(AtomicBool::new(false));
1307            let wake_completed = Arc::new(AtomicBool::new(false));
1308
1309            let page_req_clone = page_requested.clone();
1310            let wake_completed_clone = wake_completed.clone();
1311
1312            let pager_thread = std::thread::spawn(move || {
1313                let packet = port_clone.wait(zx::MonotonicInstant::INFINITE).expect("wait failed");
1314                if let zx::PacketContents::Pager(contents) = packet.contents() {
1315                    if contents.command() == ZX_PAGER_VMO_READ {
1316                        let range = contents.range();
1317                        page_req_clone.store(true, Ordering::SeqCst);
1318
1319                        // Spawn waker thread while main thread is page-faulting before acquiring the lock
1320                        let waker = std::thread::spawn(move || {
1321                            let _ = futex_table_clone.wake(&task_clone, dummy_addr, 1, u32::MAX);
1322                            wake_completed_clone.store(true, Ordering::SeqCst);
1323                        });
1324
1325                        // Verify waker completes immediately without being blocked by the page fault
1326                        waker.join().unwrap();
1327                        assert!(
1328                            wake_completed.load(Ordering::SeqCst),
1329                            "Waker thread must NOT be blocked while page fault is in progress!"
1330                        );
1331
1332                        // Supply pages to unblock the main thread's page fault
1333                        let source_vmo =
1334                            zx::Vmo::create(range.end - range.start).expect("create failed");
1335                        pager_clone
1336                            .supply_pages(&vmo_clone, range, &source_vmo, 0)
1337                            .expect("supply_pages failed");
1338                    }
1339                }
1340            });
1341
1342            // This will lock the FutexTable, then page fault on mapped_addr when reading value
1343            let _ = futex_table.wait(
1344                current_task,
1345                mapped_addr,
1346                0,
1347                u32::MAX,
1348                zx::MonotonicInstant::from_nanos(1),
1349            );
1350
1351            pager_thread.join().unwrap();
1352        })
1353        .await;
1354    }
1355
1356    #[::fuchsia::test]
1357    async fn test_wait_expired_deadline_returns_etimedout_without_queuing() {
1358        use crate::mm::memory::MemoryObject;
1359        use crate::mm::{DesiredAddress, MappingName, MappingOptions, PAGE_SIZE, ProtectionFlags};
1360        use crate::testing::spawn_kernel_and_run;
1361
1362        spawn_kernel_and_run(async move |current_task| {
1363            let mm = current_task.mm().unwrap();
1364            let addr = mm
1365                .map_memory(
1366                    DesiredAddress::Any,
1367                    Arc::new(MemoryObject::from(zx::Vmo::create(*PAGE_SIZE).unwrap())),
1368                    0,
1369                    *PAGE_SIZE as usize,
1370                    ProtectionFlags::READ | ProtectionFlags::WRITE,
1371                    MappingOptions::empty(),
1372                    MappingName::None,
1373                )
1374                .expect("map failed");
1375
1376            let futex_table = FutexTable::<PrivateFutexKey>::default();
1377
1378            assert_eq!(
1379                futex_table.wait(
1380                    current_task,
1381                    addr,
1382                    0,
1383                    FUTEX_BITSET_MATCH_ANY,
1384                    zx::MonotonicInstant::ZERO,
1385                ),
1386                error!(ETIMEDOUT)
1387            );
1388            assert_eq!(
1389                futex_table.wait(
1390                    current_task,
1391                    addr,
1392                    0,
1393                    FUTEX_BITSET_MATCH_ANY,
1394                    zx::MonotonicInstant::get(),
1395                ),
1396                error!(ETIMEDOUT)
1397            );
1398            assert_eq!(
1399                futex_table.wait_boot(
1400                    current_task,
1401                    addr,
1402                    0,
1403                    FUTEX_BITSET_MATCH_ANY,
1404                    zx::BootInstant::ZERO,
1405                    zx::BootDuration::ZERO,
1406                ),
1407                error!(ETIMEDOUT)
1408            );
1409            assert_eq!(
1410                futex_table.wait_boot(
1411                    current_task,
1412                    addr,
1413                    0,
1414                    FUTEX_BITSET_MATCH_ANY,
1415                    zx::BootInstant::get(),
1416                    zx::BootDuration::ZERO,
1417                ),
1418                error!(ETIMEDOUT)
1419            );
1420            assert!(futex_table.state.lock().waiters.is_empty());
1421
1422            // A mismatched value must still report EAGAIN ahead of an expired deadline.
1423            assert_eq!(
1424                futex_table.wait(
1425                    current_task,
1426                    addr,
1427                    1,
1428                    FUTEX_BITSET_MATCH_ANY,
1429                    zx::MonotonicInstant::ZERO,
1430                ),
1431                error!(EAGAIN)
1432            );
1433            assert_eq!(
1434                futex_table.wait_boot(
1435                    current_task,
1436                    addr,
1437                    1,
1438                    FUTEX_BITSET_MATCH_ANY,
1439                    zx::BootInstant::ZERO,
1440                    zx::BootDuration::ZERO,
1441                ),
1442                error!(EAGAIN)
1443            );
1444            assert!(futex_table.state.lock().waiters.is_empty());
1445        })
1446        .await;
1447    }
1448}
starnix_core/mm/futex_table.rs

starnix_core/mm/
futex_table.rs
