rust-kernel

//! Futexes ("fast userspace mutexes"): the one syscall threads need to
//! sleep and wake each other.
//!
//! A lock in a C library is just an integer in memory. Taking a free lock
//! is one atomic instruction in user space -- no syscall at all. Only when
//! a thread has to *wait* does it ask the kernel:
//!
//!   futex(&word, FUTEX_WAIT, expected): "put me to sleep, but only if
//!                                        word still == expected"
//!   futex(&word, FUTEX_WAKE, n):        "wake up to n sleepers on word"
//!
//! The "only if still equal" check is the whole trick: the kernel does it
//! while holding its own lock, the same lock FUTEX_WAKE takes. So if the
//! lock holder releases the lock (changes the word) and calls WAKE between
//! our user-space check and our WAIT, either WAIT sees the changed word
//! and returns EAGAIN at once, or we're already queued and WAKE finds us.
//! No wake-up can get lost.
//!
//! The kernel keeps no state per futex word: just a list of sleepers,
//! each labelled with the word it sleeps on (its *key*). For memory private
//! to one process the key is (address space, virtual address); for
//! MAP_SHARED memory, which different processes may map at different
//! addresses, it's the physical address.
//!
//! pthread_join uses it too: the kernel writes 0 to the exiting thread's
//! tid word and does a FUTEX_WAKE on it (CLONE_CHILD_CLEARTID).

use crate::errno::Errno;
use crate::memory;
use crate::process::Process;
use crate::thread::{self, Tid, WaitQueue};
use alloc::vec::Vec;
use core::sync::atomic::{AtomicU64, Ordering};
use spin::Mutex;
use x86_64::instructions::interrupts::without_interrupts;

pub const FUTEX_WAIT: u64 = 0;
pub const FUTEX_WAKE: u64 = 1;
pub const FUTEX_REQUEUE: u64 = 3;
pub const FUTEX_CMP_REQUEUE: u64 = 4;
pub const FUTEX_WAIT_BITSET: u64 = 9;
pub const FUTEX_WAKE_BITSET: u64 = 10;
pub const FUTEX_PRIVATE_FLAG: u64 = 128;
pub const FUTEX_CLOCK_REALTIME: u64 = 256;
pub const FUTEX_BITSET_MATCH_ANY: u32 = u32::MAX;

/// Which word a sleeper waits on.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
enum Key {
    /// (level-4 page table = the address space, virtual address)
    Private(u64, u64),
    /// Physical address (MAP_SHARED memory).
    Shared(u64),
}

struct Waiter {
    id: u64,
    key: Key,
    bitset: u32,
    kthread: Tid,
    woken: bool,
}

/// All sleepers. Locked with interrupts off (read in wait conditions).
static WAITERS: Mutex<Vec<Waiter>> = Mutex::new(Vec::new());
/// Only used for its sleeping logic: wakers wake threads one by one.
static QUEUE: WaitQueue = WaitQueue::new();
static NEXT_ID: AtomicU64 = AtomicU64::new(1);

/// Look up the key of `uaddr` in `p`, then run `f(key, value)` with the
/// word's current value, read through the physical mapping.
///
/// Two locking rules, both learned the hard way (the stage 18
/// `threads_and_futexes` flake):
///
/// * Whoever holds the address-space lock may be in a TLB shootdown
///   (smp.rs), waiting for every other CPU running this process to
///   answer an interrupt. A CPU spinning for that lock with interrupts
///   off never answers, and both wait for each other forever (munmap in
///   one thread + a contended futex in another was enough). So we spin
///   with `lock_answering_shootdowns`, and keep interrupts off only
///   for the short part under WAITERS.
/// * `f` runs with interrupts off and should take WAITERS itself; the
///   word is read *inside* `f`'s critical section, see `wait`.
fn with_word<R>(p: &Process, uaddr: u64, private: bool, f: impl FnOnce(Key, &dyn Fn() -> u32) -> R) -> Result<R, Errno> {
    let mm = p.mm();
    let (frame, _) = memory::user_pte(mm.l4, uaddr).ok_or(Errno::EFAULT)?;
    let phys = frame.start_address().as_u64() + (uaddr & 0xfff);
    let shared = !private && mm.vmas().iter().any(|v| v.shared && (v.start..v.end).contains(&uaddr));
    let key = if shared { Key::Shared(phys) } else { Key::Private(mm.l4.start_address().as_u64(), uaddr) };
    // The mapping can't change while we hold `mm`, so `phys` stays valid.
    let read = move || unsafe { core::ptr::read_volatile(memory::phys_to_virt(x86_64::PhysAddr::new(phys)).as_ptr::<u32>()) };
    let result = without_interrupts(|| f(key, &read));
    drop(mm);
    Ok(result)
}

/// Check a futex address and fault its page in.
fn prepare(uaddr: u64) -> Result<(), Errno> {
    if uaddr % 4 != 0 {
        return Err(Errno::EINVAL);
    }
    if !crate::mm::user_access_ok(uaddr, 4, false) {
        return Err(Errno::EFAULT);
    }
    Ok(())
}

