feat(starry-kernel): support futex WAKE_OP

os/StarryOS/kernel/CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- *(starry-kernel)* support `FUTEX_WAKE_OP` in Starry futex syscall handling.
 - *(starry-kernel)* port LKM loader + cargo xtask starry kmod build ([#851](https://github.com/rcore-os/tgoskits/pull/851))
 - *(starryos)* expose K230 KPU device ([#1054](https://github.com/rcore-os/tgoskits/pull/1054))
 - *(starry-kernel)* implement child subreaper ([#1050](https://github.com/rcore-os/tgoskits/pull/1050))

os/StarryOS/kernel/src/mm/access.rs

@@ -2,9 +2,10 @@ use alloc::string::String;
 use core::{
     alloc::Layout,
     ffi::c_char,
-    hint::unlikely,
+    hint::{spin_loop, unlikely},
     mem::{MaybeUninit, transmute},
     ptr, slice, str,
+    sync::atomic::{AtomicU32, Ordering},
 };
 
 use ax_errno::{AxError, AxResult};
@@ -200,6 +201,36 @@ impl<T> UserPtr<T> {
     }
 }
 
+pub fn atomic_update_user_u32(
+    ptr: *mut u32,
+    mut update: impl FnMut(u32) -> AxResult<u32>,
+) -> AxResult<u32> {
+    check_region(
+        VirtAddr::from_ptr_of(ptr),
+        Layout::new::<u32>(),
+        MappingFlags::READ.union(MappingFlags::WRITE),
+    )?;
+
+    let ptr = ptr.cast::<AtomicU32>();
+    access_user_memory(|| {
+        loop {
+            // SAFETY: check_region() validated that the user address is a
+            // writable, properly aligned u32 in the current address space.
+            let old = unsafe { &*ptr }.load(Ordering::SeqCst);
+            let new = update(old)?;
+            match unsafe { &*ptr }.compare_exchange_weak(
+                old,
+                new,
+                Ordering::SeqCst,
+                Ordering::SeqCst,
+            ) {
+                Ok(_) => return Ok(old),
+                Err(_) => spin_loop(),
+            }
+        }
+    })
+}
+
 /// An immutable pointer to user space memory.
 #[repr(transparent)]
 #[derive(PartialEq, Clone, Copy)]

os/StarryOS/kernel/src/syscall/fs/io.rs

@@ -727,12 +727,12 @@ fn do_send(mut src: SendFile, mut dst: SendFile, len: usize) -> AxResult<usize>
             *pos += bytes_written as u64;
             user.vm_write(*pos)?;
         }
+        total_written += bytes_written;
+        remaining -= bytes_written;
+
         if bytes_written < bytes_read {
             break;
         }
-
-        total_written += bytes_written;
-        remaining -= bytes_written;
     }
 
     Ok(total_written)

os/StarryOS/kernel/src/syscall/sync/futex.rs

@@ -4,12 +4,15 @@ use ax_errno::{AxError, AxResult};
 use ax_runtime::hal::time::{TimeValue, monotonic_time, wall_time};
 use ax_task::current;
 use linux_raw_sys::general::{
-    FUTEX_CLOCK_REALTIME, FUTEX_CMP_REQUEUE, FUTEX_REQUEUE, FUTEX_WAIT, FUTEX_WAIT_BITSET,
-    FUTEX_WAKE, FUTEX_WAKE_BITSET, robust_list_head, timespec,
+    FUTEX_CLOCK_REALTIME, FUTEX_CMP_REQUEUE, FUTEX_OP_ADD, FUTEX_OP_ANDN, FUTEX_OP_CMP_EQ,
+    FUTEX_OP_CMP_GE, FUTEX_OP_CMP_GT, FUTEX_OP_CMP_LE, FUTEX_OP_CMP_LT, FUTEX_OP_CMP_NE,
+    FUTEX_OP_OPARG_SHIFT, FUTEX_OP_OR, FUTEX_OP_SET, FUTEX_OP_XOR, FUTEX_REQUEUE, FUTEX_WAIT,
+    FUTEX_WAIT_BITSET, FUTEX_WAKE, FUTEX_WAKE_BITSET, FUTEX_WAKE_OP, robust_list_head, timespec,
 };
 use starry_vm::{VmMutPtr, VmPtr};
 
 use crate::{
+    mm::atomic_update_user_u32,
     task::{AsThread, FutexKey, FutexKeyMode, futex_table_for, get_task},
     time::TimeValueLike,
 };
@@ -26,6 +29,7 @@ enum FutexCommand {
     WakeBitset,
     Requeue,
     CmpRequeue,
+    WakeOp,
 }
 
 struct ParsedFutexOp {
@@ -50,6 +54,61 @@ fn validate_futex_word(uaddr: *const u32) -> AxResult<()> {
     Ok(())
 }
 
+fn sign_extend_12(value: u32) -> i32 {
+    ((value << 20) as i32) >> 20
+}
+
+fn futex_wake_op_arg(raw_op: u32, encoded_op: u32) -> i32 {
+    let mut oparg = sign_extend_12((encoded_op >> 12) & 0xfff);
+    if raw_op & FUTEX_OP_OPARG_SHIFT != 0 {
+        oparg = (1u32 << ((oparg & 31) as u32)) as i32;
+    }
+    oparg
+}
+
+fn apply_futex_wake_op(old_value: u32, raw_op: u32, oparg: i32) -> AxResult<u32> {
+    let op = raw_op & !FUTEX_OP_OPARG_SHIFT;
+    let new_value = match op {
+        FUTEX_OP_SET => oparg as u32,
+        FUTEX_OP_ADD => (old_value as i32).wrapping_add(oparg) as u32,
+        FUTEX_OP_OR => old_value | oparg as u32,
+        FUTEX_OP_ANDN => old_value & !(oparg as u32),
+        FUTEX_OP_XOR => old_value ^ oparg as u32,
+        _ => return Err(AxError::Unsupported),
+    };
+    Ok(new_value)
+}
+
+fn compare_futex_wake_op(old_value: u32, raw_cmp: u32, cmparg: i32) -> AxResult<bool> {
+    let old_value = old_value as i32;
+    let matched = match raw_cmp {
+        FUTEX_OP_CMP_EQ => old_value == cmparg,
+        FUTEX_OP_CMP_NE => old_value != cmparg,
+        FUTEX_OP_CMP_LT => old_value < cmparg,
+        FUTEX_OP_CMP_LE => old_value <= cmparg,
+        FUTEX_OP_CMP_GT => old_value > cmparg,
+        FUTEX_OP_CMP_GE => old_value >= cmparg,
+        _ => return Err(AxError::Unsupported),
+    };
+    Ok(matched)
+}
+
+fn futex_atomic_op_in_user(uaddr: *mut u32, encoded_op: u32) -> AxResult<bool> {
+    if !uaddr.addr().is_multiple_of(align_of::<u32>()) {
+        return Err(AxError::InvalidInput);
+    }
+
+    let raw_op = (encoded_op >> 28) & 0xf;
+    let raw_cmp = (encoded_op >> 24) & 0xf;
+    let oparg = futex_wake_op_arg(raw_op, encoded_op);
+    let cmparg = sign_extend_12(encoded_op & 0xfff);
+
+    let old_value = atomic_update_user_u32(uaddr, |old_value| {
+        apply_futex_wake_op(old_value, raw_op, oparg)
+    })?;
+    compare_futex_wake_op(old_value, raw_cmp, cmparg)
+}
+
 fn parse_futex_op(futex_op: u32) -> AxResult<ParsedFutexOp> {
     let flags = futex_op & !FUTEX_COMMAND_MASK;
     if flags & !SUPPORTED_FLAGS != 0 {
@@ -63,10 +122,14 @@ fn parse_futex_op(futex_op: u32) -> AxResult<ParsedFutexOp> {
         FUTEX_WAKE_BITSET => FutexCommand::WakeBitset,
         FUTEX_REQUEUE => FutexCommand::Requeue,
         FUTEX_CMP_REQUEUE => FutexCommand::CmpRequeue,
+        FUTEX_WAKE_OP => FutexCommand::WakeOp,
         _ => return Err(AxError::Unsupported),
     };
 
     let clock_realtime = flags & FUTEX_CLOCK_REALTIME != 0;
+    if clock_realtime && command == FutexCommand::WakeOp {
+        return Err(AxError::Unsupported);
+    }
     if clock_realtime && !matches!(command, FutexCommand::Wait | FutexCommand::WaitBitset) {
         return Err(AxError::InvalidInput);
     }
@@ -219,6 +282,25 @@ pub fn sys_futex(
                 return Err(AxError::WouldBlock);
             };
 
+            if count > 0 {
+                ax_task::yield_now();
+            }
+            Ok(count as _)
+        }
+        FutexCommand::WakeOp => {
+            let wake_count = value as usize;
+            let wake2_count = timeout.addr();
+            validate_futex_word(uaddr)?;
+
+            let key2 = FutexKey::new_current(uaddr2.addr(), op.key_mode);
+            let table2 = futex_table_for(&key2);
+
+            let source = futex_table.get_or_insert(&key);
+            let target = table2.get_or_insert(&key2);
+            let count = source.wq.wake_op(wake_count, &target.wq, wake2_count, || {
+                futex_atomic_op_in_user(uaddr2, value3)
+            })?;
+
             if count > 0 {
                 ax_task::yield_now();
             }

os/StarryOS/kernel/src/task/futex.rs

@@ -195,26 +195,29 @@ impl WaitQueue {
         )))??
     }
 
+    fn wake_locked(queue: &mut VecDeque<Waiter>, count: usize, mask: u32, wakers: &mut Vec<Waker>) {
+        let base = wakers.len();
+        queue.retain(|waiter| {
+            if waiter.state.cancelled.load(AtomicOrdering::SeqCst) {
+                false
+            } else if wakers.len() - base >= count || (waiter.bitset & mask) == 0 {
+                true
+            } else {
+                waiter.state.woken.store(true, AtomicOrdering::SeqCst);
+                wakers.push(waiter.waker.clone());
+                false
+            }
+        });
+    }
+
     /// Wakes up at most `count` tasks whose bitset intersects with the given
     /// bitmask.
     pub fn wake(&self, count: usize, mask: u32) -> usize {
-        let wakers = {
+        let mut wakers = Vec::new();
+        {
             let mut inner = self.inner.lock();
-            let mut wakers = Vec::new();
-
-            inner.queue.retain(|waiter| {
-                if waiter.state.cancelled.load(AtomicOrdering::SeqCst) {
-                    false
-                } else if wakers.len() >= count || (waiter.bitset & mask) == 0 {
-                    true
-                } else {
-                    waiter.state.woken.store(true, AtomicOrdering::SeqCst);
-                    wakers.push(waiter.waker.clone());
-                    false
-                }
-            });
-            wakers
-        };
+            Self::wake_locked(&mut inner.queue, count, mask, &mut wakers);
+        }
 
         let woke = wakers.len();
         for waker in wakers {
@@ -223,6 +226,53 @@ impl WaitQueue {
         woke
     }
 
+    /// Serializes a FUTEX_WAKE_OP user RMW with both futex wait queues.
+    pub fn wake_op(
+        &self,
+        wake_count: usize,
+        target: &WaitQueue,
+        wake2_count: usize,
+        condition: impl FnOnce() -> AxResult<bool>,
+    ) -> AxResult<usize> {
+        let mut condition = Some(condition);
+        let mut wakers = Vec::new();
+
+        match core::ptr::from_ref(self).cmp(&core::ptr::from_ref(target)) {
+            Ordering::Less => {
+                let mut src = self.inner.lock();
+                let mut dst = target.inner.lock();
+                let wake_second = condition.take().expect("condition used once")()?;
+                Self::wake_locked(&mut src.queue, wake_count, u32::MAX, &mut wakers);
+                if wake_second {
+                    Self::wake_locked(&mut dst.queue, wake2_count, u32::MAX, &mut wakers);
+                }
+            }
+            Ordering::Greater => {
+                let mut dst = target.inner.lock();
+                let mut src = self.inner.lock();
+                let wake_second = condition.take().expect("condition used once")()?;
+                Self::wake_locked(&mut src.queue, wake_count, u32::MAX, &mut wakers);
+                if wake_second {
+                    Self::wake_locked(&mut dst.queue, wake2_count, u32::MAX, &mut wakers);
+                }
+            }
+            Ordering::Equal => {
+                let mut src = self.inner.lock();
+                let wake_second = condition.take().expect("condition used once")()?;
+                Self::wake_locked(&mut src.queue, wake_count, u32::MAX, &mut wakers);
+                if wake_second {
+                    Self::wake_locked(&mut src.queue, wake2_count, u32::MAX, &mut wakers);
+                }
+            }
+        }
+
+        let woke = wakers.len();
+        for waker in wakers {
+            waker.wake();
+        }
+        Ok(woke)
+    }
+
     fn wake_requeue_locked(
         src: &mut VecDeque<Waiter>,
         dst: &mut VecDeque<Waiter>,

test-suit/starryos/normal/qemu-smp1/syscall/test-futex-wake-op/c/CMakeLists.txt

@@ -0,0 +1,18 @@
+cmake_minimum_required(VERSION 3.20)
+project(test-futex-wake-op C)
+
+set(CMAKE_C_STANDARD 11)
+set(CMAKE_C_STANDARD_REQUIRED ON)
+set(CMAKE_C_EXTENSIONS OFF)
+
+add_executable(test-futex-wake-op src/main.c)
+target_include_directories(
+    test-futex-wake-op
+    PRIVATE
+    src
+    ${CMAKE_CURRENT_SOURCE_DIR}/../../common
+)
+target_compile_options(test-futex-wake-op PRIVATE -Wall -Wextra -Werror)
+target_link_libraries(test-futex-wake-op PRIVATE pthread)
+
+install(TARGETS test-futex-wake-op RUNTIME DESTINATION usr/bin/starry-test-suit)