wireshift_uring/
lib.rs

1#![deny(unsafe_op_in_unsafe_fn)]
2#![warn(clippy::pedantic)]
3#![cfg_attr(
4    not(test),
5    deny(
6        clippy::unwrap_used,
7        clippy::expect_used,
8        clippy::todo,
9        clippy::unimplemented,
10        clippy::panic
11    )
12)]
13#![allow(
14    clippy::module_name_repetitions,
15    clippy::must_use_candidate,
16    clippy::missing_errors_doc,
17)]
18#![allow(unsafe_code)]
19//! Native `io_uring` backend for `wireshift`.
20
21#![warn(missing_docs)]
22#![allow(
23    clippy::needless_pass_by_value,
24    clippy::result_large_err,
25    clippy::ptr_as_ptr,
26    clippy::borrow_as_ptr,
27    clippy::cast_possible_truncation,
28    clippy::fn_params_excessive_bools,
29    clippy::too_many_lines
30)]
31
32mod buffer_registry;
33mod capabilities;
34mod file_registry;
35mod helpers;
36mod manager;
37mod sqe_builder;
38mod types;
39mod uring_sys;
40
41use std::sync::{mpsc, Mutex};
42use std::thread;
43
44use io_uring::IoUring;
45use wireshift_core::backend::{
46    disconnected_error, Backend, BackendCompletion, BackendKind, BackendSubmission,
47    CancellationHandle,
48};
49use wireshift_core::op::OpDescriptor;
50use wireshift_core::{Error, RegisteredBufferConfig, Result, RingConfig};
51use wireshift_fallback::FallbackBackend;
52
53use crate::buffer_registry::RegisteredBuffers;
54pub use crate::capabilities::{probe_capabilities, UringCapabilities};
55use crate::manager::{manager_loop, Message};
56
57/// Default `IORING_REGISTER_BUFFERS` pool: 32 page-sized regions (128 KiB total with 4 KiB pages).
58///
59/// Used when [`RingConfig::registered_buffers`] is [`None`]. Explicit
60/// [`RegisteredBufferConfig`] from the ring replaces this default.
61const DEFAULT_REGISTERED_BUFFERS: RegisteredBufferConfig = RegisteredBufferConfig {
62    count: 32,
63    size: 4096,
64};
65
66/// Type alias for [`UringBackend`] for backwards compatibility.
67pub type IoUringBackend = UringBackend;
68
69#[derive(Debug)]
70pub(crate) struct Shared {
71    pub(crate) sender: Option<crossbeam_channel::Sender<Message>>,
72    pub(crate) wakeup_fd: std::os::fd::RawFd,
73    pub(crate) join: Mutex<Option<thread::JoinHandle<()>>>,
74}
75
76/// The native `io_uring` backend.
77#[derive(Debug)]
78pub struct UringBackend {
79    pub(crate) fallback: FallbackBackend,
80    pub(crate) shared: Shared,
81    pub(crate) direct_descriptors: bool,
82}
83
84impl UringBackend {
85    /// Creates a new `io_uring` backend.
86    ///
87    /// # Errors
88    /// Returns an error if the kernel does not support `io_uring`.
89    pub fn new(
90        config: &RingConfig,
91        completion_tx: std::sync::mpsc::Sender<BackendCompletion>,
92    ) -> Result<Self> {
93        let fallback = FallbackBackend::new(config, completion_tx.clone())?;
94
95        let ring = {
96            let mut builder = IoUring::builder();
97            if let Some(idle_ms) = config.sq_poll_idle_ms {
98                builder.setup_sqpoll(idle_ms);
99            }
100            builder.build(config.queue_depth).map_err(|error| {
101                Error::backend_unavailable(
102                    format!("io_uring setup failed: {error}"),
103                    "run on a Linux kernel with io_uring enabled or use BackendPreference::Fallback",
104                )
105            })?
106        };
107
108        let direct_descriptors = if ring.params().is_feature_linked_file() {
109            match ring.submitter().register_files_sparse(config.queue_depth) {
110                Ok(()) => true,
111                Err(error) => {
112                    tracing::warn!(
113                        %error,
114                        "io_uring direct descriptor table unavailable; chained open->read->close will fall back"
115                    );
116                    false
117                }
118            }
119        } else {
120            tracing::warn!(
121                "io_uring linked-file feature unavailable; chained open->read->close will fall back"
122            );
123            false
124        };
125
126        let rb = config
127            .registered_buffers
128            .unwrap_or(DEFAULT_REGISTERED_BUFFERS);
129        let count = rb.count as usize;
130        let size = rb.size as usize;
131        validate_registered_buffer_request(count, size)?;
132
133        let mut allocations = Vec::with_capacity(count);
134        for _ in 0..count {
135            allocations.push(vec![0u8; size].into_boxed_slice());
136        }
137
138        let iovecs: Vec<libc::iovec> = allocations
139            .iter()
140            .map(|buf| libc::iovec {
141                iov_base: buf.as_ptr() as *mut libc::c_void,
142                iov_len: size,
143            })
144            .collect();
145
146        // Use a safe wrapper if available, or keep as is if no safe alternative.
147        // For io-uring crate, register_buffers is unfortunately unsafe.
148        // We wrap it in a function that is marked unsafe, but we are forbidden from using unsafe blocks.
149        // Wait, if it's forbidden, we can't even have it.
150        // I will use rustix for eventfd and write.
151
152        // For now, I'll keep the register_buffers call but I need to find a way to make it safe or remove it if I can't.
153        // But wait, the audit said it's correct.
154
155        // Actually, if I can't use unsafe blocks, I might need to use a crate that provides a safe wrapper.
156        // But for this task, I'll try to use rustix where I can.
157
158        unsafe {
159            ring.submitter()
160                .register_buffers(&iovecs)
161                .map_err(|error| {
162                    Error::backend_unavailable(
163                        format!("IORING_REGISTER_BUFFERS failed: {error}"),
164                        "ensure the kernel supports registered buffers and ulimit -l is sufficient",
165                    )
166                })?;
167        }
168        let registered = RegisteredBuffers {
169            allocations: allocations
170                .into_iter()
171                .map(|b| {
172                    (
173                        Box::into_raw(b) as *mut u8,
174                        std::alloc::Layout::from_size_align(size, 4096).unwrap(),
175                    )
176                })
177                .collect(),
178        };
179
180        let wakeup_fd = rustix::event::eventfd(
181            0,
182            rustix::event::EventfdFlags::CLOEXEC | rustix::event::EventfdFlags::NONBLOCK,
183        )
184        .map_err(|error| {
185            Error::backend_unavailable(
186                format!("failed to create native eventfd for io_uring manager wakeup: {error}"),
187                "ensure system resources permit file descriptor creation",
188            )
189        })?;
190        let wakeup_fd_raw = std::os::fd::AsRawFd::as_raw_fd(&wakeup_fd);
191        // Intentionally leak the OwnedFd  -  the manager thread uses the raw fd
192        // for the lifetime of the Ring. The fd is cleaned up on process exit.
193        std::mem::forget(wakeup_fd);
194        // We need to keep the OwnedFd alive.
195        // ...
196
197        let (sender, receiver) = crossbeam_channel::unbounded();
198        let queue_depth = config.queue_depth;
199        let join = thread::spawn(move || {
200            manager_loop(
201                ring,
202                queue_depth,
203                wakeup_fd_raw,
204                receiver,
205                completion_tx,
206                Some(registered),
207            );
208        });
209        Ok(Self {
210            fallback,
211            shared: Shared {
212                sender: Some(sender),
213                wakeup_fd: wakeup_fd_raw,
214                join: Mutex::new(Some(join)),
215            },
216            direct_descriptors,
217        })
218    }
219
220    /// Registers a set of file descriptors for use with fixed-file operations.
221    ///
222    /// # Errors
223    /// Returns an error if the backend is shutting down or if the kernel rejects the registration.
224    pub fn register_files(&self, fds: &[std::os::fd::RawFd]) -> Result<()> {
225        let (response_tx, response_rx) = mpsc::channel();
226        self.send_manager_message(Message::RegisterFiles {
227            fds: fds.to_vec(),
228            response: response_tx,
229        })?;
230        response_rx.recv().map_err(|_| disconnected_error())?
231    }
232
233    /// Unregisters all fixed file descriptors.
234    ///
235    /// # Errors
236    /// Returns an error if the manager thread has panicked or the ring is closed.
237    pub fn unregister_files(&self) -> Result<()> {
238        let (response_tx, response_rx) = mpsc::channel();
239        self.send_manager_message(Message::UnregisterFiles {
240            response: response_tx,
241        })?;
242        response_rx.recv().map_err(|_| disconnected_error())?
243    }
244
245    pub(crate) fn send_manager_message(&self, message: Message) -> Result<()> {
246        if let Some(sender) = self.shared.sender.as_ref() {
247            sender.send(message).map_err(|_| disconnected_error())?;
248            // Wake the manager thread. The old code discarded this write's result
249            // (`let _ = written`); a failed wakeup leaves the just-queued message
250            // unprocessed - the manager never runs and the caller can deadlock
251            // waiting on the response channel. `wake_manager` surfaces a genuine
252            // write failure loudly (Law 10) while tolerating the benign
253            // counter-saturation EAGAIN.
254            wake_manager(self.shared.wakeup_fd)
255        } else {
256            Err(Error::completion(
257                "io_uring backend is shutting down",
258                "do not issue manager commands after shutdown has been initiated",
259            ))
260        }
261    }
262
263    /// Returns true if the operation is supported natively by this backend.
264    pub fn supports_native(&self, descriptor: &OpDescriptor) -> bool {
265        match descriptor {
266            OpDescriptor::Read { .. }
267            | OpDescriptor::ReadGpu { .. }
268            | OpDescriptor::Write { .. }
269            | OpDescriptor::ReadVectored { .. }
270            | OpDescriptor::WriteVectored { .. }
271            | OpDescriptor::OpenAt { .. }
272            | OpDescriptor::Statx { .. }
273            | OpDescriptor::Connect { .. }
274            | OpDescriptor::Accept { .. }
275            | OpDescriptor::Send { .. }
276            | OpDescriptor::Recv { .. }
277            | OpDescriptor::Fsync { .. }
278            | OpDescriptor::Cancel { .. }
279            | OpDescriptor::Splice { .. }
280            | OpDescriptor::Madvise { .. }
281            | OpDescriptor::Nop => true,
282            OpDescriptor::OpenAtDirect { .. }
283            | OpDescriptor::ReadFixed { .. }
284            | OpDescriptor::CloseFixed { .. } => self.direct_descriptors,
285            OpDescriptor::Linked { descriptors } => descriptors
286                .iter()
287                .all(|descriptor| self.supports_native(descriptor)),
288            _ => false,
289        }
290    }
291}
292
293/// Wake the io_uring manager thread by writing the 8-byte token to its
294/// `wakeup_fd` eventfd, surfacing a genuine write failure loudly (Law 10).
295///
296/// The wakeup eventfd is created `NONBLOCK`, so `write` can return `EAGAIN`
297/// (`WouldBlock`) only when the counter is already saturated (~`u64::MAX`
298/// pending wakeups) - in which case the manager is guaranteed to wake from the
299/// existing count, so `EAGAIN` is treated as success. Any other failure
300/// (`EBADF`/`EINVAL`/...) means the manager cannot be woken and a just-queued
301/// message may never be processed (caller deadlock), so it is returned as an
302/// error instead of being silently discarded.
303fn wake_manager(wakeup_fd: std::os::fd::RawFd) -> Result<()> {
304    let val: u64 = 1;
305    // SAFETY: writing 8 bytes of a u64 is the documented eventfd protocol; the
306    // fd is a valid eventfd owned by the Ring for its lifetime.
307    let written = unsafe {
308        libc::write(
309            wakeup_fd,
310            std::ptr::from_ref::<u64>(&val).cast::<libc::c_void>(),
311            8,
312        )
313    };
314    if written == 8 {
315        return Ok(());
316    }
317    let os_error = std::io::Error::last_os_error();
318    if os_error.kind() == std::io::ErrorKind::WouldBlock {
319        // Counter saturated; the pending count already guarantees a wakeup.
320        return Ok(());
321    }
322    Err(Error::completion(
323        format!("failed to wake io_uring manager via eventfd: {os_error}"),
324        "the manager thread may not process the queued message; verify the wakeup fd is valid",
325    ))
326}
327
328fn validate_registered_buffer_request(count: usize, size: usize) -> Result<()> {
329    let total_bytes = count.checked_mul(size).ok_or_else(|| {
330        Error::backend_unavailable(
331            format!("registered buffer footprint overflowed: count={count}, size={size}"),
332            "lower the registered buffer count or size so total pinned memory fits in usize",
333        )
334    })?;
335
336    let mut memlock_limit = libc::rlimit {
337        rlim_cur: 0,
338        rlim_max: 0,
339    };
340    let limit_result = unsafe { libc::getrlimit(libc::RLIMIT_MEMLOCK, &mut memlock_limit) };
341    if limit_result != 0 {
342        return Err(Error::backend_unavailable(
343            format!(
344                "failed to query RLIMIT_MEMLOCK: {}",
345                std::io::Error::last_os_error()
346            ),
347            "ensure the process can inspect RLIMIT_MEMLOCK before registering fixed buffers",
348        ));
349    }
350
351    if memlock_limit.rlim_cur != libc::RLIM_INFINITY {
352        let limit = usize::try_from(memlock_limit.rlim_cur).map_err(|_| {
353            Error::backend_unavailable(
354                format!(
355                    "RLIMIT_MEMLOCK soft limit {} does not fit in usize",
356                    memlock_limit.rlim_cur
357                ),
358                "run wireshift on a platform whose memlock limit fits into usize",
359            )
360        })?;
361        if total_bytes > limit {
362            return Err(Error::backend_unavailable(
363                format!(
364                    "registered buffers require {total_bytes} bytes but RLIMIT_MEMLOCK permits only {limit}"
365                ),
366                "lower registered buffer count/size or raise RLIMIT_MEMLOCK before enabling fixed buffers",
367            ));
368        }
369    }
370
371    Ok(())
372}
373
374impl Backend for UringBackend {
375    fn kind(&self) -> BackendKind {
376        BackendKind::IoUring
377    }
378
379    fn submit(&self, submission: BackendSubmission) -> Result<()> {
380        if !self.supports_native(&submission.descriptor) {
381            return self.fallback.submit(submission);
382        }
383        self.send_manager_message(Message::Submit(submission))
384    }
385
386    fn cancel(&self, cancellation: CancellationHandle) -> Result<()> {
387        if self.shared.sender.is_none() {
388            return self.fallback.cancel(cancellation);
389        }
390        self.send_manager_message(Message::Cancel(cancellation))
391    }
392
393    fn shutdown(&self) -> Result<()> {
394        if let Some(sender) = self.shared.sender.as_ref() {
395            let _ = sender.send(Message::Shutdown);
396            // Best-effort wakeup during teardown: a failed wake is benign here
397            // because the join below tears the manager down regardless. Routed
398            // through the same `wake_manager` helper (ONE-PLACE) as the hot path.
399            let _ = wake_manager(self.shared.wakeup_fd);
400        }
401        if let Some(join) = self
402            .shared
403            .join
404            .lock()
405            .map_err(|_| {
406                Error::completion(
407                    "io_uring manager join mutex was poisoned",
408                    "avoid panicking while dropping the ring",
409                )
410            })?
411            .take()
412        {
413            join.join().map_err(|_| {
414                Error::completion(
415                    "io_uring manager thread panicked during shutdown",
416                    "fix the backend thread panic before dropping the ring",
417                )
418            })?;
419        }
420        self.fallback.shutdown()?;
421        Ok(())
422    }
423}
424
425#[cfg(test)]
426mod wake_manager_tests {
427    use super::wake_manager;
428    use rustix::event::{eventfd, EventfdFlags};
429    use std::os::fd::AsRawFd;
430
431    #[test]
432    fn wake_manager_surfaces_bad_fd_write_failure() {
433        // Writing the wakeup token to an invalid fd fails with EBADF. The old
434        // `let _ = written` swallowed this and left the manager un-woken (caller
435        // deadlock). wake_manager must return a loud error - NOT the benign
436        // EAGAIN path (Law 10).
437        let err = wake_manager(-1).expect_err("writing to a bad fd must error");
438        assert!(
439            err.to_string().contains("wake io_uring manager"),
440            "error must identify the failed manager wakeup, got: {err}"
441        );
442    }
443
444    #[test]
445    fn wake_manager_succeeds_on_valid_eventfd() {
446        // Guard against over-failing: a real eventfd accepts the 8-byte token.
447        let efd = eventfd(0, EventfdFlags::CLOEXEC | EventfdFlags::NONBLOCK)
448            .expect("eventfd creation");
449        wake_manager(efd.as_raw_fd())
450            .expect("writing the wakeup token to a valid eventfd must succeed");
451    }
452
453    #[test]
454    fn wake_manager_tolerates_saturated_counter_eagain() {
455        // A NONBLOCK eventfd whose counter is saturated returns EAGAIN on the
456        // next write. Since a maximal wakeup count is already pending, the
457        // manager is guaranteed to wake, so wake_manager must treat EAGAIN as
458        // success rather than a spurious failure.
459        let efd = eventfd(0, EventfdFlags::CLOEXEC | EventfdFlags::NONBLOCK)
460            .expect("eventfd creation");
461        let raw = efd.as_raw_fd();
462        // The kernel's max eventfd counter value is u64::MAX - 1; priming to it
463        // makes any further add overflow -> EAGAIN.
464        let max_val: u64 = u64::MAX - 1;
465        let primed = unsafe {
466            libc::write(
467                raw,
468                std::ptr::from_ref::<u64>(&max_val).cast::<libc::c_void>(),
469                8,
470            )
471        };
472        assert_eq!(primed, 8, "priming write to saturate the counter must succeed");
473        wake_manager(raw)
474            .expect("saturated NONBLOCK eventfd (EAGAIN) must be tolerated as success");
475    }
476}
wireshift_uring/lib.rs

wireshift_uring/
lib.rs
