kernel: implement FUTEX_REQUEUE operation

Added FUTEX_REQUEUE (opcode 2) to the futex syscall implementation.
Cross-referenced with Linux 7.1 kernel/futex/requeue.c futex_requeue().

Operation: wake up to  waiters on the primary futex, and
requeue up to  waiters to a secondary futex at addr2.
This avoids the thundering herd problem in pthread condition
variable broadcast: instead of waking all waiters and having
them immediately contend, most are moved to a private futex
where they wake one at a time.

Previously the futex syscall would return EINVAL for FUTEX_REQUEUE,
breaking pthread_cond_broadcast on contended condition variables.

src/syscall/futex.rs

@@ -71,7 +71,7 @@ pub fn futex(
     op: usize,
     val: usize,
     val2: usize,
-    _addr2: usize,
+    addr2: usize,
     token: &mut CleanLockToken,
 ) -> Result<usize> {
     let current_addrsp = AddrSpace::current()?;
@@ -217,6 +217,63 @@ pub fn futex(
 
             Ok(woken)
         }
+        FUTEX_REQUEUE => {
+            // Linux 7.1 kernel/futex/requeue.c futex_requeue().
+            // Wake up to val waiters on primary futex, requeue up to
+            // val2 waiters to secondary futex (addr2).
+            let mut woken = 0;
+            let mut requeued = 0;
+
+            let target_virtaddr2 = VirtualAddress::new(addr2);
+            let target_physaddr2 = validate_and_translate_virt(&addr_space_guard, target_virtaddr2)
+                .ok_or(Error::new(EFAULT))?;
+
+            {
+                drop(addr_space_guard);
+                let mut futexes_map = FUTEXES.lock(token.token());
+                let (futexes_map, mut token) = futexes_map.token_split();
+
+                if let Some(futexes) = futexes_map.get_mut(&target_physaddr) {
+                    let mut i = 0;
+                    let current_addrsp_weak = Arc::downgrade(&current_addrsp);
+                    let mut to_requeue = Vec::new();
+
+                    while i < futexes.len() {
+                        let futex = unsafe { futexes.get_unchecked_mut(i) };
+                        if futex.target_virtaddr != target_virtaddr
+                            || !current_addrsp_weak.ptr_eq(&futex.addr_space)
+                        {
+                            i += 1;
+                            continue;
+                        }
+                        if woken < val {
+                            futex.context_lock.write(token.token()).unblock();
+                            futexes.remove(i);
+                            woken += 1;
+                        } else if requeued < val2 {
+                            let entry = futexes.remove(i);
+                            to_requeue.push(entry);
+                            requeued += 1;
+                        } else {
+                            i += 1;
+                        }
+                    }
+
+                    if futexes.is_empty() {
+                        futexes_map.remove(&target_physaddr);
+                    }
+
+                    if !to_requeue.is_empty() {
+                        futexes_map
+                            .entry(target_physaddr2)
+                            .or_insert_with(Vec::new)
+                            .extend(to_requeue);
+                    }
+                }
+            }
+
+            Ok(woken + requeued)
+        }
         _ => Err(Error::new(EINVAL)),
     }
 }