From c65e260d3c27530dd1dfbde22ef88baa80b7225d Mon Sep 17 00:00:00 2001 From: Farhan Syah Date: Sun, 18 Oct 2026 14:14:24 +0800 Subject: [PATCH 1/3] fix(eventfd): own wake descriptors through shared handles A core's wake eventfd closed when its thread exited, while the Control Plane notifier and the Tokio AsyncFd registrations still held the raw descriptor number. The kernel reuses the lowest free number, so a late wake could write into an unrelated file opened afterwards. The Data Plane EventFd and its notifiers now share the descriptors through an Arc of OwnedFd, so they close once, when the last holder drops. The bridge EventFd hands out BorrowedFd instead of RawFd, and the Tokio wrapper keeps an Arc to the wake pair for the life of its AsyncFd. The Linux eventfd is also created with EFD_CLOEXEC. Closes #419 --- nodedb-bridge/src/async_bridge.rs | 25 +-- nodedb-bridge/src/eventfd.rs | 21 +- nodedb-bridge/src/tokio_fd.rs | 82 ++++---- nodedb-bridge/tests/cross_runtime.rs | 4 +- nodedb/src/bootstrap/data_plane.rs | 4 +- nodedb/src/data/eventfd.rs | 290 ++++++++++++++++++--------- nodedb/src/data/runtime/spawn.rs | 101 +++++++++- 7 files changed, 365 insertions(+), 162 deletions(-) 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/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:?}"), + } + } +} From 3da1e1768e9abd9dffcfc51f2d5e67210ba3d84f Mon Sep 17 00:00:00 2001 From: Farhan Syah Date: Sun, 18 Oct 2026 14:14:24 +0800 Subject: [PATCH 2/3] fix(io): retire a stalled io_uring ring before closing its files When a batched read stalled, queued submission entries still carried the raw descriptors of files that close as the batch returns. Entering the ring again could submit reads against reused descriptor numbers. A stalled ring is now replaced, or marked poisoned and never entered again, with reads falling back to a sequential path. Per-read state and the stall cleanup helpers move into uring_read_state. --- nodedb/src/data/io/mod.rs | 1 + nodedb/src/data/io/uring_read_state.rs | 107 +++++++++++++++++++++++ nodedb/src/data/io/uring_reader.rs | 114 +++++++++++++------------ 3 files changed, 168 insertions(+), 54 deletions(-) create mode 100644 nodedb/src/data/io/uring_read_state.rs 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::*; From 4183bdba1668c6b6f0f2edd8055d362c9137b353 Mon Sep 17 00:00:00 2001 From: Farhan Syah Date: Mon, 19 Oct 2026 04:36:52 +0800 Subject: [PATCH 3/3] test: replace wall-clock throughput assertions with correctness checks Timing thresholds in unit tests fail under machine load. The KV update test now checks one entry per key and the last written value. The timeseries ingest and aggregation tests drop their timers and keep the row-count and result assertions. Throughput belongs in benchmarks. --- nodedb/src/engine/kv/engine_write.rs | 31 +++++++++++++------- nodedb/src/engine/timeseries/verification.rs | 30 ++++--------------- 2 files changed, 26 insertions(+), 35 deletions(-) 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() - ); } }