diff --git a/h1/src/conn.rs b/h1/src/conn.rs index 41f8240..e552780 100644 --- a/h1/src/conn.rs +++ b/h1/src/conn.rs @@ -312,6 +312,9 @@ async fn send_region( tracing::trace!(offset, len, "sending a file region"); let BufResult(result, _file) = writer.send_file_all(file, offset, size).await; result?; + // With the record above, a log tells a region the socket took whole + // from one that never finished. + tracing::trace!(offset, len, "sent a file region"); outbox.extend(encoder.region_end()); Ok(()) } diff --git a/io/src/compio.rs b/io/src/compio.rs index 175ffd7..b9ad0b7 100644 --- a/io/src/compio.rs +++ b/io/src/compio.rs @@ -435,6 +435,10 @@ impl_socket!(net::UnixStream, zero_copy: false); /// A connected stream socket. pub struct Stream { inner: S, + /// The socket `inner`'s descriptor names, which every use of the + /// descriptor's number checks it still does (debug builds). + #[cfg(all(unix, debug_assertions))] + identity: crate::sys::unix::Identity, /// The pipe file transfers splice through, kept between transfers. A /// transfer takes it out while data may sit in it, so one that fails or /// is cancelled midway drops it instead of leaving stale bytes behind. @@ -472,6 +476,11 @@ impl Stream { #[cfg(not(target_os = "linux"))] let _ = io; Self { + #[cfg(all(unix, debug_assertions))] + identity: crate::sys::unix::Identity::of(std::os::fd::AsRawFd::as_raw_fd( + &inner.as_fd(), + )) + .expect("a stream's socket is open when the stream is made"), #[cfg(target_os = "linux")] file: files::File::new(inner.shared_fd(), io.file_table()), inner, @@ -498,9 +507,23 @@ impl Stream { } #[cfg(not(target_os = "linux"))] { + self.fd(); ::compio::runtime::submit(op).await } } + + /// The socket's descriptor, for a call that names it by its number. + /// + /// Debug builds check first that the number still names the stream's + /// socket, and panic if it does not (zen-rs/http-kit#159). + #[cfg(unix)] + #[track_caller] + fn fd(&self) -> std::os::fd::BorrowedFd<'_> { + let fd = self.inner.as_fd(); + #[cfg(debug_assertions)] + self.identity.check(std::os::fd::AsRawFd::as_raw_fd(&fd)); + fd + } } impl Stream { @@ -638,7 +661,10 @@ impl fmt::Debug for Stream { #[cfg(unix)] impl std::os::fd::AsFd for Stream { fn as_fd(&self) -> std::os::fd::BorrowedFd<'_> { - self.inner.as_fd() + let fd = self.inner.as_fd(); + #[cfg(debug_assertions)] + self.identity.check(std::os::fd::AsRawFd::as_raw_fd(&fd)); + fd } } @@ -917,11 +943,7 @@ mod splice { stream.pipe.set(Some(pipe)); return BufResult(Ok(filled), file); } - match crate::sys::pipe::splice( - pipe.read.as_fd(), - stream.inner.as_fd(), - filled - drained, - ) { + match crate::sys::pipe::splice(pipe.read.as_fd(), stream.fd(), filled - drained) { Ok(0) => { break io::Error::new(io::ErrorKind::WriteZero, "stream accepted no bytes"); } @@ -950,7 +972,7 @@ mod splice { /// Makes the socket `O_NONBLOCK`, once. fn nonblocking(stream: &Stream) -> io::Result<()> { if !stream.nonblocking.get() { - crate::sys::pipe::set_nonblocking(stream.inner.as_fd())?; + crate::sys::pipe::set_nonblocking(stream.fd())?; stream.nonblocking.set(true); } Ok(()) @@ -994,7 +1016,7 @@ mod splice { } let len = max.min(pipe.capacity()); loop { - match crate::sys::pipe::splice(stream.inner.as_fd(), pipe.write_fd(), len) { + match crate::sys::pipe::splice(stream.fd(), pipe.write_fd(), len) { Ok(0) => return Ok(Spliced::End(pipe)), Ok(n) => return Ok(Spliced::Filled(FilledPipe::new(pipe, n))), Err(e) if e.kind() == io::ErrorKind::WouldBlock => { @@ -1015,7 +1037,7 @@ mod splice { nonblocking(stream)?; let mut left = filled.len(); while left > 0 { - match crate::sys::pipe::splice(filled.pipe().read_fd(), stream.inner.as_fd(), left) { + match crate::sys::pipe::splice(filled.pipe().read_fd(), stream.fd(), left) { Ok(0) => { return Err(io::Error::new( io::ErrorKind::WriteZero, @@ -1188,6 +1210,10 @@ mod files { slot: Rc, /// The socket's descriptor, which the slot replaces in an entry. fd: RawFd, + /// The socket `fd` names, which an operation naming the descriptor + /// checks it still does (debug builds, zen-rs/http-kit#159). + #[cfg(debug_assertions)] + identity: crate::sys::unix::Identity, } /// What a stream and its slot's keeper share. @@ -1238,7 +1264,13 @@ mod files { } }); } - Self { slot, fd } + Self { + slot, + fd, + #[cfg(debug_assertions)] + identity: crate::sys::unix::Identity::of(fd) + .expect("a stream's socket is open when the stream is made"), + } } fn slot(&self) -> io::Result> { @@ -1258,12 +1290,22 @@ mod files { /// `op`, naming the socket by its slot when it has one; or, with /// `op` back, the error taking a slot failed with. + /// + /// Without a slot `op` names the descriptor's number, which debug + /// builds check still names the socket. + #[track_caller] pub(super) fn address(&self, op: T) -> Result, (io::Error, T)> { match self.slot() { - Ok(slot) => Ok(Addressed { - op, - slot: slot.map(|index| (index, self.fd)), - }), + Ok(slot) => { + #[cfg(debug_assertions)] + if slot.is_none() { + self.identity.check(self.fd); + } + Ok(Addressed { + op, + slot: slot.map(|index| (index, self.fd)), + }) + } Err(error) => Err((error, op)), } } diff --git a/io/src/sys/unix.rs b/io/src/sys/unix.rs index 1618259..7692498 100644 --- a/io/src/sys/unix.rs +++ b/io/src/sys/unix.rs @@ -32,3 +32,73 @@ pub(crate) fn local_addr(fd: impl AsFd) -> io::Result { pub(crate) fn peer_addr(fd: impl AsFd) -> io::Result { UnixStream::from(duplicate(fd)?).peer_addr() } + +/// The open file a descriptor names, as `fstat(2)` identifies it: its device +/// and inode, which stay the same for as long as the descriptor names that +/// file and differ for any other file open at the same time. +/// +/// A stream records its socket's at creation and compares before each use +/// of the descriptor's number, so that a number closed and reused behind the +/// stream's back (zen-rs/http-kit#159) fails at the use instead of moving +/// bytes to whatever file now holds it. It costs an `fstat` per use, so only +/// debug builds keep it. +#[cfg(all(feature = "compio", debug_assertions))] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct Identity { + dev: libc::dev_t, + ino: libc::ino_t, + /// The file type bits of `st_mode`, which name what a reused number + /// now holds. + kind: libc::mode_t, +} + +#[cfg(all(feature = "compio", debug_assertions))] +impl Identity { + /// The identity of the file `fd` names. A number is all `fstat` needs: + /// one that names nothing is reported as `EBADF`. + pub(crate) fn of(fd: std::os::fd::RawFd) -> io::Result { + let mut stat = std::mem::MaybeUninit::::uninit(); + // SAFETY: `stat` is valid for the write of one `struct stat`; the + // kernel reads nothing through `fd` but its number. + if unsafe { libc::fstat(fd, stat.as_mut_ptr()) } != 0 { + return Err(io::Error::last_os_error()); + } + // SAFETY: `fstat` succeeded, so it filled `stat`. + let stat = unsafe { stat.assume_init() }; + Ok(Self { + dev: stat.st_dev, + ino: stat.st_ino, + kind: stat.st_mode & libc::S_IFMT, + }) + } + + /// Panics unless `fd` still names the file `self` identifies. + #[track_caller] + pub(crate) fn check(self, fd: std::os::fd::RawFd) { + match Self::of(fd) { + Ok(now) if now == self => {} + Ok(now) => panic!( + "descriptor {fd} no longer names the stream's socket ({self:?}): it names a {} \ + ({now:?}), so the number was closed and reused while the stream held it", + now.kind_name(), + ), + Err(error) => panic!( + "descriptor {fd} no longer names the stream's socket ({self:?}): {error}, so the \ + number was closed while the stream held it", + ), + } + } + + fn kind_name(self) -> &'static str { + match self.kind { + libc::S_IFSOCK => "socket", + libc::S_IFREG => "regular file", + libc::S_IFIFO => "pipe", + libc::S_IFCHR => "character device", + libc::S_IFDIR => "directory", + libc::S_IFBLK => "block device", + libc::S_IFLNK => "symbolic link", + _ => "file of an unknown type", + } + } +} diff --git a/io/tests/io.rs b/io/tests/io.rs index 089ea52..544d816 100644 --- a/io/tests/io.rs +++ b/io/tests/io.rs @@ -905,6 +905,46 @@ mod compio { server.join().expect("server thread"); } + /// A stream whose descriptor number is closed and reused behind its back + /// (here by `dup2` of a regular file over it, which does both at once) + /// fails its next use of the number in a debug build instead of moving + /// bytes into the file now holding it (zen-rs/http-kit#159). On Linux the + /// use is a file region's splice into the socket. + #[cfg(all(unix, debug_assertions))] + #[test] + #[should_panic(expected = "no longer names the stream's socket")] + fn a_reused_descriptor_number_fails_its_next_use() { + use http_kit_io::compio::Compio; + use std::os::fd::{AsFd as _, AsRawFd as _}; + + let listener = std::net::TcpListener::bind(super::loopback()).expect("bind"); + let client = std::net::TcpStream::connect(listener.local_addr().expect("local addr")) + .expect("connect"); + let (accepted, _) = listener.accept().expect("accept"); + let file = super::File::open(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/io.rs")) + .expect("open this test's source"); + run(async move { + let stream = Compio::default() + .tcp_stream(::compio::net::TcpStream::from_std(accepted).expect("adopt")); + let number = stream.as_fd().as_raw_fd(); + // SAFETY: both descriptors are open; `dup2` replaces the + // stream's number with a duplicate of the file's. + assert_eq!(unsafe { libc::dup2(file.as_raw_fd(), number) }, number); + #[cfg(target_os = "linux")] + { + use super::SendFile as _; + + let mut stream = stream; + let _ = stream.send_file(super::Rc::new(file), 0, 64).await; + } + #[cfg(not(target_os = "linux"))] + { + let _ = stream.as_fd(); + } + }); + drop(client); + } + /// How many submission queue entries the current thread's io_uring has /// been handed: the tail the kernel reports in the ring's fdinfo. #[cfg(target_os = "linux")]