Add missing ops up to Linux 6.15

Cargo.toml

@@ -19,11 +19,16 @@ members = [ "io-uring-test", "io-uring-bench" ]
 
 [dependencies]
 bitflags = { version = "2.4.0", default-features = false }
-rustix = { version = "1.0.2", default-features = false, features = ["io_uring", "mm", "thread"] }
+linux-raw-sys = { version = "0.10.0", default-features = false }
+rustix = { version = "1.0.2", default-features = false, features = ["io_uring", "mm", "process", "thread"] }
 
 [dev-dependencies]
 libc = "0.2.98"
 anyhow = "1"
 rustix = "1.0.2"
 socket2 = "0.5"
 slab = "0.4"
+
+[patch.crates-io]
+linux-raw-sys = { git = "https://github.com/silvanshade/linux-raw-sys", branch = "io_uring-zcrx" }
+rustix = { git = "https://github.com/silvanshade/rustix", branch = "linux-6.15-io_uring" }

io-uring-test/Cargo.toml

@@ -10,7 +10,7 @@ publish = false
 [dependencies]
 io-uring = { path = "..", package = "rustix-uring" }
 libc = { version = "0.2", features = [ "extra_traits" ] }
-rustix = { version = "1.0.2", features = ["fs"] }
+rustix = { version = "1.0.2", features = ["fs", "pipe"] }
 anyhow = "1"
 tempfile = "3"
 once_cell = "1"

io-uring-test/src/main.rs

@@ -89,8 +89,16 @@ fn test<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
     tests::cancel::test_async_cancel_fd(&mut ring, &test)?;
     tests::cancel::test_async_cancel_fd_all(&mut ring, &test)?;
 
+    // epoll
+    tests::epoll::test_ready(&mut ring, &test)?;
+    tests::epoll::test_not_ready(&mut ring, &test)?;
+    tests::epoll::test_delete(&mut ring, &test)?;
+    tests::epoll::test_remove(&mut ring, &test)?;
+    tests::epoll::test_race(&mut ring, &test)?;
+
     // fs
     tests::fs::test_file_write_read(&mut ring, &test)?;
+    tests::fs::test_pipe_read_multishot(&mut ring, &test)?;
     tests::fs::test_file_writev_readv(&mut ring, &test)?;
     tests::fs::test_file_cur_pos(&mut ring, &test)?;
     tests::fs::test_file_fsync(&mut ring, &test)?;
@@ -106,6 +114,9 @@ fn test<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
     tests::fs::test_file_splice(&mut ring, &test)?;
     tests::fs::test_ftruncate(&mut ring, &test)?;
     tests::fs::test_fixed_fd_install(&mut ring, &test)?;
+    tests::fs::test_get_set_xattr(&mut ring, &test)?;
+    tests::fs::test_f_get_set_xattr(&mut ring, &test)?;
+    tests::fs::test_pipe_fixed_writev_readv(&mut ring, &test)?;
 
     // timeout
     tests::timeout::test_timeout(&mut ring, &test)?;
@@ -139,6 +150,7 @@ fn test<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
 
     tests::net::test_tcp_shutdown(&mut ring, &test)?;
     tests::net::test_socket(&mut ring, &test)?;
+    tests::net::test_socket_bind_listen(&mut ring, &test)?;
     tests::net::test_udp_recvmsg_multishot(&mut ring, &test)?;
     tests::net::test_udp_recvmsg_multishot_trunc(&mut ring, &test)?;
     tests::net::test_udp_send_with_dest(&mut ring, &test)?;
@@ -155,6 +167,9 @@ fn test<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
     tests::futex::test_futex_wake(&mut ring, &test)?;
     tests::futex::test_futex_waitv(&mut ring, &test)?;
 
+    // wait
+    tests::waitid::test_waitid(&mut ring, &test)?;
+
     // regression test
     tests::regression::test_issue154(&mut ring, &test)?;
 

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

