Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
229 changes: 152 additions & 77 deletions Cargo.lock

Large diffs are not rendered by default.

7 changes: 6 additions & 1 deletion Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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 }
Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Lets try and avoid adding any dependencies. This should really just be rustix + any supporting helpers.

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" }
Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also lets only land this when there is a release we can point to.

3 changes: 2 additions & 1 deletion io-uring-test/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,8 @@ publish = false
[dependencies]
io-uring = { path = "..", package = "rustix-uring" }
libc = { version = "0.2", features = [ "extra_traits" ] }
rustix = { version = "1.0.2", features = ["fs"] }
linux-raw-sys = { version = "0.10.0", default-features = false }
rustix = { version = "1.0.2", features = ["fs", "param"] }
anyhow = "1"
tempfile = "3"
once_cell = "1"
Expand Down
17 changes: 17 additions & 0 deletions io-uring-test/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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)?;
Expand All @@ -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)?;
Expand Down Expand Up @@ -139,11 +150,14 @@ 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)?;
tests::net::test_udp_sendzc_with_dest(&mut ring, &test)?;

tests::net::test_tcp_recvzc::<S>(&test)?;

// queue
tests::poll::test_eventfd_poll(&mut ring, &test)?;
tests::poll::test_eventfd_poll_remove(&mut ring, &test)?;
Expand All @@ -155,6 +169,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)?;

Expand Down
291 changes: 291 additions & 0 deletions io-uring-test/src/tests/epoll.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,291 @@
use crate::Test;
use ::core::{mem::MaybeUninit, time::Duration};
use ::rustix::{event::epoll, fd::OwnedFd, io};
use ::std::{
io::{PipeReader, PipeWriter},
os::{
fd::{AsFd, BorrowedFd},
unix::io::AsRawFd,
},
thread,
};
use io_uring::{cqueue, opcode, squeue, types, IoUring};

// Tests translated from liburing/test/epwait.c.

#[derive(Debug)]
struct RxTxPipe {
rx: PipeReader,
tx: PipeWriter,
}

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 REQ_TYPE_EPOLL_WAIT: u64 = 1;

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(REQ_TYPE_EPOLL_WAIT)
.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(), REQ_TYPE_EPOLL_WAIT);
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 REQ_TYPE_EPOLL_WAIT: u64 = 1;

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(REQ_TYPE_EPOLL_WAIT)
.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(), REQ_TYPE_EPOLL_WAIT);
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 REQ_TYPE_EPOLL_WAIT: u64 = 1;

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(REQ_TYPE_EPOLL_WAIT)
.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(), REQ_TYPE_EPOLL_WAIT);
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 REQ_TYPE_EPOLL_WAIT: u64 = 1;

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(REQ_TYPE_EPOLL_WAIT)
.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(), REQ_TYPE_EPOLL_WAIT);
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 REQ_TYPE_EPOLL_WAIT: u64 = 1;

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(REQ_TYPE_EPOLL_WAIT)
.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(), REQ_TYPE_EPOLL_WAIT);
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) = ::std::io::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))
}
Loading