rust-kernel

@@ -67,17 +67,32 @@ static WAITERS: Mutex<Vec<Waiter>> = Mutex::new(Vec::new()); static QUEUE: WaitQueue = WaitQueue::new(); static NEXT_ID: AtomicU64 = AtomicU64::new(1); -/// Look up the key of `uaddr` in `p` and read the word there (through the-/// physical mapping, with the address space locked so it can't change-/// under us). The page must have been faulted in.-fn key_and_value(p: &Process, uaddr: u64, private: bool) -> Result<(Key, u32), Errno> {-    let mm = p.mm.lock();+/// 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 value = unsafe { core::ptr::read_volatile(memory::phys_to_virt(x86_64::PhysAddr::new(phys)).as_ptr::<u32>()) };     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) };-    Ok((key, value))+    // 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.@@ -99,16 +114,18 @@ pub fn wait(p: &Process, uaddr: u64, val: u32, deadline: Option<u64>, bitset: u3     }     prepare(uaddr)?;     let id = NEXT_ID.fetch_add(1, Ordering::Relaxed);-    // Compare and enqueue in one step under WAITERS (see the top).-    without_interrupts(|| {-        let (key, value) = key_and_value(p, uaddr, private)?;+    // 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 value != val {+        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),@@ -154,12 +171,23 @@ fn wake_key(key: Key, n: u64, bitset: u32, requeue: Option<(Key, u64)>) -> u64 {  /// 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(|_| without_interrupts(|| key_and_value(p, uaddr, false))) {-        Ok((key, _)) => wake_key(key, n, FUTEX_BITSET_MATCH_ANY, None),+    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())@@ -199,17 +227,17 @@ pub fn sys_futex(p: &Process, uaddr: u64, op: u64, val: u64, timeout: u64, uaddr                 return Err(Errno::EINVAL);             }             prepare(uaddr)?;-            let (key, _) = without_interrupts(|| key_and_value(p, uaddr, private))?;+            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) = without_interrupts(|| key_and_value(p, uaddr, private))?;+            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, _) = without_interrupts(|| key_and_value(p, uaddr2, private))?;+            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))))         }

