opcode: add Futex{Wake,Wait,Waitv} (#272)

* opcode: add Futex{Wake,Wait,Waitv}

These opcodes are available since Linux 6.7.

* Use different user data for each futex test

Author: oxalica

build.rs

@@ -13,6 +13,7 @@ fn main() {
 #include <linux/stat.h>
 #include <linux/openat2.h>
 #include <linux/io_uring.h>
+#include <linux/futex.h>
     "#;
 
     #[cfg(not(feature = "overwrite"))]
@@ -38,7 +39,7 @@ fn main() {
         .derive_default(true)
         .generate_comments(true)
         .use_core()
-        .allowlist_type("io_uring_.*|io_.qring_.*|__kernel_timespec|open_how")
+        .allowlist_type("io_uring_.*|io_.qring_.*|__kernel_timespec|open_how|futex_waitv")
         .allowlist_var("__NR_io_uring.*|IOSQE_.*|IORING_.*|IO_URING_.*|SPLICE_F_FD_IN_FIXED")
         .generate()
         .unwrap()

io-uring-test/src/main.rs

@@ -160,6 +160,11 @@ fn test<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
     tests::poll::test_eventfd_poll_remove_failed(&mut ring, &test)?;
     tests::poll::test_eventfd_poll_multi(&mut ring, &test)?;
 
+    // futex
+    tests::futex::test_futex_wait(&mut ring, &test)?;
+    tests::futex::test_futex_wake(&mut ring, &test)?;
+    tests::futex::test_futex_waitv(&mut ring, &test)?;
+
     // regression test
     tests::regression::test_issue154(&mut ring, &test)?;
 

io-uring-test/src/tests/futex.rs

@@ -0,0 +1,187 @@
+use crate::Test;
+use io_uring::types::FutexWaitV;
+use io_uring::{cqueue, opcode, squeue, IoUring};
+use std::sync::atomic::{AtomicU32, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+use std::{io, ptr, thread};
+
+// Not defined by libc.
+//
+// From: https://github.com/torvalds/linux/blob/v6.7/include/uapi/linux/futex.h#L63
+const FUTEX2_SIZE_U32: u32 = 2;
+
+const INIT_VAL: u32 = 0xDEAD_BEEF;
+
+fn syscall_futex(futex: *const u32, op: libc::c_int, val: u32) -> io::Result<i64> {
+    let ret = unsafe {
+        libc::syscall(
+            libc::SYS_futex,
+            futex,
+            op,
+            val,
+            ptr::null::<u8>(),
+            ptr::null::<u8>(),
+            0u32,
+        )
+    };
+    if ret >= 0 {
+        Ok(ret)
+    } else {
+        Err(io::Error::from_raw_os_error(-ret as _))
+    }
+}
+
+pub fn test_futex_wait<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
+    ring: &mut IoUring<S, C>,
+    test: &Test,
+) -> anyhow::Result<()> {
+    require!(
+        test;
+        test.probe.is_supported(opcode::FutexWait::CODE);
+    );
+
+    const USER_DATA: u64 = 0xDEAD_BEEF_DEAD_BEEF;
+
+    println!("test futex_wait");
+
+    let mut futex = INIT_VAL;
+
+    let futex_wait_e = opcode::FutexWait::new(
+        &futex,
+        INIT_VAL as u64,
+        // NB. FUTEX_BITSET_MATCH_ANY is signed. We are operating on 32-bit futex, thus this mask
+        // must be 32-bit. Converting directly from c_int to u64 will yield `u64::MAX`, which is
+        // invalid.
+        libc::FUTEX_BITSET_MATCH_ANY as u32 as u64,
+        FUTEX2_SIZE_U32,
+    );
+
+    unsafe {
+        let mut queue = ring.submission();
+        queue
+            .push(&futex_wait_e.build().user_data(USER_DATA).into())
+            .expect("queue is full");
+    }
+
+    ring.submit()?;
+    thread::sleep(Duration::from_millis(100));
+    assert_eq!(ring.completion().len(), 0);
+
+    futex += 1;
+    let ret = syscall_futex(&futex, libc::FUTEX_WAKE, 1)?;
+    assert_eq!(ret, 1);
+
+    ring.submit_and_wait(1)?;
+    let cqes: Vec<cqueue::Entry> = ring.completion().map(Into::into).collect();
+
+    assert_eq!(cqes.len(), 1);
+    assert_eq!(cqes[0].user_data(), USER_DATA);
+    assert_eq!(cqes[0].result(), 0);
+
+    Ok(())
+}
+
+pub fn test_futex_wake<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
+    ring: &mut IoUring<S, C>,
+    test: &Test,
+) -> anyhow::Result<()> {
+    require!(
+        test;
+        test.probe.is_supported(opcode::FutexWake::CODE);
+    );
+
+    const USER_DATA: u64 = 0xBEEF_DEAD_BEEF_DEAD;
+
+    println!("test futex_wake");
+
+    let futex = Arc::new(AtomicU32::new(INIT_VAL));
+
+    let wait_thread = std::thread::spawn({
+        let futex = futex.clone();
+        move || {
+            while futex.load(Ordering::Relaxed) == INIT_VAL {
+                let ret = syscall_futex(futex.as_ptr(), libc::FUTEX_WAIT, INIT_VAL);
+                assert_eq!(ret.unwrap(), 0);
+            }
+        }
+    });
+
+    thread::sleep(Duration::from_millis(100));
+    assert!(!wait_thread.is_finished());
+
+    futex.store(INIT_VAL + 1, Ordering::Relaxed);
+
+    let futex_wake_e = opcode::FutexWake::new(
+        futex.as_ptr(),
+        1,
+        // NB. See comments above for why it cannot be a single `as u64`.
+        libc::FUTEX_BITSET_MATCH_ANY as u32 as u64,
+        FUTEX2_SIZE_U32,
+    );
+    unsafe {
+        let mut queue = ring.submission();
+        queue
+            .push(&futex_wake_e.build().user_data(USER_DATA).into())
+            .expect("queue is full");
+    }
+    ring.submit_and_wait(1)?;
+    let cqes: Vec<cqueue::Entry> = ring.completion().map(Into::into).collect();
+    assert_eq!(cqes.len(), 1);
+    assert_eq!(cqes[0].user_data(), USER_DATA);
+    assert_eq!(cqes[0].result(), 1);
+
+    wait_thread.join().unwrap();
+
+    Ok(())
+}
+
+pub fn test_futex_waitv<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
+    ring: &mut IoUring<S, C>,
+    test: &Test,
+) -> anyhow::Result<()> {
+    require!(
+        test;
+        test.probe.is_supported(opcode::FutexWaitV::CODE);
+    );
+
+    const USER_DATA: u64 = 0xDEAD_BEEF_BEEF_DEAD;
+
+    println!("test futex_waitv");
+
+    const FUTEX_CNT: usize = 5;
+    const TRIGGER_IDX: usize = 3;
+
+    let mut futexes = [INIT_VAL; FUTEX_CNT];
+    let mut waitv = [FutexWaitV::default(); FUTEX_CNT];
+    for (futex, waitv) in futexes.iter().zip(&mut waitv) {
+        *waitv = FutexWaitV::new()
+            .val(INIT_VAL as u64)
+            .uaddr(std::ptr::from_ref(futex) as _)
+            .flags(FUTEX2_SIZE_U32);
+    }
+
+    let futex_waitv_e = opcode::FutexWaitV::new(waitv.as_ptr().cast(), waitv.len() as _);
+    unsafe {
+        let mut queue = ring.submission();
+        queue
+            .push(&futex_waitv_e.build().user_data(USER_DATA).into())
+            .expect("queue is full");
+    }
+
+    ring.submit()?;
+    thread::sleep(Duration::from_millis(100));
+    assert_eq!(ring.completion().len(), 0);
+
+    futexes[TRIGGER_IDX] += 1;
+    let ret = syscall_futex(&futexes[TRIGGER_IDX], libc::FUTEX_WAKE, 1)?;
+    assert_eq!(ret, 1);
+
+    ring.submit_and_wait(1)?;
+    let cqes: Vec<cqueue::Entry> = ring.completion().map(Into::into).collect();
+    assert_eq!(cqes.len(), 1);
+    assert_eq!(cqes[0].user_data(), USER_DATA);
+    assert_eq!(cqes[0].result(), TRIGGER_IDX as _);
+
+    Ok(())
+}

io-uring-test/src/tests/mod.rs

@@ -1,5 +1,6 @@
 pub mod cancel;
 pub mod fs;
+pub mod futex;
 pub mod net;
 pub mod poll;
 pub mod queue;

src/opcode.rs

@@ -1735,3 +1735,102 @@ opcode! {
         Entry(sqe)
     }
 }
