diff --git a/nodedb-bridge/src/async_bridge.rs b/nodedb-bridge/src/async_bridge.rs index 5a52a1fc0..0e0064d28 100644 --- a/nodedb-bridge/src/async_bridge.rs +++ b/nodedb-bridge/src/async_bridge.rs @@ -23,6 +23,7 @@ //! with eventfd-based waking. use std::marker::PhantomData; +use std::os::fd::BorrowedFd; use std::sync::Arc; use crate::backpressure::{BackpressureController, PressureState}; @@ -181,13 +182,13 @@ impl ControlHandle { self.backpressure.state() } - /// Raw fd for the response-available signal (register with Tokio's AsyncFd). - pub fn response_wake_fd(&self) -> std::os::unix::io::RawFd { + /// Borrowed fd for the response-available signal. + pub fn response_wake_fd(&self) -> BorrowedFd<'_> { self.rsp_wake.consumer_wake.as_fd() } - /// Raw fd for the request-space-available signal. - pub fn request_space_fd(&self) -> std::os::unix::io::RawFd { + /// Borrowed fd for the request-space-available signal. + pub fn request_space_fd(&self) -> BorrowedFd<'_> { self.req_wake.producer_wake.as_fd() } } @@ -255,13 +256,13 @@ impl DataHandle { self.pressure() == PressureState::Suspended } - /// Raw fd for the request-available signal (register with TPC event loop). - pub fn request_wake_fd(&self) -> std::os::unix::io::RawFd { + /// Borrowed fd for the request-available signal (register with TPC event loop). + pub fn request_wake_fd(&self) -> BorrowedFd<'_> { self.req_wake.consumer_wake.as_fd() } - /// Raw fd for the response-space-available signal. - pub fn response_space_fd(&self) -> std::os::unix::io::RawFd { + /// Borrowed fd for the response-space-available signal. + pub fn response_space_fd(&self) -> BorrowedFd<'_> { self.rsp_wake.producer_wake.as_fd() } } @@ -297,13 +298,13 @@ impl PinnedDataHandle { self.inner.should_suspend() } - /// Raw fd for the request-available signal. - pub fn request_wake_fd(&self) -> std::os::unix::io::RawFd { + /// Borrowed fd for the request-available signal. + pub fn request_wake_fd(&self) -> BorrowedFd<'_> { self.inner.request_wake_fd() } - /// Raw fd for the response-space-available signal. - pub fn response_space_fd(&self) -> std::os::unix::io::RawFd { + /// Borrowed fd for the response-space-available signal. + pub fn response_space_fd(&self) -> BorrowedFd<'_> { self.inner.response_space_fd() } } diff --git a/nodedb-bridge/src/eventfd.rs b/nodedb-bridge/src/eventfd.rs index f9463dc1b..97494d0be 100644 --- a/nodedb-bridge/src/eventfd.rs +++ b/nodedb-bridge/src/eventfd.rs @@ -19,7 +19,7 @@ //! done by the caller. This module only provides raw fd-based signaling. use std::io; -use std::os::unix::io::{AsRawFd, FromRawFd, OwnedFd, RawFd}; +use std::os::fd::{AsFd, AsRawFd, BorrowedFd, FromRawFd, OwnedFd}; /// A cross-runtime wake signal backed by a Linux eventfd. /// @@ -147,26 +147,25 @@ impl EventFd { } } - /// Get the raw file descriptor for registration with an event loop. + /// Borrow the readable descriptor for registration with an event loop. /// - /// The caller can register this fd with: - /// - Tokio: `AsyncFd::new()` - /// - Glommio: `GlommioDma::from_raw_fd()` or similar - /// - io_uring: `IORING_OP_READ` on the fd - pub fn as_fd(&self) -> RawFd { + /// The borrow cannot outlive this `EventFd`, so a registration can never + /// keep a descriptor number past its close. A holder that must outlive + /// the borrow shares the owning `Arc` instead. + pub fn as_fd(&self) -> BorrowedFd<'_> { #[cfg(target_os = "linux")] { - self.fd.as_raw_fd() + self.fd.as_fd() } #[cfg(not(target_os = "linux"))] { - self.read_fd.as_raw_fd() + self.read_fd.as_fd() } } } #[cfg(not(target_os = "linux"))] -fn set_pipe_flags(fd: RawFd) -> io::Result<()> { +fn set_pipe_flags(fd: std::os::fd::RawFd) -> io::Result<()> { // SAFETY: fcntl is called with a valid fd returned by pipe(). let status_flags = unsafe { libc::fcntl(fd, libc::F_GETFL) }; if status_flags < 0 { @@ -268,6 +267,6 @@ mod tests { #[test] fn fd_is_valid() { let efd = EventFd::new().unwrap(); - assert!(efd.as_fd() >= 0); + assert!(efd.as_fd().as_raw_fd() >= 0); } } diff --git a/nodedb-bridge/src/tokio_fd.rs b/nodedb-bridge/src/tokio_fd.rs index 574eea042..09c3e0ddc 100644 --- a/nodedb-bridge/src/tokio_fd.rs +++ b/nodedb-bridge/src/tokio_fd.rs @@ -16,42 +16,68 @@ //! instance. When the eventfd becomes readable (the TPC core wrote to it), //! Tokio wakes the waiting task — no polling, no busy-waiting. -use std::os::unix::io::{AsRawFd, RawFd}; +use std::os::fd::{AsRawFd, RawFd}; +use std::sync::Arc; use tokio::io::Interest; use tokio::io::unix::AsyncFd; use crate::async_bridge::ControlHandle; use crate::error::{BridgeError, Result}; +use crate::eventfd::{EventFd, WakePair}; -/// Wraps a raw eventfd for use with Tokio's `AsyncFd`. +/// One eventfd of a shared `WakePair`, watched by a Tokio `AsyncFd`. /// -/// Needed because `AsyncFd` requires `AsRawFd` on the inner type. -struct RawFdWrapper { - fd: RawFd, +/// Holds an `Arc` to the pair, so the descriptor stays open for the whole +/// life of the `AsyncFd`. No drop order elsewhere can close it early. +struct WatchedWakeFd { + pair: Arc, + select: fn(&WakePair) -> &EventFd, } -impl AsRawFd for RawFdWrapper { +impl AsRawFd for WatchedWakeFd { fn as_raw_fd(&self) -> RawFd { - self.fd + (self.select)(&self.pair).as_fd().as_raw_fd() } } +/// The eventfd the consumer signals when the request queue frees a slot. +fn producer_wake(pair: &WakePair) -> &EventFd { + &pair.producer_wake +} + +/// The eventfd the producer signals when the response queue has data. +fn consumer_wake(pair: &WakePair) -> &EventFd { + &pair.consumer_wake +} + +/// Register one eventfd of `pair` with the current Tokio reactor. +fn watch( + pair: &Arc, + select: fn(&WakePair) -> &EventFd, +) -> std::io::Result> { + let watched = WatchedWakeFd { + pair: Arc::clone(pair), + select, + }; + // SAFETY: `watched` owns an `Arc` to the pair, so the descriptor stays + // open for the whole life of the `AsyncFd`. `select` is a fixed function + // of an immutable pair, so every `as_raw_fd` call returns the same fd. + unsafe { AsyncFd::register_with_interest(watched, Interest::READABLE) } + .map_err(|error| error.into_parts().1) +} + /// Async-capable Control Plane handle. /// /// Wraps `ControlHandle` with Tokio `AsyncFd` instances for non-blocking -/// async push/pop across the bridge. -/// -/// Fields drop in declaration order. Both `AsyncFd`s come before `inner`, which -/// owns the eventfds they watch, so each `AsyncFd` deregisters while its fd is -/// still open. +/// async push/pop across the bridge. Each `AsyncFd` shares ownership of +/// the eventfd it watches. pub struct AsyncControlHandle { /// AsyncFd for "request queue has space" signal (producer can push). - req_space_fd: AsyncFd, + req_space_fd: AsyncFd, /// AsyncFd for "response available" signal (consumer can pop). - rsp_ready_fd: AsyncFd, - /// The underlying synchronous handle. Private, so it cannot be moved out - /// and dropped while the `AsyncFd`s above still watch its fds. + rsp_ready_fd: AsyncFd, + /// The underlying synchronous handle. inner: ControlHandle, } @@ -60,28 +86,8 @@ impl AsyncControlHandle { /// /// Must be called from within a Tokio runtime context. pub fn new(handle: ControlHandle) -> std::io::Result { - // SAFETY: both fds belong to `handle`, which `Self` keeps in `inner` - // and drops after both `AsyncFd`s. Each fd stays open, and each - // wrapper returns the same fd, for the whole life of its `AsyncFd`. - let req_space_fd = unsafe { - AsyncFd::register_with_interest( - RawFdWrapper { - fd: handle.request_space_fd(), - }, - Interest::READABLE, - ) - } - .map_err(|error| error.into_parts().1)?; - // SAFETY: as above. - let rsp_ready_fd = unsafe { - AsyncFd::register_with_interest( - RawFdWrapper { - fd: handle.response_wake_fd(), - }, - Interest::READABLE, - ) - } - .map_err(|error| error.into_parts().1)?; + let req_space_fd = watch(&handle.req_wake, producer_wake)?; + let rsp_ready_fd = watch(&handle.rsp_wake, consumer_wake)?; Ok(Self { req_space_fd, diff --git a/nodedb-bridge/tests/cross_runtime.rs b/nodedb-bridge/tests/cross_runtime.rs index 523fadc8e..a78aec029 100644 --- a/nodedb-bridge/tests/cross_runtime.rs +++ b/nodedb-bridge/tests/cross_runtime.rs @@ -14,6 +14,7 @@ #![cfg(all(feature = "tokio", target_os = "linux"))] +use std::os::fd::AsRawFd; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::thread; @@ -35,7 +36,8 @@ fn tpc_event_loop( assert!(epoll_fd >= 0, "epoll_create1 failed"); // Register the request-available eventfd with epoll (edge-triggered). - let req_fd = data.request_wake_fd(); + // `data` outlives the epoll registration, so the raw number stays valid. + let req_fd = data.request_wake_fd().as_raw_fd(); let mut event = libc::epoll_event { events: (libc::EPOLLIN | libc::EPOLLET) as u32, u64: req_fd as u64, diff --git a/nodedb/src/bootstrap/data_plane.rs b/nodedb/src/bootstrap/data_plane.rs index 3b29aabba..0f578619d 100644 --- a/nodedb/src/bootstrap/data_plane.rs +++ b/nodedb/src/bootstrap/data_plane.rs @@ -375,8 +375,8 @@ pub fn spawn_data_plane_cores( notifiers.push((core_id, notifier)); } - for (core_id, notifier) in ¬ifiers { - dispatcher.set_notifier(*core_id, *notifier); + for (core_id, notifier) in notifiers { + dispatcher.set_notifier(core_id, notifier); } info!(num_cores, "data plane cores running (eventfd-driven)"); diff --git a/nodedb/src/data/eventfd.rs b/nodedb/src/data/eventfd.rs index 6abe24fab..aae1929a8 100644 --- a/nodedb/src/data/eventfd.rs +++ b/nodedb/src/data/eventfd.rs @@ -6,6 +6,13 @@ //! signals the Data Plane core to wake from `libc::poll`. This replaces the //! 50µs busy-poll sleep with an interrupt-driven wake. //! +//! ## Descriptor ownership +//! The `EventFd` and every `EventFdNotifier` share ownership of the wake +//! descriptors. The descriptors close when the last holder drops. A notify +//! after the core thread exits writes into the still-open wake channel, +//! which nothing reads. The kernel cannot hand the descriptor number to +//! another file while any notifier lives. +//! //! ## Platform backends //! - **Linux**: `eventfd` in `EFD_SEMAPHORE` mode — the production path. //! - **Other Unix (macOS, BSD)**: a self-pipe — slower per-wake than eventfd @@ -13,23 +20,59 @@ //! machines. This is a dev/correctness path, not a performance path. //! - **Non-Unix (wasm, Windows)**: unsupported; `EventFd::new()` errors. +use std::sync::Arc; + #[cfg(unix)] -use std::os::unix::io::RawFd; -#[cfg(not(unix))] -type RawFd = i32; +use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd}; -/// A file descriptor for cross-thread wake signaling. +/// The descriptors behind one core wake channel. /// -/// `!Send` and `!Sync` — each core owns its own EventFd on the Data Plane side. -/// The Control Plane holds a cloneable `EventFdNotifier` (which is `Send + Sync`). -pub struct EventFd { +/// Shared through `Arc` by the `EventFd` and its notifiers. `OwnedFd` closes +/// each descriptor exactly once, when the last `Arc` drops. +struct WakeFds { #[cfg(target_os = "linux")] - fd: RawFd, - // Self-pipe: read end lives with the core, the notifier writes to `write_fd`. + fd: OwnedFd, + // Self-pipe: the core reads `read_fd`, notifiers write `write_fd`. + // Both ends stay open together, so a write never hits a reader-less pipe. #[cfg(all(unix, not(target_os = "linux")))] - read_fd: RawFd, + read_fd: OwnedFd, #[cfg(all(unix, not(target_os = "linux")))] - write_fd: RawFd, + write_fd: OwnedFd, +} + +#[cfg(unix)] +impl WakeFds { + /// The descriptor the core polls and drains. + fn poll_fd(&self) -> RawFd { + #[cfg(target_os = "linux")] + { + self.fd.as_raw_fd() + } + #[cfg(not(target_os = "linux"))] + { + self.read_fd.as_raw_fd() + } + } + + /// The descriptor a notifier writes to. + fn wake_fd(&self) -> RawFd { + #[cfg(target_os = "linux")] + { + self.fd.as_raw_fd() + } + #[cfg(not(target_os = "linux"))] + { + self.write_fd.as_raw_fd() + } + } +} + +/// The Data Plane side of a core wake channel. +/// +/// Each core owns one `EventFd`, moved into its thread at spawn. The Control +/// Plane holds `EventFdNotifier` clones, which are `Send + Sync`. +pub struct EventFd { + fds: Arc, } impl EventFd { @@ -38,33 +81,41 @@ impl EventFd { #[cfg(target_os = "linux")] { // SAFETY: eventfd2 is a standard Linux syscall. EFD_SEMAPHORE makes - // each read decrement by 1, EFD_NONBLOCK prevents blocking reads. - let fd = unsafe { libc::eventfd(0, libc::EFD_NONBLOCK | libc::EFD_SEMAPHORE) }; - if fd < 0 { + // each read decrement by 1, EFD_NONBLOCK prevents blocking reads, + // EFD_CLOEXEC keeps the descriptor out of exec'd children. + let raw = unsafe { + libc::eventfd( + 0, + libc::EFD_NONBLOCK | libc::EFD_SEMAPHORE | libc::EFD_CLOEXEC, + ) + }; + if raw < 0 { return Err(crate::Error::Io(std::io::Error::last_os_error())); } - Ok(Self { fd }) + // SAFETY: eventfd returned a fresh descriptor that nothing else owns. + let fd = unsafe { OwnedFd::from_raw_fd(raw) }; + Ok(Self { + fds: Arc::new(WakeFds { fd }), + }) } #[cfg(all(unix, not(target_os = "linux")))] { - let mut fds = [0 as libc::c_int; 2]; - // SAFETY: `fds` points to two valid c_int slots for pipe() to fill. - if unsafe { libc::pipe(fds.as_mut_ptr()) } < 0 { + let mut raw = [0 as libc::c_int; 2]; + // SAFETY: `raw` points to two valid c_int slots for pipe() to fill. + if unsafe { libc::pipe(raw.as_mut_ptr()) } < 0 { return Err(crate::Error::Io(std::io::Error::last_os_error())); } + // SAFETY: pipe() returned two fresh descriptors that nothing else owns. + let (read_fd, write_fd) = + unsafe { (OwnedFd::from_raw_fd(raw[0]), OwnedFd::from_raw_fd(raw[1])) }; // Both ends non-blocking + close-on-exec: drain() must not block on // an empty pipe, and notify() must not block on a full one. - if let Err(e) = set_pipe_flags(fds[0]).and_then(|()| set_pipe_flags(fds[1])) { - // SAFETY: both fds were just returned by pipe() and are still open. - unsafe { - libc::close(fds[0]); - libc::close(fds[1]); - } - return Err(crate::Error::Io(e)); - } + // On error both `OwnedFd`s drop and close their descriptors. + set_pipe_flags(read_fd.as_raw_fd()) + .and_then(|()| set_pipe_flags(write_fd.as_raw_fd())) + .map_err(crate::Error::Io)?; Ok(Self { - read_fd: fds[0], - write_fd: fds[1], + fds: Arc::new(WakeFds { read_fd, write_fd }), }) } #[cfg(not(unix))] @@ -76,22 +127,6 @@ impl EventFd { } } - /// Get the raw fd for use with `libc::poll`. - pub fn as_raw_fd(&self) -> RawFd { - #[cfg(target_os = "linux")] - { - self.fd - } - #[cfg(all(unix, not(target_os = "linux")))] - { - self.read_fd - } - #[cfg(not(unix))] - { - -1 - } - } - /// Drain all pending notifications (non-blocking). /// /// Returns the number of signals accumulated since the last drain. @@ -101,9 +136,10 @@ impl EventFd { { let mut buf = 0u64; // SAFETY: reading 8 bytes from an eventfd is the documented API. + // `self.fds` keeps the descriptor open for the whole call. let ret = unsafe { libc::read( - self.fd, + self.fds.poll_fd(), &mut buf as *mut u64 as *mut libc::c_void, std::mem::size_of::(), ) @@ -117,10 +153,11 @@ impl EventFd { let mut total: u64 = 0; let mut buf = [0u8; 256]; loop { - // SAFETY: reading into a valid local buffer from a non-blocking fd. + // SAFETY: reading into a valid local buffer from a non-blocking + // fd that `self.fds` keeps open for the whole call. let ret = unsafe { libc::read( - self.read_fd, + self.fds.poll_fd(), buf.as_mut_ptr() as *mut libc::c_void, buf.len(), ) @@ -150,11 +187,11 @@ impl EventFd { #[cfg(unix)] { let mut pfd = libc::pollfd { - fd: self.as_raw_fd(), + fd: self.fds.poll_fd(), events: libc::POLLIN, revents: 0, }; - // SAFETY: standard poll syscall on a valid fd. + // SAFETY: standard poll syscall on a fd that `self.fds` keeps open. let ret = unsafe { libc::poll(&mut pfd, 1, timeout_ms) }; ret > 0 && (pfd.revents & libc::POLLIN) != 0 } @@ -166,39 +203,12 @@ impl EventFd { } /// Create a `Send + Sync` notifier handle for the Control Plane. + /// + /// The notifier shares ownership of the wake descriptors. They stay open + /// until this `EventFd` and every notifier drop. pub fn notifier(&self) -> EventFdNotifier { - #[cfg(target_os = "linux")] - { - EventFdNotifier { fd: self.fd } - } - #[cfg(all(unix, not(target_os = "linux")))] - { - EventFdNotifier { fd: self.write_fd } - } - #[cfg(not(unix))] - { - EventFdNotifier {} - } - } -} - -#[cfg(target_os = "linux")] -impl Drop for EventFd { - fn drop(&mut self) { - // SAFETY: `self.fd` is owned by this EventFd and closed exactly once. - unsafe { - libc::close(self.fd); - } - } -} - -#[cfg(all(unix, not(target_os = "linux")))] -impl Drop for EventFd { - fn drop(&mut self) { - // SAFETY: both pipe ends are owned by this EventFd and closed once. - unsafe { - libc::close(self.read_fd); - libc::close(self.write_fd); + EventFdNotifier { + fds: Arc::clone(&self.fds), } } } @@ -229,19 +239,14 @@ fn set_pipe_flags(fd: RawFd) -> std::io::Result<()> { /// A `Send + Sync` handle for signaling an `EventFd` from the Control Plane. /// -/// The underlying fd is owned by the `EventFd` on the Data Plane side. -/// The notifier only writes to it — it does not close the fd on drop. -#[derive(Clone, Copy)] +/// Every clone shares ownership of the wake descriptors with the `EventFd`. +/// A notify after the core thread exits lands in the still-open wake channel. +/// It never reaches a file that reused the descriptor number. +#[derive(Clone)] pub struct EventFdNotifier { - #[cfg(unix)] - fd: RawFd, + fds: Arc, } -// SAFETY: the underlying write (eventfd 8-byte / pipe 1-byte) is atomic and -// thread-safe, so the notifier can be shared and copied across threads. -unsafe impl Send for EventFdNotifier {} -unsafe impl Sync for EventFdNotifier {} - impl EventFdNotifier { /// Signal the Data Plane core to wake up. pub fn notify(&self) { @@ -249,10 +254,12 @@ impl EventFdNotifier { { let val: u64 = 1; // SAFETY: writing 8 bytes to an eventfd is the documented API. - // This is atomic and thread-safe per the Linux man page. + // The write is atomic and thread-safe per the Linux man page. + // `self.fds` keeps the descriptor open for the whole call. + // A full counter returns EAGAIN, which leaves a wake pending. unsafe { libc::write( - self.fd, + self.fds.wake_fd(), &val as *const u64 as *const libc::c_void, std::mem::size_of::(), ); @@ -261,13 +268,22 @@ impl EventFdNotifier { #[cfg(all(unix, not(target_os = "linux")))] { let val: u8 = 1; - // SAFETY: writing 1 byte to the self-pipe write end. A full pipe - // (EAGAIN) is intentionally ignored — it already means a wake is - // pending, so the level-triggered poll will still fire. + // SAFETY: writing 1 byte to the self-pipe write end, which + // `self.fds` keeps open for the whole call. A full pipe (EAGAIN) + // already means a wake is pending, so the level-triggered poll + // still fires. unsafe { - libc::write(self.fd, &val as *const u8 as *const libc::c_void, 1); + libc::write( + self.fds.wake_fd(), + &val as *const u8 as *const libc::c_void, + 1, + ); } } + #[cfg(not(unix))] + { + let _ = &self.fds; + } } } @@ -275,6 +291,13 @@ impl EventFdNotifier { mod tests { use super::*; + /// Whether `fd` names an open descriptor in this process. + fn is_open(fd: RawFd) -> bool { + // SAFETY: F_GETFD only reads descriptor flags. An unused number fails + // with EBADF and touches nothing. + unsafe { libc::fcntl(fd, libc::F_GETFD) >= 0 } + } + #[test] fn create_and_signal() { let efd = EventFd::new().unwrap(); @@ -345,4 +368,77 @@ mod tests { fn assert_send_sync() {} assert_send_sync::(); } + + /// A notify after the core's `EventFd` drops must not touch a file opened + /// afterwards. The notifier keeps the descriptor number out of reuse. + #[test] + fn notify_after_the_core_dropped_leaves_a_later_file_intact() { + let efd = EventFd::new().unwrap(); + let wake_fd = efd.fds.wake_fd(); + let notifier = efd.notifier(); + drop(efd); // The core thread exits. + + // A file opened now takes the lowest free number. + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("system.redb"); + let mut header = b"redb\x1a\n\xa9\r\n".to_vec(); + header.resize(4096, 0); + std::fs::write(&path, &header).unwrap(); + let file = std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(&path) + .unwrap(); + assert_ne!( + file.as_raw_fd(), + wake_fd, + "the kernel reused the wake descriptor number while a notifier lives" + ); + + notifier.notify(); // The Control Plane wakes a core that no longer exists. + drop(file); + + let after = std::fs::read(&path).unwrap(); + assert_eq!( + &after[..9], + &header[..9], + "redb header overwritten: {:02x?}", + &after[..16] + ); + assert_eq!(after, header, "file body modified by a notify"); + } + + /// The wake descriptors stay open while any holder lives and close when + /// the last one drops. + #[test] + fn descriptor_closes_only_when_the_last_holder_drops() { + let efd = EventFd::new().unwrap(); + let poll_fd = efd.fds.poll_fd(); + let wake_fd = efd.fds.wake_fd(); + let first = efd.notifier(); + let second = first.clone(); + + drop(efd); + assert!( + is_open(poll_fd) && is_open(wake_fd), + "closed with two notifiers alive" + ); + + drop(first); + assert!( + is_open(poll_fd) && is_open(wake_fd), + "closed with one notifier alive" + ); + second.notify(); + + drop(second); + assert!( + !is_open(poll_fd), + "poll descriptor leaked after the last holder dropped" + ); + assert!( + !is_open(wake_fd), + "wake descriptor leaked after the last holder dropped" + ); + } } diff --git a/nodedb/src/data/io/mod.rs b/nodedb/src/data/io/mod.rs index 8bf35c210..20a5959e4 100644 --- a/nodedb/src/data/io/mod.rs +++ b/nodedb/src/data/io/mod.rs @@ -4,6 +4,7 @@ pub mod aligned_buf; pub mod fadvise; pub mod io_metrics; pub mod prefetch; +pub mod uring_read_state; pub mod uring_reader; pub mod uring_seq_reader; pub mod uring_writer; diff --git a/nodedb/src/data/io/uring_read_state.rs b/nodedb/src/data/io/uring_read_state.rs new file mode 100644 index 000000000..61f460702 --- /dev/null +++ b/nodedb/src/data/io/uring_read_state.rs @@ -0,0 +1,107 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! Per-read bookkeeping for the batched io_uring reader, plus the cleanup +//! and fallback paths it takes when its ring stalls. + +#![cfg(target_os = "linux")] + +use std::path::Path; + +use super::aligned_buf::{ALIGNMENT, AlignedBuf, round_up}; + +/// Progress of one read through the submit/complete cycle. +#[derive(Clone, Copy)] +pub(super) enum ReadState { + /// Never accepted by the submission queue — the kernel never saw the buffer. + NotSubmitted, + /// Accepted by the submission queue, completion not yet reaped. + InFlight, + /// Completed successfully. + Done, + /// Completed with a kernel error. + Failed, +} + +/// Tracks which buffer a read uses. +#[derive(Clone, Copy)] +pub(super) enum BufSource { + Pool(usize), + Oversized(usize), + None, +} + +/// A pending read operation. +pub(super) struct PendingRead { + pub(super) index: usize, + pub(super) file: Option, + pub(super) size: usize, + pub(super) buf_source: BufSource, +} + +impl PendingRead { + pub(super) fn failed(index: usize) -> Self { + Self { + index, + file: None, + size: 0, + buf_source: BufSource::None, + } + } +} + +/// Round a read size up to ALIGNMENT. +#[inline] +pub(super) fn round_up_read(size: usize) -> usize { + round_up(size, ALIGNMENT) +} + +/// Leak every buffer the kernel can still write after a batch stalled. +/// +/// An in-flight pool slot gets a fresh buffer and returns to `free`. +/// Without a fresh buffer the slot stays withheld from later reads. +/// Oversized buffers of in-flight reads are leaked, the rest drop. +pub(super) fn leak_kernel_owned_buffers( + reads: &[PendingRead], + states: &[ReadState], + pool: &mut [AlignedBuf], + free: &mut Vec, + buf_size: usize, + oversized: Vec, +) { + let mut kernel_owned = vec![false; oversized.len()]; + for read in reads { + if !matches!(states.get(read.index), Some(ReadState::InFlight)) { + continue; + } + match read.buf_source { + BufSource::Pool(slot) => { + if let (Some(stale), Ok(fresh)) = (pool.get_mut(slot), AlignedBuf::new(buf_size)) { + std::mem::forget(std::mem::replace(stale, fresh)); + free.push(slot); + } + } + BufSource::Oversized(idx) => { + if let Some(owned) = kernel_owned.get_mut(idx) { + *owned = true; + } + } + BufSource::None => {} + } + } + for (buf, owned) in oversized.into_iter().zip(kernel_owned) { + if owned { + std::mem::forget(buf); + } + } +} + +/// Read each file with blocking `std::fs::read`. +/// +/// Used only while the reader has no usable ring. An unreadable file +/// yields an empty `Vec`, matching the io_uring path. +pub(super) fn read_sequential(paths: &[&Path]) -> Vec> { + paths + .iter() + .map(|path| std::fs::read(path).unwrap_or_default()) + .collect() +} diff --git a/nodedb/src/data/io/uring_reader.rs b/nodedb/src/data/io/uring_reader.rs index 3ccce94ce..068f192a0 100644 --- a/nodedb/src/data/io/uring_reader.rs +++ b/nodedb/src/data/io/uring_reader.rs @@ -32,10 +32,14 @@ use std::os::unix::io::AsRawFd; use std::time::Instant; #[cfg(target_os = "linux")] -use super::aligned_buf::{ALIGNMENT, AlignedBuf}; +use super::aligned_buf::AlignedBuf; use super::io_metrics::IoMetrics; #[cfg(target_os = "linux")] use super::io_metrics::{TIER_CRITICAL, TIER_HIGH, TIER_LOW}; +#[cfg(target_os = "linux")] +use super::uring_read_state::{ + BufSource, PendingRead, ReadState, leak_kernel_owned_buffers, read_sequential, round_up_read, +}; use crate::bridge::envelope::Priority; /// Queue depth for the io_uring instance. @@ -71,6 +75,9 @@ pub struct UringReader { buf_size: usize, /// Submission queue depth the ring was created with. queue_depth: u32, + /// Set when a stalled ring could not be replaced. A poisoned ring is + /// never entered again, so its queued SQEs never reach the kernel. + poisoned: bool, } #[cfg(not(target_os = "linux"))] @@ -111,6 +118,7 @@ impl UringReader { free, buf_size, queue_depth: queue_depth.max(1), + poisoned: false, }) } @@ -128,6 +136,9 @@ impl UringReader { if paths.is_empty() { return Vec::new(); } + if self.poisoned && !self.replace_ring() { + return read_sequential(paths); + } // Open files, determine sizes, assign buffers. let mut reads: Vec = Vec::with_capacity(paths.len()); @@ -235,6 +246,8 @@ impl UringReader { // submit/reap calls below only advance ring state, so a buffer // whose SQE is already in flight is untouched by them. File fds // stay valid until `reads` is dropped at the end of the function. + // A stalled ring is retired before that drop, so no queued SQE + // can reach the kernel after its file closes. let mut pushed = unsafe { self.ring.submission().push(&read_op).is_ok() }; if !pushed { self.submit_queued(&mut queued, &mut in_flight, &mut states); @@ -302,15 +315,59 @@ impl UringReader { } ReadState::InFlight => { // The ring stopped reporting completions, so the kernel may - // still write into this buffer. The pool slot is withheld - // rather than handed to a later read. + // still write into this buffer. `retire_stalled_batch` + // below decides what happens to it. } } } + if queued + in_flight > 0 { + self.retire_stalled_batch(&reads, &states, oversized); + } + results } + /// Stop using a ring that stopped delivering completions. + /// + /// Its queued SQEs carry raw fds of this batch's files, which close when + /// `read_files` returns. Entering the ring again would submit them + /// against whatever file reuses those numbers. The ring is replaced, or + /// poisoned so it is never entered again. Buffers the kernel can still + /// write are leaked, so a late completion never lands in freed memory. + fn retire_stalled_batch( + &mut self, + reads: &[PendingRead], + states: &[ReadState], + oversized: Vec, + ) { + self.poisoned = true; + self.replace_ring(); + leak_kernel_owned_buffers( + reads, + states, + &mut self.pool, + &mut self.free, + self.buf_size, + oversized, + ); + } + + /// Swap in a fresh ring. Dropping the old ring discards its queued SQEs. + /// + /// Returns `false` and leaves the reader poisoned when io_uring setup + /// fails. + fn replace_ring(&mut self) -> bool { + match io_uring::IoUring::new(self.queue_depth) { + Ok(fresh) => { + self.ring = fresh; + self.poisoned = false; + true + } + Err(_) => false, + } + } + /// Hand every queued SQE to the kernel, moving them to in-flight. /// /// A full completion queue can reject the submission, so ready CQEs are @@ -430,57 +487,6 @@ impl UringReader { } } -/// Progress of one read through the submit/complete cycle. -#[derive(Clone, Copy)] -#[cfg(target_os = "linux")] -enum ReadState { - /// Never accepted by the submission queue — the kernel never saw the buffer. - NotSubmitted, - /// Handed to the kernel, completion not yet reaped. - InFlight, - /// Completed successfully. - Done, - /// Completed with a kernel error. - Failed, -} - -/// Tracks which buffer a read uses. -#[derive(Clone, Copy)] -#[cfg(target_os = "linux")] -enum BufSource { - Pool(usize), - Oversized(usize), - None, -} - -/// A pending read operation. -#[cfg(target_os = "linux")] -struct PendingRead { - index: usize, - file: Option, - size: usize, - buf_source: BufSource, -} - -#[cfg(target_os = "linux")] -impl PendingRead { - fn failed(index: usize) -> Self { - Self { - index, - file: None, - size: 0, - buf_source: BufSource::None, - } - } -} - -/// Round a read size up to ALIGNMENT. -#[inline] -#[cfg(target_os = "linux")] -fn round_up_read(size: usize) -> usize { - super::aligned_buf::round_up(size, ALIGNMENT) -} - #[cfg(all(test, target_os = "linux"))] mod tests { use super::*; diff --git a/nodedb/src/data/runtime/spawn.rs b/nodedb/src/data/runtime/spawn.rs index b116b2caa..cc9f1b236 100644 --- a/nodedb/src/data/runtime/spawn.rs +++ b/nodedb/src/data/runtime/spawn.rs @@ -54,7 +54,9 @@ pub fn spawn_core( let data_dir = data_dir.to_path_buf(); - // Create eventfd and extract notifier before moving EventFd to core thread. + // The notifier shares ownership of the eventfd with the core thread. + // The descriptor stays open after the thread exits. A late wake lands + // in the dead core's eventfd, never in a reused descriptor number. let efd = EventFd::new().map_err(std::io::Error::other)?; let notifier = efd.notifier(); @@ -211,3 +213,100 @@ fn wire_core_dependencies(core: &mut CoreLoop, deps: WiredDependencies) { // 2b. Wire the quarantine registry for corrupt-segment detection. core.set_quarantine_registry(quarantine_registry); } + +#[cfg(all(test, target_os = "linux"))] +mod tests { + use std::sync::Arc; + + use super::*; + use crate::bridge::dispatch::Dispatcher; + use crate::bridge::dispatch::test_requests::make_request; + use crate::bridge::envelope::{ErrorCode, Status}; + + /// A core that panics while it opens its engines drops its `EventFd`. + /// Later Control Plane wakes for that core must not touch a file opened + /// after the panic, and the request must fail with "core-0 is gone". + #[test] + fn wakes_after_a_core_panicked_at_open_leave_a_later_file_intact() { + let dir = tempfile::tempdir().unwrap(); + // A regular file in place of the data directory makes every engine open fail. + let data_dir = dir.path().join("not-a-directory"); + std::fs::write(&data_dir, b"").unwrap(); + + let (mut dispatcher, data_sides) = Dispatcher::new(1, 64); + let data_side = data_sides.into_iter().next().unwrap(); + // no-plane-separation: the one-shot readiness Sender `SpawnCoreParams` requires; no tokio runtime runs here. + let (replay_done_tx, _replay_done_rx) = tokio::sync::oneshot::channel(); + let (handle, notifier) = spawn_core(SpawnCoreParams { + core_id: 0, + node_id: 1, + request_rx: data_side.request_rx, + response_tx: data_side.response_tx, + data_dir: &data_dir, + wal_records: Arc::from(Vec::new()), + tombstones: nodedb_wal::TombstoneSet::default(), + num_cores: 1, + compaction_config: crate::data::runtime::CoreCompactionConfig::default(), + system_metrics: None, + event_producer: None, + event_interest: crate::event::interest::EventInterest::new(), + governor: crate::data::executor::core_loop::test_governor(), + quiesce: None, + hlc: Arc::new(nodedb_types::OrdinalClock::new()), + array_catalog: crate::control::array_catalog::ArrayCatalog::handle(), + quarantine_registry: Arc::new(crate::storage::quarantine::QuarantineRegistry::new()), + maintenance_budget: Arc::new( + crate::control::maintenance::MaintenanceBudgetTracker::new(), + ), + doc_config_seed: Arc::new(Vec::new()), + vector_index_param_seed: Arc::new(Vec::new()), + columnar_schema_seed: Arc::new(Vec::new()), + replay_done: replay_done_tx, + }) + .unwrap(); + assert!( + handle.join().is_err(), + "the core thread must panic while it opens its engines" + ); + dispatcher.set_notifier(0, notifier); + + // The core is gone. A file opened now takes the lowest free number. + let path = dir.path().join("system.redb"); + let mut header = b"redb\x1a\n\xa9\r\n".to_vec(); + header.resize(4096, 0); + std::fs::write(&path, &header).unwrap(); + let file = std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(&path) + .unwrap(); + + dispatcher.wake_all_cores(); + dispatcher.dispatch(make_request(0)).unwrap(); + dispatcher.wake_all_cores(); + drop(file); + + let after = std::fs::read(&path).unwrap(); + assert_eq!( + after, header, + "a wake for the dead core modified a later file" + ); + + let responses = dispatcher.poll_responses(); + assert_eq!( + responses.len(), + 1, + "the request to the dead core must be answered" + ); + assert_eq!(responses[0].status, Status::Error); + match responses[0].error_code.as_deref() { + Some(ErrorCode::Internal { detail }) => { + assert!( + detail.contains("core-0 is gone"), + "unexpected detail: {detail}" + ); + } + other => panic!("expected a core-gone refusal, got {other:?}"), + } + } +} diff --git a/nodedb/src/engine/kv/engine_write.rs b/nodedb/src/engine/kv/engine_write.rs index 084607184..6fc58d8e1 100644 --- a/nodedb/src/engine/kv/engine_write.rs +++ b/nodedb/src/engine/kv/engine_write.rs @@ -701,14 +701,16 @@ mod tests { ); } + /// Repeated PUTs of existing keys update in place: one entry per key, and + /// a read returns the last value. Throughput lives in the KV benchmark, + /// never in a unit test, where machine load decides the outcome. #[test] - fn raw_put_timing() { + fn updates_of_existing_keys_keep_one_entry_each() { let mut e = make_engine(); let n = now(); let keys: Vec> = (0..10_000u32).map(|i| i.to_be_bytes().to_vec()).collect(); let value = [0u8; 64]; - // Warmup: insert all keys once. for key in &keys { e.put(KvPutParams { database_id: 0, @@ -723,26 +725,33 @@ mod tests { .expect("a bound row writes"); } - // Timed: 100K updates (keys already exist). - let iters = 100_000u64; - let start = std::time::Instant::now(); + // 100K updates: every key is rewritten ten times, the last with its + // own index as the value. + let iters = 100_000u32; for i in 0..iters { - let key = &keys[(i as usize) % 10_000]; + let key = &keys[(i as usize) % keys.len()]; + let last = i >= iters - keys.len() as u32; + let updated = i.to_be_bytes(); e.put(KvPutParams { database_id: 0, tenant_id: 1, collection: "b", key, - value: &value, + value: if last { &updated } else { &value }, ttl_ms: 0, now_ms: n, surrogate: row_surrogate(key), }) .expect("a bound row writes"); } - let elapsed = start.elapsed(); - let ns_per_op = elapsed.as_nanos() / iters as u128; - // 691 ns/op measured — well under document's 12μs. - assert!(ns_per_op < 5_000, "PUT too slow: {ns_per_op} ns/op"); + assert_eq!(e.collection_len(0, 1, "b"), keys.len()); + for (index, key) in keys.iter().enumerate() { + let expected = (iters - keys.len() as u32 + index as u32).to_be_bytes(); + assert_eq!( + e.get(0, 1, "b", key, n), + Some(expected.to_vec()), + "key {index}" + ); + } } } diff --git a/nodedb/src/engine/timeseries/verification.rs b/nodedb/src/engine/timeseries/verification.rs index 462256048..68e419a2a 100644 --- a/nodedb/src/engine/timeseries/verification.rs +++ b/nodedb/src/engine/timeseries/verification.rs @@ -545,17 +545,18 @@ mod tests { assert!(!orphan_dir.exists()); } - // ── Ingest throughput: 1M rows ─────────────────────────────────── + // ── Ingest: 1M rows ────────────────────────────────────────────── + /// One million rows over 100 series all land in the memtable. Throughput + /// lives in the ingest benchmark, never in a unit test. #[test] - fn ingest_throughput_1m_rows() { + fn ingest_1m_rows() { let mut mt = ColumnarMemtable::new_metric(ColumnarMemtableConfig { max_memory_bytes: 256 * 1024 * 1024, hard_memory_limit: 512 * 1024 * 1024, max_tag_cardinality: 100_000, }); - let start = std::time::Instant::now(); for i in 0..1_000_000u64 { mt.ingest_row( i % 100, @@ -566,29 +567,20 @@ mod tests { ) .unwrap(); } - let elapsed = start.elapsed(); - let rows_per_sec = 1_000_000.0 / elapsed.as_secs_f64(); - assert_eq!(mt.row_count(), 1_000_000); - // Should achieve at least 1M rows/sec on any modern CPU. - assert!( - rows_per_sec > 500_000.0, - "ingest too slow: {rows_per_sec:.0} rows/sec (expected >500K)" - ); } // ── SIMD aggregation: SUM over 10M f64 values ──────────────────── + /// SUM, COUNT, MIN and MAX over 10M values match a scalar reference. #[test] - fn simd_aggregation_performance() { + fn simd_aggregation_over_10m_values() { use crate::engine::timeseries::columnar_agg::aggregate_f64; // 10M values (80MB). let values: Vec = (0..10_000_000).map(|i| (i as f64) * 0.001).collect(); - let start = std::time::Instant::now(); let result = aggregate_f64(&values); - let elapsed = start.elapsed(); // Verify correctness: sum of 0..9,999,999 * 0.001 let expected_sum = (0..10_000_000u64).map(|i| i as f64 * 0.001).sum::(); @@ -603,15 +595,5 @@ mod tests { assert_eq!(result.count, 10_000_000); assert!((result.min - 0.0).abs() < f64::EPSILON); assert!((result.max - 9999.999).abs() < 0.001); - - // Should complete in <100ms on any modern CPU with SIMD (release mode). - // Debug builds are significantly slower due to no optimizations — use a - // generous threshold to avoid flaky test failures in CI debug runs. - let threshold_ms = if cfg!(debug_assertions) { 5000 } else { 500 }; - assert!( - elapsed.as_millis() < threshold_ms, - "SIMD aggregation too slow: {}ms for 10M values (threshold {threshold_ms}ms)", - elapsed.as_millis() - ); } }