@@ -0,0 +1,278 @@
+use crate::Test;
+use ::core::{mem::MaybeUninit, time::Duration};
+use ::rustix::{event::epoll, fd::OwnedFd, io, pipe};
+use ::std::{
+    os::fd::{AsFd, BorrowedFd},
+    thread,
+};
+use io_uring::{cqueue, opcode, squeue, types, IoUring};
+use std::os::unix::io::AsRawFd;
+
+// Tests translated from liburing/test/epwait.c.
+
+#[derive(Debug)]
+struct RxTxPipe {
+    rx: OwnedFd,
+    tx: OwnedFd,
+}
+
+pub fn test_ready<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
+    ring: &mut IoUring<S, C>,
+    test: &Test,
+) -> anyhow::Result<()> {
+    require!(
+        test;
+        test.probe.is_supported(opcode::EpollWait::CODE);
+    );
+
+    println!("test ready");
+
+    const NPIPES: usize = 2;
+    let (efd, pipes, mut events) = init::<NPIPES>()?;
+
+    for pipe in &pipes {
+        let tx = pipe.tx.as_fd();
+        io::write(tx, b"foo")?;
+    }
+
+    let sqe = opcode::EpollWait::new(types::Fd(efd.as_raw_fd()), events.as_mut_ptr(), NPIPES as _)
+        .build()
+        .user_data(0x666)
+        .into();
+    unsafe { ring.submission().push(&sqe) }?;
+
+    ring.submit_and_wait(1)?;
+    for cqe in ring.completion().map(Into::<cqueue::Entry>::into).take(1) {
+        assert_eq!(cqe.user_data_u64(), 0x666);
+        cqe.result()?;
+    }
+
+    let mut tmp = [0u8; 16];
+
+    for event in &events {
+        let fd = unsafe { BorrowedFd::borrow_raw(event.data.u64() as _) };
+        io::read(fd, &mut tmp)?;
+    }
+
+    Ok(())
+}
+
+pub fn test_not_ready<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
+    ring: &mut IoUring<S, C>,
+    test: &Test,
+) -> anyhow::Result<()> {
+    require!(
+        test;
+        test.probe.is_supported(opcode::EpollWait::CODE);
+    );
+
+    println!("test not ready");
+
+    const NPIPES: usize = 2;
+    let (efd, pipes, mut events) = init::<NPIPES>()?;
+
+    let sqe = opcode::EpollWait::new(types::Fd(efd.as_raw_fd()), events.as_mut_ptr(), NPIPES as _)
+        .build()
+        .user_data(0x666)
+        .into();
+    unsafe { ring.submission().push(&sqe) }?;
+
+    for pipe in &pipes {
+        thread::sleep(Duration::from_micros(10000));
+        let tx = pipe.tx.as_fd();
+        io::write(tx, b"foo")?;
+    }
+
+    let mut nr = 0;
+    ring.submit_and_wait(1)?;
+    for cqe in ring.completion().map(Into::<cqueue::Entry>::into).take(1) {
+        assert_eq!(cqe.user_data_u64(), 0x666);
+        nr = cqe.result()?;
+        assert!(nr.cast_signed() >= 0);
+    }
+
+    let mut tmp = [0u8; 16];
+
+    for event in events.iter().take(nr as _) {
+        let fd = unsafe { BorrowedFd::borrow_raw(event.data.u64() as _) };
+        io::read(fd, &mut tmp)?;
+    }
+
+    Ok(())
+}
+
+pub fn test_delete<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
+    ring: &mut IoUring<S, C>,
+    test: &Test,
+) -> anyhow::Result<()> {
+    require!(
+        test;
+        test.probe.is_supported(opcode::EpollWait::CODE);
+    );
+
+    println!("test delete");
+
+    const NPIPES: usize = 2;
+    let (efd, pipes, mut events) = init::<NPIPES>()?;
+
+    let sqe = opcode::EpollWait::new(types::Fd(efd.as_raw_fd()), events.as_mut_ptr(), NPIPES as _)
+        .build()
+        .user_data(0x666)
+        .into();
+    unsafe { ring.submission().push(&sqe) }?;
+
+    epoll::delete(efd.as_fd(), pipes[0].rx.as_fd())?;
+
+    let mut tmp = [0u8; 16];
+
+    for pipe in &pipes {
+        io::write(pipe.tx.as_fd(), &tmp)?;
+    }
+
+    ring.submit_and_wait(1)?;
+    for cqe in ring.completion().map(Into::<cqueue::Entry>::into).take(1) {
+        assert_eq!(cqe.user_data_u64(), 0x666);
+        cqe.result()?;
+    }
+
+    for pipe in &pipes {
+        io::read(pipe.rx.as_fd(), &mut tmp)?;
+    }
+
+    let data = epoll::EventData::new_u64(pipes[0].rx.as_raw_fd().cast_unsigned().into());
+    let flags = epoll::EventFlags::IN;
+    epoll::add(efd, pipes[0].rx.as_fd(), data, flags)?;
+
+    Ok(())
+}
+
+pub fn test_remove<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
+    ring: &mut IoUring<S, C>,
+    test: &Test,
+) -> anyhow::Result<()> {
+    require!(
+        test;
+        test.probe.is_supported(opcode::EpollWait::CODE);
+    );
+
+    println!("test remove");
+
+    const NPIPES: usize = 2;
+    let (efd, pipes, mut events) = init::<NPIPES>()?;
+
+    let sqe = opcode::EpollWait::new(types::Fd(efd.as_raw_fd()), events.as_mut_ptr(), NPIPES as _)
+        .build()
+        .user_data(0x666)
+        .into();
+    unsafe { ring.submission().push(&sqe) }?;
+
+    drop(efd);
+
+    thread::sleep(Duration::from_micros(10000));
+    for pipe in &pipes {
+        io::write(pipe.tx.as_fd(), b"foo")?;
+    }
+
+    ring.submit_and_wait(1)?;
+    for cqe in ring.completion().map(Into::<cqueue::Entry>::into).take(1) {
+        assert_eq!(cqe.user_data_u64(), 0x666);
+        let err = cqe.result().unwrap_err();
+        assert!([io::Errno::AGAIN, io::Errno::BADF].contains(&err));
+    }
+
+    Ok(())
+}
+
+pub fn test_race<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
+    ring: &mut IoUring<S, C>,
+    test: &Test,
+) -> anyhow::Result<()> {
+    require!(
+        test;
+        test.probe.is_supported(opcode::EpollWait::CODE);
+    );
+
+    println!("test race");
+
+    const LOOPS: usize = 500;
+    const NPIPES: usize = 8;
+
+    fn prune(events: &[epoll::Event], nr: usize) -> anyhow::Result<()> {
+        let mut tmp = [0u8; 32];
+
+        for event in events.iter().take(nr) {
+            let fd = unsafe { BorrowedFd::borrow_raw(event.data.u64() as _) };
+            io::read(fd, &mut tmp)?;
+        }
+
+        Ok(())
+    }
+
+    thread::scope(|scope| -> anyhow::Result<()> {
+        let (efd, pipes, mut events) = init::<NPIPES>()?;
+
+        let handle = scope.spawn(move || -> anyhow::Result<()> {
+            for _ in 0..LOOPS {
+                thread::sleep(Duration::from_micros(150));
+                for pipe in &pipes {
+                    io::write(pipe.tx.as_fd(), b"foo")?;
+                }
+            }
+            Ok(())
+        });
+
+        for _ in 0..LOOPS {
+            let sqe = opcode::EpollWait::new(
+                types::Fd(efd.as_raw_fd()),
+                events.as_mut_ptr(),
+                NPIPES as _,
+            )
+            .build()
+            .user_data(0x666)
+            .into();
+            unsafe { ring.submission().push(&sqe) }?;
+            ring.submit_and_wait(1)?;
+            let cqe = ring
+                .completion()
+                .next()
+                .map(Into::<cqueue::Entry>::into)
+                .unwrap();
+            assert_eq!(cqe.user_data_u64(), 0x666);
+            let nr = cqe.result()?;
+            prune(&events, nr as _)?;
+            thread::sleep(Duration::from_micros(100));
+        }
+
+        handle.join().unwrap()?;
+
+        Ok(())
+    })?;
+
+    Ok(())
+}
+
+fn init<const NPIPES: usize>(
+) -> anyhow::Result<(OwnedFd, [RxTxPipe; NPIPES], [epoll::Event; NPIPES])> {
+    let pipes: [RxTxPipe; NPIPES] = {
+        let mut pipes: [MaybeUninit<RxTxPipe>; NPIPES] = [const { MaybeUninit::uninit() }; NPIPES];
+        for pipe in &mut pipes {
+            let (rx, tx) = pipe::pipe()?;
+            pipe.write(RxTxPipe { rx, tx });
+        }
+        unsafe { ::core::mem::transmute_copy(&pipes) }
+    };
+
+    let efd = epoll::create(epoll::CreateFlags::empty())?;
+
+    for pipe in &pipes {
+        let efd = efd.as_fd();
+        let rx = pipe.rx.as_fd();
+        let data = epoll::EventData::new_u64(rx.as_raw_fd().cast_unsigned().into());
+        let flags = epoll::EventFlags::IN;
+        epoll::add(efd, rx, data, flags)?;
+    }
+
+    let events: [epoll::Event; NPIPES] = unsafe { ::core::mem::zeroed() };
+
+    Ok((efd, pipes, events))
+}