dope_core/backend/uring/
sqe.rs

1use std::io::{self, Error, ErrorKind};
2use std::mem::MaybeUninit;
3use std::os::fd::RawFd;
4use std::slice;
5
6use io_uring::types;
7use o3::marker::ThreadBound;
8
9use crate::driver::token::SHUTDOWN;
10use crate::driver::token::{Epoch, ROUTE_FRAMEWORK, SlotIndex, Token, kind};
11use crate::io::fd::{Fd, FdSlot};
12use io_uring::opcode::Accept;
13use io_uring::opcode::AsyncCancel;
14use io_uring::opcode::Bind;
15use io_uring::opcode::Close;
16use io_uring::opcode::Connect;
17use io_uring::opcode::Listen;
18use io_uring::opcode::OpenAt;
19use io_uring::opcode::PollAdd;
20use io_uring::opcode::Read;
21use io_uring::opcode::Recv;
22use io_uring::opcode::RecvMsgMulti;
23use io_uring::opcode::RecvMulti;
24use io_uring::opcode::Send;
25use io_uring::opcode::SendMsg;
26use io_uring::opcode::SetSockOpt;
27use io_uring::opcode::Shutdown;
28use io_uring::opcode::Socket;
29use io_uring::opcode::Statx;
30use io_uring::opcode::Timeout;
31use io_uring::opcode::Write;
32use io_uring::squeue::Entry;
33use io_uring::squeue::Flags;
34use io_uring::types::DestinationSlot;
35
36#[derive(Clone, Copy)]
37pub struct Create {
38    pub slot: FdSlot,
39    pub user_data: u64,
40}
41
42pub struct Sqe {
43    entry: Entry,
44    create: Option<Create>,
45    _thread: ThreadBound,
46}
47
48impl Sqe {
49    pub fn entry(&self) -> &Entry {
50        &self.entry
51    }
52
53    pub fn create_meta(&self) -> Option<Create> {
54        self.create
55    }
56
57    fn new(entry: Entry) -> Self {
58        Self {
59            entry,
60            create: None,
61            _thread: ThreadBound::NEW,
62        }
63    }
64
65    fn create(entry: Entry, slot: FdSlot, user_data: u64) -> Self {
66        let token = Token::new(ROUTE_FRAMEWORK, SlotIndex::new(slot.raw()), Epoch::ZERO)
67            .with_kind(kind::CREATE);
68        Self {
69            entry: entry.user_data(token.raw()),
70            create: Some(Create { slot, user_data }),
71            _thread: ThreadBound::NEW,
72        }
73    }
74
75    fn framework(slot: FdSlot, op_kind: u8) -> u64 {
76        Token::new(ROUTE_FRAMEWORK, SlotIndex::new(slot.raw()), Epoch::ZERO)
77            .with_kind(op_kind)
78            .raw()
79    }
80
81    pub fn from_entry(entry: Entry) -> Self {
82        Self::new(entry)
83    }
84
85    pub fn send(fd: &Fd, buf: &[u8], op: Token) -> Self {
86        Self::send_at(fd.slot(), buf, op)
87    }
88
89    pub fn send_at(slot: FdSlot, buf: &[u8], op: Token) -> Self {
90        Self::new(
91            Send::new(types::Fixed(slot.raw()), buf.as_ptr(), buf.len() as u32)
92                .flags(libc::MSG_NOSIGNAL)
93                .build()
94                .user_data(op.with_kind(kind::SEND).raw()),
95        )
96    }
97
98    /// # Safety
99    /// `fd` must stay open and `buf` stable and unchanged until completion.
100    pub unsafe fn write_fd(fd: RawFd, buf: &[u8], offset: u64, op: Token) -> Self {
101        Self::new(
102            Write::new(types::Fd(fd), buf.as_ptr(), buf.len() as u32)
103                .offset(offset)
104                .build()
105                .user_data(op.with_kind(kind::WRITE).raw()),
106        )
107    }
108
109    pub fn openat(dir: RawFd, path: *const libc::c_char, flags: i32, mode: u32, op: Token) -> Self {
110        Self::new(
111            OpenAt::new(types::Fd(dir), path)
112                .flags(flags)
113                .mode(mode as libc::mode_t)
114                .build()
115                .user_data(op.with_kind(kind::OPEN).raw()),
116        )
117    }
118
119    /// # Safety
120    /// `fd` must stay open and `buf` stable and unaliased until completion.
121    pub unsafe fn read(fd: RawFd, buf: &mut [u8], offset: u64, op: Token) -> Self {
122        let buf = unsafe {
123            slice::from_raw_parts_mut(buf.as_mut_ptr().cast::<MaybeUninit<u8>>(), buf.len())
124        };
125        unsafe { Self::read_uninit(fd, buf, offset, op.with_kind(kind::READ)) }
126    }
127
128    /// # Safety
129    /// `fd` must stay open and `buf` stable and unaliased until completion.
130    pub unsafe fn read_uninit(
131        fd: RawFd,
132        buf: &mut [MaybeUninit<u8>],
133        offset: u64,
134        op: Token,
135    ) -> Self {
136        Self::new(
137            Read::new(types::Fd(fd), buf.as_mut_ptr().cast(), buf.len() as u32)
138                .offset(offset)
139                .build()
140                .user_data(op.raw()),
141        )
142    }
143
144    pub fn stat_path(path: *const libc::c_char, stat: *mut libc::statx, op: Token) -> Self {
145        Self::new(
146            Statx::new(types::Fd(libc::AT_FDCWD), path, stat.cast::<types::statx>())
147                .mask(libc::STATX_TYPE | libc::STATX_SIZE | libc::STATX_MTIME)
148                .build()
149                .user_data(op.with_kind(kind::STAT).raw()),
150        )
151    }
152
153    pub fn stat_fd(fd: RawFd, stat: *mut libc::statx, op: Token) -> Self {
154        Self::new(
155            Statx::new(types::Fd(fd), c"".as_ptr(), stat.cast::<types::statx>())
156                .flags(libc::AT_EMPTY_PATH)
157                .mask(libc::STATX_TYPE | libc::STATX_SIZE | libc::STATX_MTIME)
158                .build()
159                .user_data(op.with_kind(kind::STAT).raw()),
160        )
161    }
162
163    /// # Safety
164    /// `fd` must belong to the receiving driver and stay live until completion.
165    pub unsafe fn recv_multi(fd: &Fd, buf_group: u16, op: Token) -> Self {
166        Self::new(
167            RecvMulti::new(types::Fixed(fd.slot().raw()), buf_group)
168                .build()
169                .user_data(op.with_kind(kind::RECV).raw()),
170        )
171    }
172
173    pub const SUPPORTS_RECV_DISCARD: bool = true;
174
175    /// # Safety
176    /// `fd` must belong to the receiving driver and stay live until completion.
177    pub unsafe fn recv_discard(fd: &Fd, remaining: u64, op: Token) -> Self {
178        const DISCARD_CAP: u64 = 1 << 30;
179        static SCRATCH: u8 = 0;
180        let len = remaining.min(DISCARD_CAP) as u32;
181        Self::new(
182            Recv::new(
183                types::Fixed(fd.slot().raw()),
184                &SCRATCH as *const u8 as *mut u8,
185                len,
186            )
187            .flags(libc::MSG_TRUNC)
188            .build()
189            .user_data(op.with_kind(kind::RECV_DISCARD).raw()),
190        )
191    }
192
193    pub fn accept_oneshot(
194        listener: &Fd,
195        addr_ptr: *mut libc::sockaddr,
196        addrlen_ptr: *mut libc::socklen_t,
197        op: Token,
198    ) -> Self {
199        Self::new(
200            Accept::new(types::Fixed(listener.slot().raw()), addr_ptr, addrlen_ptr)
201                .file_index(Some(DestinationSlot::auto_target()))
202                .flags(0)
203                .build()
204                .user_data(op.with_kind(kind::ACCEPT).raw()),
205        )
206    }
207
208    pub fn recv_msg_multi(fd: &Fd, msghdr: &libc::msghdr, buf_group: u16, op: Token) -> Self {
209        Self::new(
210            RecvMsgMulti::new(types::Fixed(fd.slot().raw()), msghdr, buf_group)
211                .build()
212                .user_data(op.with_kind(kind::RECV).raw()),
213        )
214    }
215
216    pub fn send_msg(fd: &Fd, msg: &libc::msghdr, op: Token) -> Self {
217        Self::new(
218            SendMsg::new(types::Fixed(fd.slot().raw()), msg)
219                .flags(libc::MSG_NOSIGNAL as u32)
220                .build()
221                .user_data(op.with_kind(kind::SEND).raw()),
222        )
223    }
224
225    pub fn close_at(slot: FdSlot) -> Self {
226        Self::new(
227            Close::new(types::Fixed(slot.raw()))
228                .build()
229                .user_data(Self::framework(slot, kind::CLOSE)),
230        )
231    }
232
233    pub fn quickack(fd: &Fd) -> Self {
234        const TCP_QUICKACK: u32 = 12;
235        static QUICKACK_ON: libc::c_int = 1;
236        Self::new(
237            SetSockOpt::new(
238                types::Fixed(fd.slot().raw()),
239                libc::IPPROTO_TCP as u32,
240                TCP_QUICKACK,
241                &QUICKACK_ON as *const libc::c_int as *const libc::c_void,
242                size_of::<libc::c_int>() as u32,
243            )
244            .build()
245            .user_data(0),
246        )
247    }
248
249    pub fn shutdown(fd: &Fd, how: i32) -> Self {
250        Self::new(
251            Shutdown::new(types::Fixed(fd.slot().raw()), how)
252                .build()
253                .user_data(0),
254        )
255    }
256
257    pub fn shutdown_linked_at(slot: FdSlot, how: i32) -> Self {
258        Self::new(
259            Shutdown::new(types::Fixed(slot.raw()), how)
260                .build()
261                .flags(Flags::IO_HARDLINK)
262                .user_data(Self::framework(slot, kind::CLOSE_PREP)),
263        )
264    }
265
266    pub fn poll_shutdown(fd: RawFd) -> Self {
267        Self::new(
268            PollAdd::new(types::Fd(fd), libc::POLLIN as u32)
269                .build()
270                .user_data(SHUTDOWN.raw()),
271        )
272    }
273
274    pub fn cancel(target: Token, op_kind: u8) -> Self {
275        Self::new(
276            AsyncCancel::new(target.with_kind(op_kind).raw())
277                .build()
278                .user_data(0),
279        )
280    }
281
282    /// Arms a kernel-owned recurring timer.
283    ///
284    /// The timer specification is referenced by the multishot operation and
285    /// must therefore remain live until the driver is torn down or the timer
286    /// is cancelled.
287    pub fn interval(timer: &'static types::Timespec, op: Token) -> Self {
288        Self::new(
289            Timeout::new(timer)
290                .count(0)
291                .flags(types::TimeoutFlags::MULTISHOT)
292                .build()
293                .user_data(op.with_kind(kind::TIMER).raw()),
294        )
295    }
296
297    pub fn cancel_create(slot: FdSlot) -> Self {
298        Self::new(
299            AsyncCancel::new(Self::framework(slot, kind::CREATE))
300                .build()
301                .user_data(0),
302        )
303    }
304
305    pub fn socket(
306        domain: i32,
307        socket_type: i32,
308        protocol: i32,
309        fd: &Fd,
310        op: Token,
311    ) -> io::Result<Self> {
312        Self::socket_at(domain, socket_type, protocol, fd.slot(), op)
313    }
314
315    pub fn socket_at(
316        domain: i32,
317        socket_type: i32,
318        protocol: i32,
319        slot: FdSlot,
320        op: Token,
321    ) -> io::Result<Self> {
322        let dest = DestinationSlot::try_from_slot_target(slot.raw())
323            .map_err(|_| Error::new(ErrorKind::InvalidInput, "dope: socket slot out of range"))?;
324        Ok(Self::create(
325            Socket::new(domain, socket_type, protocol)
326                .file_index(Some(dest))
327                .build(),
328            slot,
329            op.with_kind(kind::SOCKET).raw(),
330        ))
331    }
332
333    pub fn bind_at(
334        slot: FdSlot,
335        addr_ptr: *const libc::sockaddr,
336        addr_len: u32,
337        op: Token,
338    ) -> Self {
339        Self::new(
340            Bind::new(types::Fixed(slot.raw()), addr_ptr, addr_len)
341                .build()
342                .user_data(op.with_kind(kind::SOCKET).raw()),
343        )
344    }
345
346    pub fn listen_at(slot: FdSlot, backlog: i32, op: Token) -> Self {
347        Self::new(
348            Listen::new(types::Fixed(slot.raw()), backlog)
349                .build()
350                .user_data(op.with_kind(kind::SOCKET).raw()),
351        )
352    }
353
354    pub fn connect(fd: &Fd, addr_ptr: *const libc::sockaddr, addr_len: u32, op: Token) -> Self {
355        Self::new(
356            Connect::new(types::Fixed(fd.slot().raw()), addr_ptr, addr_len)
357                .build()
358                .user_data(op.with_kind(kind::CONNECT).raw()),
359        )
360    }
361}
dope_core/backend/uring/sqe.rs

dope_core/backend/uring/
sqe.rs