+
+// === 6.7 ===
+
+opcode! {
+    /// Wait on a futex, like but not equivalant to `futex(2)`'s `FUTEX_WAIT_BITSET`.
+    ///
+    /// Wait on a futex at address `futex` and which still has the value `val` and with `futex2(2)`
+    /// flags of `futex_flags`. `musk` can be set to a specific bitset mask, which will be matched
+    /// by the waking side to decide who to wake up. To always get woken, an application may use
+    /// `FUTEX_BITSET_MATCH_ANY` (truncated to futex bits). `futex_flags` follows the `futex2(2)`
+    /// flags, not the `futex(2)` v1 interface flags. `flags` are currently unused and hence `0`
+    /// must be passed.
+    #[derive(Debug)]
+    pub struct FutexWait {
+        futex: { *const u32 },
+        val: { u64 },
+        mask: { u64 },
+        futex_flags: { u32 },
+        ;;
+        flags: u32 = 0
+    }
+
+    pub const CODE = sys::IORING_OP_FUTEX_WAIT;
+
+    pub fn build(self) -> Entry {
+        let FutexWait { futex, val, mask, futex_flags, flags } = self;
+
+        let mut sqe = sqe_zeroed();
+        sqe.opcode = Self::CODE;
+        sqe.fd = futex_flags as _;
+        sqe.__bindgen_anon_2.addr = futex as usize as _;
+        sqe.__bindgen_anon_1.off = val;
+        unsafe { sqe.__bindgen_anon_6.__bindgen_anon_1.as_mut().addr3 = mask };
+        sqe.__bindgen_anon_3.futex_flags = flags;
+        Entry(sqe)
+    }
+}
+
+opcode! {
+    /// Wake up waiters on a futex, like but not equivalant to `futex(2)`'s `FUTEX_WAKE_BITSET`.
+    ///
+    /// Wake any waiters on the futex indicated by `futex` and at most `val` futexes. `futex_flags`
+    /// indicates the `futex2(2)` modifier flags. If a given bitset for who to wake is desired,
+    /// then that must be set in `mask`. Use `FUTEX_BITSET_MATCH_ANY` (truncated to futex bits) to
+    /// match any waiter on the given futex. `flags` are currently unused and hence `0` must be
+    /// passed.
+    #[derive(Debug)]
+    pub struct FutexWake {
+        futex: { *const u32 },
+        val: { u64 },
+        mask: { u64 },
+        futex_flags: { u32 },
+        ;;
+        flags: u32 = 0
+    }
+
+    pub const CODE = sys::IORING_OP_FUTEX_WAKE;
+
+    pub fn build(self) -> Entry {
+        let FutexWake { futex, val, mask, futex_flags, flags } = self;
+
+        let mut sqe = sqe_zeroed();
+        sqe.opcode = Self::CODE;
+        sqe.fd = futex_flags as _;
+        sqe.__bindgen_anon_2.addr = futex as usize as _;
+        sqe.__bindgen_anon_1.off = val;
+        unsafe { sqe.__bindgen_anon_6.__bindgen_anon_1.as_mut().addr3 = mask };
+        sqe.__bindgen_anon_3.futex_flags = flags;
+        Entry(sqe)
+    }
+}
+
+opcode! {
+    /// Wait on multiple futexes.
+    ///
+    /// Wait on multiple futexes at the same time. Futexes are given by `futexv` and `nr_futex` is
+    /// the number of futexes in that array. Unlike `FutexWait`, the desired bitset mask and values
+    /// are passed in `futexv`. `flags` are currently unused and hence `0` must be passed.
+    #[derive(Debug)]
+    pub struct FutexWaitV {
+        futexv: { *const types::FutexWaitV },
+        nr_futex: { u32 },
+        ;;
+        flags: u32 = 0
+    }
+
+    pub const CODE = sys::IORING_OP_FUTEX_WAITV;
+
+    pub fn build(self) -> Entry {
+        let FutexWaitV { futexv, nr_futex, flags } = self;
+
+        let mut sqe = sqe_zeroed();
+        sqe.opcode = Self::CODE;
+        sqe.__bindgen_anon_2.addr = futexv as usize as _;
+        sqe.len = nr_futex;
+        sqe.__bindgen_anon_3.futex_flags = flags;
+        Entry(sqe)
+    }
+}