/// FUTEX_WAIT: sleep while *uaddr == val, until woken, `deadline` (PIT
/// tick) or a signal.
pub fn wait(p: &Process, uaddr: u64, val: u32, deadline: Option<u64>, bitset: u32, private: bool) -> Result<u64, Errno> {
    if bitset == 0 {
        return Err(Errno::EINVAL);
    }
    prepare(uaddr)?;
    let id = NEXT_ID.fetch_add(1, Ordering::Relaxed);
    // Compare and enqueue in one step under WAITERS (see the top). The
    // word must be read while WAITERS is held: a waker changes the word
    // first and then takes WAITERS, so it either sees us queued or we
    // see its new value.
    with_word(p, uaddr, private, |key, read| {
        let mut waiters = WAITERS.lock();
        if read() != val {
            return Err(Errno::EAGAIN);
        }
        waiters.push(Waiter { id, key, bitset, kthread: thread::current_id(), woken: false });
        Ok(())
    })??;
    let woken = || WAITERS.lock().iter().any(|w| w.id == id && w.woken).then_some(());
    let result = match deadline {
        Some(deadline) => QUEUE.wait_until_deadline_intr(deadline, woken),
        None => QUEUE.wait_until_intr(woken).map(Some),
    };
    // Leave the list; if a wake-up raced with the timeout/signal, it counts.
    let was_woken = without_interrupts(|| {
        let mut waiters = WAITERS.lock();
        let i = waiters.iter().position(|w| w.id == id).expect("futex waiter vanished");
        waiters.swap_remove(i).woken
    });
    match result {
        _ if was_woken => Ok(0),
        Ok(_) => Err(Errno::ETIMEDOUT),
        Err(_) => Err(Errno::EINTR),
    }
}

/// Wake up to `n` sleepers on `key` (matching `bitset`); move up to
/// `requeue` more to `to` instead. Returns how many were woken (+ moved).
fn wake_key(key: Key, n: u64, bitset: u32, requeue: Option<(Key, u64)>) -> u64 {
    let mut kthreads = Vec::new();
    let mut moved = 0;
    without_interrupts(|| {
        let mut waiters = WAITERS.lock();
        for w in waiters.iter_mut().filter(|w| w.key == key && !w.woken && w.bitset & bitset != 0) {
            if (kthreads.len() as u64) < n {
                w.woken = true;
                kthreads.push(w.kthread);
            } else if let Some((to, max)) = requeue {
                if moved < max {
                    w.key = to;
                    moved += 1;
                }
            }
        }
    });
    for &k in &kthreads {
        thread::wake(k);
    }
    kthreads.len() as u64 + moved
}

/// FUTEX_WAKE from the kernel (thread exit, CLONE_CHILD_CLEARTID).
pub fn wake(p: &Process, uaddr: u64, n: u64) -> u64 {
    match prepare(uaddr).and_then(|_| with_word(p, uaddr, false, |key, _| key)) {
        Ok(key) => wake_key(key, n, FUTEX_BITSET_MATCH_ANY, None),
        Err(_) => 0,
    }
}

/// Debugging aid: every sleeper (key, kernel thread, woken).
pub fn debug_dump() -> alloc::string::String {
    without_interrupts(|| {
        let mut out = alloc::string::String::new();
        for w in WAITERS.lock().iter() {
            out += &alloc::format!("  futex waiter {:?} kthread {} woken={}\n", w.key, w.kthread, w.woken);
        }
        out
    })
}

/// How many sleepers there are (tests: nothing left behind).
pub fn waiters() -> usize {
    without_interrupts(|| WAITERS.lock().len())
}

/// The futex syscall (202): futex(uaddr, op, val, timeout/val2, uaddr2, val3).
pub fn sys_futex(p: &Process, uaddr: u64, op: u64, val: u64, timeout: u64, uaddr2: u64, val3: u64) -> Result<u64, Errno> {
    let private = op & FUTEX_PRIVATE_FLAG != 0;
    let realtime = op & FUTEX_CLOCK_REALTIME != 0;
    let cmd = op & !(FUTEX_PRIVATE_FLAG | FUTEX_CLOCK_REALTIME);
    match cmd {
        FUTEX_WAIT | FUTEX_WAIT_BITSET => {
            let bitset = if cmd == FUTEX_WAIT { FUTEX_BITSET_MATCH_ANY } else { val3 as u32 };
            let deadline = match timeout {
                0 => None,
                ptr => {
                    let us = crate::syscall::read_timespec_us(ptr)?;
                    // FUTEX_WAIT: relative. FUTEX_WAIT_BITSET: an absolute
                    // time on CLOCK_MONOTONIC (or REALTIME with the flag).
                    let us = if cmd == FUTEX_WAIT {
                        us
                    } else {
                        let now = if realtime { crate::time::realtime_us() } else { crate::time::now_us() };
                        us.saturating_sub(now)
                    };
                    if us == 0 {
                        return Err(Errno::ETIMEDOUT);
                    }
                    Some(crate::time::deadline_after_us(us))
                }
            };
            wait(p, uaddr, val as u32, deadline, bitset, private)
        }
        FUTEX_WAKE | FUTEX_WAKE_BITSET => {
            let bitset = if cmd == FUTEX_WAKE { FUTEX_BITSET_MATCH_ANY } else { val3 as u32 };
            if bitset == 0 {
                return Err(Errno::EINVAL);
            }
            prepare(uaddr)?;
            let key = with_word(p, uaddr, private, |key, _| key)?;
            Ok(wake_key(key, val, bitset, None))
        }
        FUTEX_REQUEUE | FUTEX_CMP_REQUEUE => {
            prepare(uaddr)?;
            prepare(uaddr2)?;
            let (key, value) = with_word(p, uaddr, private, |key, read| (key, read()))?;
            if cmd == FUTEX_CMP_REQUEUE && value != val3 as u32 {
                return Err(Errno::EAGAIN);
            }
            let to = with_word(p, uaddr2, private, |key, _| key)?;
            // (val2 travels in the timeout argument's register.)
            Ok(wake_key(key, val, FUTEX_BITSET_MATCH_ANY, Some((to, timeout))))
        }
        _ => Err(Errno::ENOSYS),
    }
}