kossLAN/wry
1
0
Fork 0
forked from wry/wry
Code Pull requests Activity

io_uring: use strongly typed ids

Browse source
This commit is contained in:
Julian Orth 2024-09-07 17:05:43 +02:00
parent e1d1fe7fda
commit 6cbf6119de
10 changed files with 55 additions and 46 deletions
Download patch file Download diff file Expand all files Collapse all files

35
src/io_uring.rs
View file

@ -22,7 +22,6 @@ use {
copyhashmap::CopyHashMap, copyhashmap::CopyHashMap,
errorfmt::ErrorFmt, errorfmt::ErrorFmt,
mmap::{mmap, Mmapped}, mmap::{mmap, Mmapped},
numcell::NumCell,
oserror::OsError, oserror::OsError,
ptr_ext::{MutPtrExt, PtrExt}, ptr_ext::{MutPtrExt, PtrExt},
stack::Stack, stack::Stack,
@ -254,10 +253,10 @@ struct IoUringData {
cqes_consumed: AsyncEvent, cqes_consumed: AsyncEvent,
next: NumCell<u64>, next: IoUringTaskIds,
to_encode: SyncQueue<u64>, to_encode: SyncQueue<IoUringTaskId>,
pending_in_kernel: CopyHashMap<u64, ()>, pending_in_kernel: CopyHashMap<IoUringTaskId, ()>,
tasks: CopyHashMap<u64, Box<dyn Task>>, tasks: CopyHashMap<IoUringTaskId, Box<dyn Task>>,
pending_results: PendingResults, pending_results: PendingResults,
@ -276,7 +275,7 @@ struct IoUringData {
} }
unsafe trait Task { unsafe trait Task {
fn id(&self) -> u64; fn id(&self) -> IoUringTaskId;
fn complete(self: Box<Self>, ring: &IoUringData, res: i32); fn complete(self: Box<Self>, ring: &IoUringData, res: i32);
fn encode(&self, sqe: &mut io_uring_sqe); fn encode(&self, sqe: &mut io_uring_sqe);
@ -347,8 +346,9 @@ impl IoUringData {
let entry = self.cqmap.deref()[idx].get(); let entry = self.cqmap.deref()[idx].get();
head = head.wrapping_add(1); head = head.wrapping_add(1);
self.cqhead.deref().store(head, Release); self.cqhead.deref().store(head, Release);
if let Some(pending) = self.tasks.remove(&entry.user_data) { let id = IoUringTaskId(entry.user_data);
self.pending_in_kernel.remove(&entry.user_data); if let Some(pending) = self.tasks.remove(&id) {
self.pending_in_kernel.remove(&id);
pending.complete(self, entry.res); pending.complete(self, entry.res);
} }
} }
@ -384,7 +384,7 @@ impl IoUringData {
let sqe = self.sqesmap.deref()[idx].get().deref_mut(); let sqe = self.sqesmap.deref()[idx].get().deref_mut();
self.sqmap.deref()[idx].set(idx as _); self.sqmap.deref()[idx].set(idx as _);
*sqe = Default::default(); *sqe = Default::default();
sqe.user_data = id; sqe.user_data = id.raw();
task.encode(sqe); task.encode(sqe);
if has_timeout { if has_timeout {
sqe.flags |= IOSQE_IO_LINK; sqe.flags |= IOSQE_IO_LINK;
@ -404,11 +404,11 @@ impl IoUringData {
} }
} }
fn id_raw(&self) -> u64 { fn id_raw(&self) -> IoUringTaskId {
self.next.fetch_add(1) self.next.next()
} }
fn cancel_task(&self, id: u64) { fn cancel_task(&self, id: IoUringTaskId) {
if !self.tasks.contains(&id) { if !self.tasks.contains(&id) {
return; return;
} }
@ -466,8 +466,17 @@ impl IoUringData {
} }
} }
linear_ids!(IoUringTaskIds, IoUringTaskId, u64);
#[expect(clippy::derivable_impls)]
impl Default for IoUringTaskId {
fn default() -> Self {
Self(0)
}
}
struct Cancellable<'a> { struct Cancellable<'a> {
id: u64, id: IoUringTaskId,
data: &'a IoUringData, data: &'a IoUringData,
} }

6
src/io_uring/ops/accept.rs
View file

