io: AioSource now employs IO Safety (#7992)

This commit is contained in:
Alan Somers
2026-04-08 22:59:32 +02:00
committed by GitHub
parent fccc28f1d0
commit d7db722bb4
6 changed files with 51 additions and 28 deletions
+2 -2
View File
@@ -125,7 +125,7 @@ libc = { version = "0.2.168", optional = true }
[target.'cfg(unix)'.dev-dependencies]
libc = { version = "0.2.168" }
nix = { version = "0.29.0", default-features = false, features = ["aio", "fs", "socket"] }
nix = { version = "0.31.0", default-features = false, features = ["aio", "fs", "socket"] }
[target.'cfg(windows)'.dependencies.windows-sys]
version = "0.61"
@@ -160,7 +160,7 @@ rand = "0.9"
wasm-bindgen-test = "0.3.0"
[target.'cfg(target_os = "freebsd")'.dev-dependencies]
mio-aio = { version = "1", features = ["tokio"] }
mio-aio = { version = "2", features = ["tokio"] }
[target.'cfg(loom)'.dev-dependencies]
loom = { version = "0.7", features = ["futures", "checkpoint"] }
+27 -6
View File
@@ -9,17 +9,36 @@ use mio::Token;
use std::fmt;
use std::io;
use std::ops::{Deref, DerefMut};
use std::os::unix::io::AsRawFd;
use std::os::unix::prelude::RawFd;
use std::os::fd::{AsFd, BorrowedFd};
use std::os::unix::io::{AsRawFd, RawFd};
use std::task::{ready, Context, Poll};
/// Like [`mio::event::Source`], but for POSIX AIO only.
///
/// Tokio's consumer must pass an implementor of this trait to create a
/// [`Aio`] object.
/// [`Aio`] object. Implementors must implement at least one of [`AioSource::register`] and
/// [`AioSource::register_borrowed`].
pub trait AioSource {
/// Registers this AIO event source with Tokio's reactor.
fn register(&mut self, kq: RawFd, token: usize);
///
/// # Safety
///
/// It's memory-safe, but not I/O safe. If the file referenced by `kq` gets dropped, then this
/// source may end up notifying the wrong file.
#[deprecated(since = "1.52.0", note = "use register_borrowed instead")]
fn register(&mut self, _kq: RawFd, _token: usize) {
// This default implementation exists so new AioSource implementors that implement the
// register_borrowed method can compile without the need to implement register.
unimplemented!("Use AioSource::register_borrowed instead")
}
/// Registers this AIO event source with Tokio's reactor.
fn register_borrowed(&mut self, kq: BorrowedFd<'_>, token: usize) {
// This default implementation serves to provide backwards compatibility with AioSource
// implementors written before 1.52.0 that only implemented the unsafe `register` method.
#[allow(deprecated)]
self.register(kq.as_raw_fd(), token)
}
/// Deregisters this AIO event source with Tokio's reactor.
fn deregister(&mut self);
@@ -37,7 +56,8 @@ impl<T: AioSource> Source for MioSource<T> {
interests: mio::Interest,
) -> io::Result<()> {
assert!(interests.is_aio() || interests.is_lio());
self.0.register(registry.as_raw_fd(), usize::from(token));
self.0
.register_borrowed(registry.as_fd(), usize::from(token));
Ok(())
}
@@ -53,7 +73,8 @@ impl<T: AioSource> Source for MioSource<T> {
interests: mio::Interest,
) -> io::Result<()> {
assert!(interests.is_aio() || interests.is_lio());
self.0.register(registry.as_raw_fd(), usize::from(token));
self.0
.register_borrowed(registry.as_fd(), usize::from(token));
Ok(())
}
}
+7 -6
View File
@@ -10,6 +10,7 @@ use std::time::Duration;
use std::{
future::Future,
io::{self, ErrorKind, Read, Write},
os::fd::OwnedFd,
task::{Context, Waker},
};
@@ -69,7 +70,7 @@ impl AsRawFd for FileDescriptor {
impl Read for &FileDescriptor {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
read(self.fd.as_raw_fd(), buf).map_err(io::Error::from)
read(&self.fd, buf).map_err(io::Error::from)
}
}
@@ -99,10 +100,10 @@ impl Write for FileDescriptor {
}
}
fn set_nonblocking(fd: RawFd) {
fn set_nonblocking(fd: &OwnedFd) {
use nix::fcntl::{OFlag, F_GETFL, F_SETFL};
let flags = nix::fcntl::fcntl(fd, F_GETFL).expect("fcntl(F_GETFD)");
let flags = nix::fcntl::fcntl(fd, F_GETFL).expect("fcntl(F_GETFL)");
if flags < 0 {
panic!(
@@ -114,7 +115,7 @@ fn set_nonblocking(fd: RawFd) {
let flags = OFlag::from_bits_truncate(flags) | OFlag::O_NONBLOCK;
nix::fcntl::fcntl(fd, F_SETFL(flags)).expect("fcntl(F_SETFD)");
nix::fcntl::fcntl(fd, F_SETFL(flags)).expect("fcntl(F_SETFL)");
}
fn socketpair() -> (FileDescriptor, FileDescriptor) {
@@ -129,8 +130,8 @@ fn socketpair() -> (FileDescriptor, FileDescriptor) {
.expect("socketpair");
let fds = (FileDescriptor { fd: fd_a }, FileDescriptor { fd: fd_b });
set_nonblocking(fds.0.fd.as_raw_fd());
set_nonblocking(fds.1.fd.as_raw_fd());
set_nonblocking(&fds.0.fd);
set_nonblocking(&fds.1.fd);
fds
}
+5 -4
View File
@@ -61,6 +61,7 @@ fn allocated_bytes() -> usize {
#[tokio::test]
async fn memory_leak_when_fd_closed_before_drop() {
use nix::sys::socket::{self, AddressFamily, SockFlag, SockType};
use std::os::fd::OwnedFd;
use std::os::unix::io::{AsRawFd, RawFd};
use std::sync::Arc;
use tokio::io::unix::AsyncFd;
@@ -83,7 +84,7 @@ async fn memory_leak_when_fd_closed_before_drop() {
}
}
fn set_nonblocking(fd: RawFd) {
fn set_nonblocking(fd: &OwnedFd) {
use nix::fcntl::{OFlag, F_GETFL, F_SETFL};
let flags = nix::fcntl::fcntl(fd, F_GETFL).expect("fcntl(F_GETFL)");
@@ -119,7 +120,7 @@ async fn memory_leak_when_fd_closed_before_drop() {
.unwrap();
let raw_fd = fd_a.as_raw_fd();
set_nonblocking(raw_fd);
set_nonblocking(&fd_a);
std::mem::forget(fd_a);
let wrapper = Arc::new(RawFdWrapper { fd: raw_fd });
@@ -148,7 +149,7 @@ async fn memory_leak_when_fd_closed_before_drop() {
.unwrap();
let raw_fd = fd_a.as_raw_fd();
set_nonblocking(raw_fd);
set_nonblocking(&fd_a);
std::mem::forget(fd_a);
let wrapper = Arc::new(RawFdWrapper { fd: raw_fd });
@@ -175,7 +176,7 @@ async fn memory_leak_when_fd_closed_before_drop() {
.unwrap();
let raw_fd = fd_a.as_raw_fd();
set_nonblocking(raw_fd);
set_nonblocking(&fd_a);
std::mem::forget(fd_a);
let wrapper = Arc::new(RawFdWrapper { fd: raw_fd });
+7 -7
View File
@@ -5,8 +5,8 @@ use mio_aio::{AioFsyncMode, SourceApi};
use std::{
future::Future,
io, mem,
os::fd::AsFd,
os::unix::io::{AsRawFd, RawFd},
os::fd::{AsFd, BorrowedFd},
os::unix::io::AsRawFd,
pin::{pin, Pin},
task::{Context, Poll},
};
@@ -21,7 +21,7 @@ mod aio {
struct TokioSource<'fd>(mio_aio::Source<nix::sys::aio::AioFsync<'fd>>);
impl<'fd> AioSource for TokioSource<'fd> {
fn register(&mut self, kq: RawFd, token: usize) {
fn register_borrowed(&mut self, kq: BorrowedFd<'_>, token: usize) {
self.0.register_raw(kq, token)
}
fn deregister(&mut self) {
@@ -81,10 +81,10 @@ mod aio {
}
impl AioSource for LlSource {
fn register(&mut self, kq: RawFd, token: usize) {
fn register_borrowed(&mut self, kq: BorrowedFd<'_>, token: usize) {
let mut sev: libc::sigevent = unsafe { mem::MaybeUninit::zeroed().assume_init() };
sev.sigev_notify = libc::SIGEV_KEVENT;
sev.sigev_signo = kq;
sev.sigev_signo = kq.as_raw_fd();
sev.sigev_value = libc::sigval {
sival_ptr: token as *mut libc::c_void,
};
@@ -222,10 +222,10 @@ mod lio {
}
impl<'a> AioSource for LioSource<'a> {
fn register(&mut self, kq: RawFd, token: usize) {
fn register_borrowed(&mut self, kq: BorrowedFd<'_>, token: usize) {
let mut sev: libc::sigevent = unsafe { mem::MaybeUninit::zeroed().assume_init() };
sev.sigev_notify = libc::SIGEV_KEVENT;
sev.sigev_signo = kq;
sev.sigev_signo = kq.as_raw_fd();
sev.sigev_value = libc::sigval {
sival_ptr: token as *mut libc::c_void,
};
+3 -3
View File
@@ -8,8 +8,8 @@ use tokio_test::{assert_err, assert_ok, assert_pending, assert_ready_ok};
use std::fs::File;
use std::io;
use std::os::fd::AsFd;
use std::os::unix::fs::OpenOptionsExt;
use std::os::unix::io::AsRawFd;
use std::path::{Path, PathBuf};
/// Helper struct which will clean up temporary files once dropped.
@@ -277,8 +277,8 @@ async fn from_file_detects_wrong_access_mode() -> io::Result<()> {
Ok(())
}
fn is_nonblocking<T: AsRawFd>(fd: &T) -> io::Result<bool> {
let flags = nix::fcntl::fcntl(fd.as_raw_fd(), nix::fcntl::F_GETFL)?;
fn is_nonblocking<T: AsFd>(fd: &T) -> io::Result<bool> {
let flags = nix::fcntl::fcntl(fd.as_fd(), nix::fcntl::F_GETFL)?;
Ok((flags & libc::O_NONBLOCK) != 0)
}