@@ -191,7 +191,7 @@ fn page_fault(frame: &mut TrapFrame) {         let exec = error.contains(PageFaultErrorCode::INSTRUCTION_FETCH);         let p = crate::thread::current_process().expect("user fault without a process");         let result = if (crate::memory::USER_START..crate::memory::USER_END).contains(&addr) {-            p.mm.lock().handle_fault(addr, write, exec)+            p.mm().handle_fault(addr, write, exec)         } else {             Err(Fault::Unmapped) // a kernel address         };@@ -221,7 +221,7 @@ fn page_fault(frame: &mut TrapFrame) {                 x86_64::instructions::interrupts::enable(); // as it was             }             let write = error.contains(PageFaultErrorCode::CAUSED_BY_WRITE);-            let mut mm = p.mm.lock();+            let mut mm = p.mm();             if mm.handle_fault(addr, write, false).is_ok() {                 return;             }

@@ -540,7 +540,7 @@ fn flush_all() { /// kernel read or overwrite kernel memory for it. pub fn user_access_ok(addr: u64, len: u64, write: bool) -> bool {     match crate::thread::current_process() {-        Some(p) => p.mm.lock().prepare(addr, len, write),+        Some(p) => p.mm().prepare(addr, len, write),         None => false, // kernel threads have no user memory     } }

@@ -506,7 +506,7 @@ pub fn spawn_image(path: &str, elf_image: &[u8], spawn: SpawnArgs) -> Result<Arc fn start_process(process: Process, start: TrapFrame, fs_base: u64, blocked: u64, fpu: FpuState, altstack: signal::AltStack, comm: [u8; 16]) -> Arc<Process> {     let process = Arc::new(process);     without_interrupts(|| PROCESSES.lock().insert(process.pid, process.clone()));-    start_task(&process, process.pid, start, fs_base, blocked, fpu, altstack, comm);+    start_task(&process, process.pid, start, fs_base, blocked, fpu, altstack, comm, 0);     process } @@ -514,13 +514,13 @@ fn start_process(process: Process, start: TrapFrame, fs_base: u64, blocked: u64, /// `fpu`: its FPU/SSE registers (a copy of the creator's, or clean). /// `altstack`: its sigaltstack (fork keeps the caller's); `comm`: its name. #[allow(clippy::too_many_arguments)]-fn start_task(process: &Arc<Process>, tid: u64, start: TrapFrame, fs_base: u64, blocked: u64, fpu: FpuState, altstack: signal::AltStack, comm: [u8; 16]) -> Arc<Task> {+fn start_task(process: &Arc<Process>, tid: u64, start: TrapFrame, fs_base: u64, blocked: u64, fpu: FpuState, altstack: signal::AltStack, comm: [u8; 16], clear_child_tid: u64) -> Arc<Task> {     let task = Arc::new(Task {         tid: AtomicU64::new(tid),         process: process.clone(),         kthread: AtomicUsize::new(usize::MAX),         fs_base: AtomicU64::new(fs_base),-        clear_child_tid: AtomicU64::new(0),+        clear_child_tid: AtomicU64::new(clear_child_tid),         start: Mutex::new(Some(start)),         killed: AtomicBool::new(false),         sig: Mutex::new(ThreadSig::with_altstack(blocked, altstack)),@@ -530,7 +530,7 @@ fn start_task(process: &Arc<Process>, tid: u64, start: TrapFrame, fs_base: u64,          cpu_ticks: AtomicU64::new(0),     });     without_interrupts(|| process.tasks.lock().push(task.clone()));-    let l4 = process.mm.lock().l4;+    let l4 = process.mm().l4;     thread::spawn_with(&process.name(), user_thread_entry, l4, Some(task.clone()));     task }@@ -652,6 +652,14 @@ pub fn load_program(mm: &mut Mm, image: &[u8], argv: &[&[u8]], envp: &[&[u8]], e }  impl Process {+    /// Lock the address space. Always through here, not `mm.lock()`: it+    /// answers TLB shootdowns while it waits (see+    /// `smp::lock_answering_shootdowns`), so waiting for it can't deadlock+    /// with a thread that holds it and is shooting down our TLB.+    pub fn mm(&self) -> spin::MutexGuard<'_, Mm> {+        crate::smp::lock_answering_shootdowns(&self.mm)+    }+     /// The open file behind `fd`.     pub fn file(&self, fd: u64) -> Result<Arc<dyn File>, Errno> {         self.fds.lock().get(fd)@@ -1194,7 +1202,7 @@ pub fn fork(parent: &Arc<Process>, frame: &TrapFrame, stack: u64, vfork: bool) -     }     // The heart of it: copy the page tables, sharing the frames     // copy-on-write (mm.rs).-    let mm = parent.mm.lock().fork().ok_or(Errno::ENOMEM)?;+    let mm = parent.mm().fork().ok_or(Errno::ENOMEM)?;     let mut start = *frame;     start.rax = 0; // "you're the child"     if stack != 0 {@@ -1283,8 +1291,10 @@ pub fn clone_thread(p: &Arc<Process>, frame: &TrapFrame, stack: u64, tls: Option     // A new thread has no sigaltstack (its creator's stack isn't its own)     // and starts with its creator's name.     let comm = *me.comm.lock();-    let task = start_task(p, tid, start, fs_base, blocked, crate::fpu::save_live(), signal::AltStack::DISABLED, comm);-    task.clear_child_tid.store(clear_tid, Ordering::Relaxed);+    // clear_tid must be in place *before* the thread can run: a short+    // thread may finish on another CPU before start_task even returns, and+    // without it nobody would wake pthread_join (the stage 18 flake).+    start_task(p, tid, start, fs_base, blocked, crate::fpu::save_live(), signal::AltStack::DISABLED, comm, clear_tid);     serial_println!("[process] pid {} new thread {}", p.pid, tid);     Ok(tid) }@@ -1322,7 +1332,7 @@ pub fn execve(p: &Arc<Process>, frame: &mut TrapFrame, path: &str, argv: &[Vec<u     // --- point of no return ---     kill_other_threads(p);     let loaded = {-        let mut mm = p.mm.lock();+        let mut mm = p.mm();         mm.clear();         load_program(&mut mm, &image, &args, &env, path)     };

@@ -103,6 +103,25 @@ pub fn tlb_shootdown(cr3: u64) {     } } +/// Lock `m`, answering TLB shootdowns while we wait. The holder of an+/// address-space lock may be inside `tlb_shootdown`, waiting for *this*+/// CPU to flush; if we spun with interrupts off we would never see its+/// interrupt and both CPUs would wait forever. Use this for any lock a+/// shooter may hold when the caller might have interrupts off.+pub fn lock_answering_shootdowns<T>(m: &spin::Mutex<T>) -> spin::MutexGuard<'_, T> {+    loop {+        if let Some(guard) = m.try_lock() {+            return guard;+        }+        without_interrupts(|| {+            if FLUSH_PENDING[cpu_id()].load(Ordering::SeqCst) {+                handle_tlb_shootdown();+            }+        });+        core::hint::spin_loop();+    }+}+ /// The shootdown IPI arrived (interrupts.rs): forget all user translations. pub fn handle_tlb_shootdown() {     use x86_64::registers::control::Cr3;

@@ -674,7 +674,7 @@ extern "C" fn syscall_dispatch(frame: &mut TrapFrame) {         SYS_MMAP => current().and_then(|p| sys_mmap(&p, a1, a2, a3, a4, a5, frame.r9)),         SYS_MPROTECT => current().and_then(|p| sys_mprotect(&p, a1, a2, a3)),         SYS_MUNMAP => current().and_then(|p| sys_munmap(&p, a1, a2)),-        SYS_BRK => current().map(|p| p.mm.lock().set_brk(a1)),+        SYS_BRK => current().map(|p| p.mm().set_brk(a1)),         SYS_RT_SIGACTION => sys_rt_sigaction(a1, a2, a3, a4),         SYS_RT_SIGPROCMASK => sys_rt_sigprocmask(a1, a2, a3, a4),         SYS_IOCTL => sys_ioctl(a1, a2, a3),@@ -1891,7 +1891,7 @@ fn sys_mmap(p: &process::Process, addr: u64, len: u64, prot: u64, flags: u64, fd         let inode = p.file(fd)?.inode().ok_or(Errno::ENODEV)?;         Some((inode, offset))     };-    let mut mm = p.mm.lock();+    let mut mm = p.mm();     let start = if flags & MAP_FIXED != 0 {         if addr % 4096 != 0 || !Mm::range_ok(addr, addr.saturating_add(len)) {             return Err(Errno::EINVAL);@@ -1926,7 +1926,7 @@ fn sys_munmap(p: &process::Process, addr: u64, len: u64) -> Result<u64, Errno> {     if len == 0 {         return Err(Errno::EINVAL);     }-    p.mm.lock().unmap(start, end);+    p.mm().unmap(start, end);     Ok(0) } @@ -1938,7 +1938,7 @@ fn sys_mprotect(p: &process::Process, addr: u64, len: u64, prot: u64) -> Result<     if start == end {         return Ok(0);     }-    p.mm.lock().protect(start, end, prot).map(|_| 0)+    p.mm().protect(start, end, prot).map(|_| 0) }  // --- Stage 16: libc batch 1 -------------------------------------------------@@ -1967,7 +1967,7 @@ fn sys_madvise(addr: u64, len: u64, advice: u64) -> Result<u64, Errno> {         return Err(Errno::ENOMEM);     }     if discard {-        current()?.mm.lock().discard(addr, end)?;+        current()?.mm().discard(addr, end)?;     }     Ok(0) }

@@ -502,7 +502,7 @@ fn sys_getrusage(who: i64, buf: u64) -> Result<u64, Errno> {     let (ss, su) = tv(system);     b.u64(0, us).u64(8, uu).u64(16, ss).u64(24, su);     if who != -1 {-        let resident = p.mm.lock().usage().0;+        let resident = p.mm().usage().0;         b.u64(32, resident * 4);     }     put(buf, &b.0)?;

@@ -534,6 +534,22 @@ pub fn join(id: Tid) {     THREAD_EXITED.wait_until(|| (!lock().threads.contains_key(&id)).then_some(())); } +/// Debugging aid: one line per thread (id, name, state, on_cpu, woken),+/// plus each CPU's current thread and the ready queue. Uses `try_lock` so+/// it still says something if the scheduler lock is stuck.+pub fn debug_dump() -> String {+    use core::fmt::Write;+    interrupts::without_interrupts(|| {+        let mut out = String::new();+        let Some(s) = (0..1_000_000).find_map(|_| SCHEDULER.try_lock()) else { return String::from("scheduler lock is held") };+        for (id, t) in &s.threads {+            let _ = writeln!(out, "  thread {} '{}' {:?} on_cpu={} woken={} idle={}", id, t.name, t.state, t.on_cpu, t.woken, t.idle);+        }+        let _ = writeln!(out, "  current {:?} ready {:?}", &s.current[..4], s.ready);+        out+    })+}+ /// Number of threads (including idle threads and "main"). pub fn thread_count() -> usize {     interrupts::without_interrupts(|| lock().threads.len())

@@ -0,0 +1,80 @@+//! Stage 19: run the `threads` test program 40 times in a row.+//!+//! It used to hang about once in 40 runs (the "threads_and_futexes flake").+//! Three kernel bugs were behind it, all races between CPUs, so they only+//! showed up now and then:+//!+//! 1. clone() armed CLONE_CHILD_CLEARTID *after* the new thread was already+//!    running. A thread that finished quickly exited before that, nobody+//!    woke its joiner, and `join` waited forever.+//! 2. A futex call spun on the address-space lock with interrupts off,+//!    while the thread holding that lock (munmap of a finished thread's+//!    stack) waited for *this* CPU to answer a TLB-shootdown interrupt.+//!    Each waited for the other: a deadlock.+//! 3. futex_wait read the futex word before taking the waiter-list lock,+//!    so a wake in between could be missed.+//!+//! A watchdog thread fails the test (with a dump of every thread and+//! futex waiter) if no run finishes for 8 seconds, instead of letting the+//! whole test hang until its timeout.+#![no_std]+#![no_main]+#![feature(custom_test_frameworks)]+#![test_runner(rustbelt::test_runner)]+#![reexport_test_harness_main = "test_main"]++use bootloader::{BootInfo, entry_point};+use core::panic::PanicInfo;+use core::sync::atomic::{AtomicU64, Ordering};+use rustbelt::process::{self, ExitStatus, SpawnArgs};+use rustbelt::{fs, futex, serial_println, thread};++entry_point!(main);++/// How many runs have finished.+static PROGRESS: AtomicU64 = AtomicU64::new(0);+const RUNS: u64 = 40;++fn main(boot_info: &'static BootInfo) -> ! {+    rustbelt::init();+    rustbelt::init_memory(boot_info);+    rustbelt::init_smp();+    fs::init();+    rustbelt::init_storage();+    rustbelt::init_net();+    test_main();+    rustbelt::hlt_loop();+}++fn watchdog() {+    let mut last = 0;+    loop {+        thread::sleep_ticks(18 * 8); // ~8 s (one tick is ~55 ms)+        let now = PROGRESS.load(Ordering::SeqCst);+        if now == RUNS {+            return;+        }+        if now == last {+            serial_println!("\n[watchdog] stuck in run {}\n{}{}", now + 1, thread::debug_dump(), futex::debug_dump());+            rustbelt::exit_qemu(rustbelt::QemuExitCode::Failed);+        }+        last = now;+    }+}++#[test_case]+fn threads_forty_times() {+    thread::spawn("watchdog", watchdog);+    for i in 0..RUNS {+        let p = process::spawn_path("/mnt/disk0/bin/threads", SpawnArgs::kernel("threads")).expect("spawn threads");+        let status = process::wait(&p);+        assert_eq!(status, ExitStatus::Exited(0), "run {} failed", i + 1);+        PROGRESS.store(i + 1, Ordering::SeqCst);+    }+    serial_println!("all {} runs ok", RUNS);+}++#[panic_handler]+fn panic(info: &PanicInfo) -> ! {+    rustbelt::test_panic_handler(info)+}