@ -2,7 +2,7 @@ use {
crate::io_uring::{ crate::io_uring::{
pending_result::PendingResult, pending_result::PendingResult,
sys::{io_uring_sqe, IORING_OP_ACCEPT}, sys::{io_uring_sqe, IORING_OP_ACCEPT},
IoUring, IoUringData, IoUringError, Task, TaskResultExt, IoUring, IoUringData, IoUringError, IoUringTaskId, Task, TaskResultExt,
}, },
std::rc::Rc, std::rc::Rc,
uapi::{c, OwnedFd}, uapi::{c, OwnedFd},
@ -39,14 +39,14 @@ struct Data {
#[derive(Default)] #[derive(Default)]
pub struct AcceptTask { pub struct AcceptTask {
id: u64, id: IoUringTaskId,
fd: i32, fd: i32,
flags: u32, flags: u32,
data: Option<Data>, data: Option<Data>,
} }
unsafe impl Task for AcceptTask { unsafe impl Task for AcceptTask {
fn id(&self) -> u64 { fn id(&self) -> IoUringTaskId {
self.id self.id
} }

12
src/io_uring/ops/async_cancel.rs
View file

@ -2,7 +2,7 @@ use {
crate::{ crate::{
io_uring::{ io_uring::{
sys::{io_uring_sqe, IORING_OP_ASYNC_CANCEL}, sys::{io_uring_sqe, IORING_OP_ASYNC_CANCEL},
IoUringData, Task, IoUringData, IoUringTaskId, Task,
}, },
utils::errorfmt::ErrorFmt, utils::errorfmt::ErrorFmt,
}, },
@ -11,12 +11,12 @@ use {
#[derive(Default)] #[derive(Default)]
pub struct AsyncCancelTask { pub struct AsyncCancelTask {
id: u64, id: IoUringTaskId,
target: u64, target: IoUringTaskId,
} }
impl IoUringData { impl IoUringData {
pub fn cancel_task_in_kernel(&self, target: u64) { pub fn cancel_task_in_kernel(&self, target: IoUringTaskId) {
let id = self.id_raw(); let id = self.id_raw();
let mut task = self.cached_cancels.pop().unwrap_or_default(); let mut task = self.cached_cancels.pop().unwrap_or_default();
task.id = id; task.id = id;
@ -26,7 +26,7 @@ impl IoUringData {
} }
unsafe impl Task for AsyncCancelTask { unsafe impl Task for AsyncCancelTask {
fn id(&self) -> u64 { fn id(&self) -> IoUringTaskId {
self.id self.id
} }
@ -41,7 +41,7 @@ unsafe impl Task for AsyncCancelTask {
fn encode(&self, sqe: &mut io_uring_sqe) { fn encode(&self, sqe: &mut io_uring_sqe) {
sqe.opcode = IORING_OP_ASYNC_CANCEL; sqe.opcode = IORING_OP_ASYNC_CANCEL;
sqe.u2.addr = self.target; sqe.u2.addr = self.target.raw();
} }
fn is_cancel(&self) -> bool { fn is_cancel(&self) -> bool {

8
src/io_uring/ops/connect.rs
View file

@ -2,7 +2,7 @@ use {
crate::io_uring::{ crate::io_uring::{
pending_result::PendingResult, pending_result::PendingResult,
sys::{io_uring_sqe, IORING_OP_CONNECT}, sys::{io_uring_sqe, IORING_OP_CONNECT},
IoUring, IoUringData, IoUringError, Task, TaskResultExt, IoUring, IoUringData, IoUringError, IoUringTaskId, Task, TaskResultExt,
}, },
std::{mem, ptr, rc::Rc}, std::{mem, ptr, rc::Rc},
uapi::{c, OwnedFd, SockAddr}, uapi::{c, OwnedFd, SockAddr},
@ -37,7 +37,7 @@ struct Data {
} }
pub struct ConnectTask { pub struct ConnectTask {
id: u64, id: IoUringTaskId,
fd: i32, fd: i32,
sockaddr: c::sockaddr_storage, sockaddr: c::sockaddr_storage,
addrlen: u64, addrlen: u64,
@ -47,7 +47,7 @@ pub struct ConnectTask {
impl Default for ConnectTask { impl Default for ConnectTask {
fn default() -> Self { fn default() -> Self {
Self { Self {
id: 0, id: Default::default(),
fd: 0, fd: 0,
sockaddr: uapi::pod_zeroed(), sockaddr: uapi::pod_zeroed(),
addrlen: 0, addrlen: 0,
@ -57,7 +57,7 @@ impl Default for ConnectTask {
} }
unsafe impl Task for ConnectTask { unsafe impl Task for ConnectTask {
fn id(&self) -> u64 { fn id(&self) -> IoUringTaskId {
self.id self.id
} }

6
src/io_uring/ops/poll.rs
View file

@ -3,7 +3,7 @@ use {
ops::TaskResult, ops::TaskResult,
pending_result::PendingResult, pending_result::PendingResult,
sys::{io_uring_sqe, IORING_OP_POLL_ADD}, sys::{io_uring_sqe, IORING_OP_POLL_ADD},
IoUring, IoUringData, IoUringError, Task, TaskResultExt, IoUring, IoUringData, IoUringError, IoUringTaskId, Task, TaskResultExt,
}, },
std::rc::Rc, std::rc::Rc,
uapi::{c, OwnedFd}, uapi::{c, OwnedFd},
@ -45,14 +45,14 @@ struct Data {
#[derive(Default)] #[derive(Default)]
pub struct PollTask { pub struct PollTask {
id: u64, id: IoUringTaskId,
events: u16, events: u16,
fd: i32, fd: i32,
data: Option<Data>, data: Option<Data>,
} }
unsafe impl Task for PollTask { unsafe impl Task for PollTask {
fn id(&self) -> u64 { fn id(&self) -> IoUringTaskId {
self.id self.id
} }

6
src/io_uring/ops/read_write.rs
View file

@ -3,7 +3,7 @@ use {
io_uring::{ io_uring::{
pending_result::PendingResult, pending_result::PendingResult,
sys::{io_uring_sqe, IORING_OP_READ, IORING_OP_WRITE}, sys::{io_uring_sqe, IORING_OP_READ, IORING_OP_WRITE},
IoUring, IoUringData, IoUringError, Task, TaskResultExt, IoUring, IoUringData, IoUringError, IoUringTaskId, Task, TaskResultExt,
}, },
time::Time, time::Time,
utils::buf::Buf, utils::buf::Buf,
@ -66,7 +66,7 @@ struct ReadWriteTaskData {
#[derive(Default)] #[derive(Default)]
pub struct ReadWriteTask { pub struct ReadWriteTask {
id: u64, id: IoUringTaskId,
has_timeout: bool, has_timeout: bool,
fd: c::c_int, fd: c::c_int,
buf: usize, buf: usize,
@ -76,7 +76,7 @@ pub struct ReadWriteTask {
} }
unsafe impl Task for ReadWriteTask { unsafe impl Task for ReadWriteTask {
fn id(&self) -> u64 { fn id(&self) -> IoUringTaskId {
self.id self.id
} }

8
src/io_uring/ops/recvmsg.rs
View file

@ -3,7 +3,7 @@ use {
io_uring::{ io_uring::{
pending_result::PendingResult, pending_result::PendingResult,
sys::{io_uring_sqe, IORING_OP_RECVMSG}, sys::{io_uring_sqe, IORING_OP_RECVMSG},
IoUring, IoUringData, IoUringError, Task, IoUring, IoUringData, IoUringError, IoUringTaskId, Task,
}, },
utils::buf::Buf, utils::buf::Buf,
}, },
@ -85,7 +85,7 @@ struct Data {
} }
pub struct RecvmsgTask { pub struct RecvmsgTask {
id: u64, id: IoUringTaskId,
fd: c::c_int, fd: c::c_int,
bufs: Vec<Buf>, bufs: Vec<Buf>,
iovecs: Vec<c::iovec>, iovecs: Vec<c::iovec>,
@ -97,7 +97,7 @@ pub struct RecvmsgTask {
impl Default for RecvmsgTask { impl Default for RecvmsgTask {
fn default() -> Self { fn default() -> Self {
RecvmsgTask { RecvmsgTask {
id: 0, id: Default::default(),
fd: 0, fd: 0,
bufs: vec![], bufs: vec![],
iovecs: vec![], iovecs: vec![],
@ -109,7 +109,7 @@ impl Default for RecvmsgTask {
} }
unsafe impl Task for RecvmsgTask { unsafe impl Task for RecvmsgTask {
fn id(&self) -> u64 { fn id(&self) -> IoUringTaskId {
self.id self.id
} }

8
src/io_uring/ops/sendmsg.rs
View file

@ -3,7 +3,7 @@ use {
io_uring::{ io_uring::{
pending_result::PendingResult, pending_result::PendingResult,
sys::{io_uring_sqe, IORING_OP_SENDMSG}, sys::{io_uring_sqe, IORING_OP_SENDMSG},
IoUring, IoUringData, IoUringError, Task, IoUring, IoUringData, IoUringError, IoUringTaskId, Task,
}, },
time::Time, time::Time,
utils::{buf::Buf, vec_ext::UninitVecExt}, utils::{buf::Buf, vec_ext::UninitVecExt},
@ -91,7 +91,7 @@ struct SendmsgTaskData {
} }
pub struct SendmsgTask { pub struct SendmsgTask {
id: u64, id: IoUringTaskId,
iovecs: Vec<c::iovec>, iovecs: Vec<c::iovec>,
msghdr: c::msghdr, msghdr: c::msghdr,
bufs: Vec<Buf>, bufs: Vec<Buf>,
@ -106,7 +106,7 @@ impl Default for SendmsgTask {
fn default() -> Self { fn default() -> Self {
unsafe { unsafe {
SendmsgTask { SendmsgTask {
id: 0, id: Default::default(),
iovecs: vec![], iovecs: vec![],
msghdr: MaybeUninit::zeroed().assume_init(), msghdr: MaybeUninit::zeroed().assume_init(),
bufs: vec![], bufs: vec![],
@ -121,7 +121,7 @@ impl Default for SendmsgTask {
} }
unsafe impl Task for SendmsgTask { unsafe impl Task for SendmsgTask {
fn id(&self) -> u64 { fn id(&self) -> IoUringTaskId {
self.id self.id
} }

6
src/io_uring/ops/timeout.rs
View file

@ -2,7 +2,7 @@ use {
crate::io_uring::{ crate::io_uring::{
pending_result::PendingResult, pending_result::PendingResult,
sys::{io_uring_sqe, IORING_OP_TIMEOUT, IORING_TIMEOUT_ABS}, sys::{io_uring_sqe, IORING_OP_TIMEOUT, IORING_TIMEOUT_ABS},
IoUring, IoUringData, IoUringError, Task, IoUring, IoUringData, IoUringError, IoUringTaskId, Task,
}, },
uapi::c, uapi::c,
}; };
@ -16,7 +16,7 @@ pub(super) struct timespec64 {
#[derive(Default)] #[derive(Default)]
pub struct TimeoutTask { pub struct TimeoutTask {
id: u64, id: IoUringTaskId,
timespec: timespec64, timespec: timespec64,
pr: Option<PendingResult>, pr: Option<PendingResult>,
} }
@ -42,7 +42,7 @@ impl IoUring {
} }
unsafe impl Task for TimeoutTask { unsafe impl Task for TimeoutTask {
fn id(&self) -> u64 { fn id(&self) -> IoUringTaskId {
self.id self.id
} }

6
src/io_uring/ops/timeout_link.rs
View file

@ -2,14 +2,14 @@ use crate::{
io_uring::{ io_uring::{
ops::timeout::timespec64, ops::timeout::timespec64,
sys::{io_uring_sqe, IORING_OP_LINK_TIMEOUT, IORING_TIMEOUT_ABS}, sys::{io_uring_sqe, IORING_OP_LINK_TIMEOUT, IORING_TIMEOUT_ABS},
IoUring, IoUringData, Task, IoUring, IoUringData, IoUringTaskId, Task,
}, },
time::Time, time::Time,
}; };
#[derive(Default)] #[derive(Default)]
pub struct TimeoutLinkTask { pub struct TimeoutLinkTask {
id: u64, id: IoUringTaskId,
timespec: timespec64, timespec: timespec64,
} }
@ -27,7 +27,7 @@ impl IoUring {
} }
unsafe impl Task for TimeoutLinkTask { unsafe impl Task for TimeoutLinkTask {
fn id(&self) -> u64 { fn id(&self) -> IoUringTaskId {
self.id self.id
} }