src/sys/sys.rs

@@ -2766,3 +2766,66 @@ fn bindgen_test_layout_io_uring_recvmsg_out() {
         )
     );
 }
+#[repr(C)]
+#[derive(Debug, Default, Copy, Clone)]
+pub struct futex_waitv {
+    pub val: __u64,
+    pub uaddr: __u64,
+    pub flags: __u32,
+    pub __reserved: __u32,
+}
+#[test]
+fn bindgen_test_layout_futex_waitv() {
+    const UNINIT: ::core::mem::MaybeUninit<futex_waitv> = ::core::mem::MaybeUninit::uninit();
+    let ptr = UNINIT.as_ptr();
+    assert_eq!(
+        ::core::mem::size_of::<futex_waitv>(),
+        24usize,
+        concat!("Size of: ", stringify!(futex_waitv))
+    );
+    assert_eq!(
+        ::core::mem::align_of::<futex_waitv>(),
+        8usize,
+        concat!("Alignment of ", stringify!(futex_waitv))
+    );
+    assert_eq!(
+        unsafe { ::core::ptr::addr_of!((*ptr).val) as usize - ptr as usize },
+        0usize,
+        concat!(
+            "Offset of field: ",
+            stringify!(futex_waitv),
+            "::",
+            stringify!(val)
+        )
+    );
+    assert_eq!(
+        unsafe { ::core::ptr::addr_of!((*ptr).uaddr) as usize - ptr as usize },
+        8usize,
+        concat!(
+            "Offset of field: ",
+            stringify!(futex_waitv),
+            "::",
+            stringify!(uaddr)
+        )
+    );
+    assert_eq!(
+        unsafe { ::core::ptr::addr_of!((*ptr).flags) as usize - ptr as usize },
+        16usize,
+        concat!(
+            "Offset of field: ",
+            stringify!(futex_waitv),
+            "::",
+            stringify!(flags)
+        )
+    );
+    assert_eq!(
+        unsafe { ::core::ptr::addr_of!((*ptr).__reserved) as usize - ptr as usize },
+        20usize,
+        concat!(
+            "Offset of field: ",
+            stringify!(futex_waitv),
+            "::",
+            stringify!(__reserved)
+        )
+    );
+}