From 13a0c8ba75c186ebc60cfad2a204d41719c4d375 Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Fri, 16 Oct 2026 13:48:50 +0800 Subject: [PATCH] fix(runtime): keep a core's wake descriptor open while a notifier can still write EventFdNotifier was Copy and held only the raw descriptor number of the eventfd that wakes a data-plane core, while the core thread's EventFd closed that descriptor in Drop. When a core thread ended, for example on the panic at CoreLoop open, the Control Plane kept its notifier copies and kept calling notify(). The kernel had already given the freed number to the next open(), so each notify() wrote 1u64 into whichever file held it. In a single-node boot that file was a catalog or Raft redb file, and the write destroyed its magic number. EventFd and EventFdNotifier now share the descriptor through Arc. The notifier is Clone, no longer Copy, and the descriptor closes when the core and the last notifier are gone, so its number cannot be reused while a notifier can still write. The two Drop impls and the unsafe Send and Sync impls go away: OwnedFd already provides them. The self-pipe backend gets the same ownership. One call site changes: the data plane bootstrap clones the notifier it hands to the dispatcher. Two tests: a file opened after the core drops cannot receive the descriptor number and stays empty after notify(), and the descriptor outlives the core until the last clone drops. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01Pv7TPc3Q7xPYXWzocBbZbq --- nodedb/src/bootstrap/data_plane.rs | 2 +- nodedb/src/data/eventfd.rs | 169 ++++++++++++++++++++--------- 2 files changed, 118 insertions(+), 53 deletions(-) diff --git a/nodedb/src/bootstrap/data_plane.rs b/nodedb/src/bootstrap/data_plane.rs index 3b29aabba..4a76d4cba 100644 --- a/nodedb/src/bootstrap/data_plane.rs +++ b/nodedb/src/bootstrap/data_plane.rs @@ -376,7 +376,7 @@ pub fn spawn_data_plane_cores( } for (core_id, notifier) in ¬ifiers { - dispatcher.set_notifier(*core_id, *notifier); + dispatcher.set_notifier(*core_id, notifier.clone()); } 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..3283d2962 100644 --- a/nodedb/src/data/eventfd.rs +++ b/nodedb/src/data/eventfd.rs @@ -13,8 +13,13 @@ //! machines. This is a dev/correctness path, not a performance path. //! - **Non-Unix (wasm, Windows)**: unsupported; `EventFd::new()` errors. +#[cfg(unix)] +use std::os::fd::{AsRawFd as _, FromRawFd as _, OwnedFd}; #[cfg(unix)] use std::os::unix::io::RawFd; +#[cfg(unix)] +use std::sync::Arc; + #[cfg(not(unix))] type RawFd = i32; @@ -23,13 +28,15 @@ type RawFd = i32; /// `!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 with every `EventFdNotifier`: the descriptor closes when the core + // and the last notifier are gone, never while a notifier can still write. #[cfg(target_os = "linux")] - fd: RawFd, + fd: Arc, // Self-pipe: read end lives with the core, the notifier writes to `write_fd`. #[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: Arc, } impl EventFd { @@ -43,7 +50,10 @@ impl EventFd { if fd < 0 { return Err(crate::Error::Io(std::io::Error::last_os_error())); } - Ok(Self { fd }) + // SAFETY: `fd` was just returned by eventfd2 and nothing else owns it. + Ok(Self { + fd: Arc::new(unsafe { OwnedFd::from_raw_fd(fd) }), + }) } #[cfg(all(unix, not(target_os = "linux")))] { @@ -52,19 +62,18 @@ impl EventFd { if unsafe { libc::pipe(fds.as_mut_ptr()) } < 0 { return Err(crate::Error::Io(std::io::Error::last_os_error())); } + // SAFETY: both fds were just returned by pipe() and nothing else owns + // them. The owners close both if the flag setup below fails. + let (read_fd, write_fd) = + unsafe { (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[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)); - } + 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], + read_fd, + write_fd: Arc::new(write_fd), }) } #[cfg(not(unix))] @@ -80,11 +89,11 @@ impl EventFd { pub fn as_raw_fd(&self) -> RawFd { #[cfg(target_os = "linux")] { - self.fd + self.fd.as_raw_fd() } #[cfg(all(unix, not(target_os = "linux")))] { - self.read_fd + self.read_fd.as_raw_fd() } #[cfg(not(unix))] { @@ -103,7 +112,7 @@ impl EventFd { // SAFETY: reading 8 bytes from an eventfd is the documented API. let ret = unsafe { libc::read( - self.fd, + self.fd.as_raw_fd(), &mut buf as *mut u64 as *mut libc::c_void, std::mem::size_of::(), ) @@ -120,7 +129,7 @@ impl EventFd { // SAFETY: reading into a valid local buffer from a non-blocking fd. let ret = unsafe { libc::read( - self.read_fd, + self.read_fd.as_raw_fd(), buf.as_mut_ptr() as *mut libc::c_void, buf.len(), ) @@ -169,11 +178,15 @@ impl EventFd { pub fn notifier(&self) -> EventFdNotifier { #[cfg(target_os = "linux")] { - EventFdNotifier { fd: self.fd } + EventFdNotifier { + fd: Arc::clone(&self.fd), + } } #[cfg(all(unix, not(target_os = "linux")))] { - EventFdNotifier { fd: self.write_fd } + EventFdNotifier { + fd: Arc::clone(&self.write_fd), + } } #[cfg(not(unix))] { @@ -182,27 +195,6 @@ impl EventFd { } } -#[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); - } - } -} - /// Set `O_NONBLOCK` and `FD_CLOEXEC` on a freshly-created pipe fd. #[cfg(all(unix, not(target_os = "linux")))] fn set_pipe_flags(fd: RawFd) -> std::io::Result<()> { @@ -229,19 +221,19 @@ 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)] +/// The notifier shares ownership of the descriptor with the `EventFd`. A core +/// that exits, or panics while it opens its engines, drops its `EventFd`, and +/// the descriptor stays open for as long as any notifier is alive. A notifier +/// that held only the descriptor number would write into whatever file the +/// kernel next handed that number to, such as a database file opened after the +/// core died. The write (eventfd 8-byte / pipe 1-byte) is atomic and +/// thread-safe, so a notifier can be cloned across threads. +#[derive(Clone)] pub struct EventFdNotifier { #[cfg(unix)] - fd: RawFd, + fd: 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) { @@ -252,7 +244,7 @@ impl EventFdNotifier { // This is atomic and thread-safe per the Linux man page. unsafe { libc::write( - self.fd, + self.fd.as_raw_fd(), &val as *const u64 as *const libc::c_void, std::mem::size_of::(), ); @@ -265,7 +257,11 @@ impl EventFdNotifier { // (EAGAIN) is intentionally ignored — it already means a wake is // pending, so the level-triggered poll will still fire. unsafe { - libc::write(self.fd, &val as *const u8 as *const libc::c_void, 1); + libc::write( + self.fd.as_raw_fd(), + &val as *const u8 as *const libc::c_void, + 1, + ); } } } @@ -345,4 +341,73 @@ mod tests { fn assert_send_sync() {} assert_send_sync::(); } + + /// A core that exits drops its `EventFd`. The Control Plane keeps notifying + /// it until it learns the core is gone, so the descriptor must outlive the + /// core. A closed descriptor number goes to the next `open()`, and the + /// notifier's 8-byte write then lands in that file. + #[test] + #[cfg(target_os = "linux")] + fn notifier_outlives_the_core_without_reaching_a_reused_descriptor() { + use std::io::Read as _; + use std::os::fd::AsRawFd as _; + + let efd = EventFd::new().unwrap(); + let notifier = efd.notifier(); + let notifier_fd = notifier.fd.as_raw_fd(); + drop(efd); + + // The descriptor is still open: the kernel cannot hand its number out. + // SAFETY: F_GETFD only reads the descriptor's flags. + assert_ne!( + unsafe { libc::fcntl(notifier_fd, libc::F_GETFD) }, + -1, + "the notifier's descriptor was closed with the core" + ); + + let dir = tempfile::tempdir().unwrap(); + let mut files = Vec::new(); + for i in 0..8 { + let file = std::fs::File::create(dir.path().join(format!("db-{i}.redb"))).unwrap(); + assert_ne!( + file.as_raw_fd(), + notifier_fd, + "a file opened after the core died reused the notifier's descriptor" + ); + files.push(file); + } + + // Signals after the core died go to the eventfd, not to any file. + for _ in 0..4 { + notifier.notify(); + } + for i in 0..8 { + let mut contents = Vec::new(); + std::fs::File::open(dir.path().join(format!("db-{i}.redb"))) + .unwrap() + .read_to_end(&mut contents) + .unwrap(); + assert!(contents.is_empty(), "db-{i}.redb received notifier bytes"); + } + } + + #[test] + fn notifier_clones_keep_the_descriptor_until_the_last_one_drops() { + let efd = EventFd::new().unwrap(); + let first = efd.notifier(); + let second = first.clone(); + drop(efd); + drop(first); + // `second` still signals without error and the descriptor is valid. + second.notify(); + #[cfg(target_os = "linux")] + { + use std::os::fd::AsRawFd as _; + // SAFETY: F_GETFD only reads the descriptor's flags. + assert_ne!( + unsafe { libc::fcntl(second.fd.as_raw_fd(), libc::F_GETFD) }, + -1 + ); + } + } }