fix(eventfd): close a wake descriptor only when its last holder drops

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<Req, Rsp> ControlHandle<Req, Rsp> {
         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<Req, Rsp> DataHandle<Req, Rsp> {
         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<Req, Rsp> PinnedDataHandle<Req, Rsp> {
         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()
     }
 }

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<WakePair>` 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);
     }
 }

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<WakePair>,
+    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<WakePair>,
+    select: fn(&WakePair) -> &EventFd,
+) -> std::io::Result<AsyncFd<WatchedWakeFd>> {
+    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<Req, Rsp> {
     /// AsyncFd for "request queue has space" signal (producer can push).
-    req_space_fd: AsyncFd<RawFdWrapper>,
+    req_space_fd: AsyncFd<WatchedWakeFd>,
     /// AsyncFd for "response available" signal (consumer can pop).
-    rsp_ready_fd: AsyncFd<RawFdWrapper>,
-    /// 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<WatchedWakeFd>,
+    /// The underlying synchronous handle.
     inner: ControlHandle<Req, Rsp>,
 }
 
@@ -60,28 +86,8 @@ impl<Req, Rsp> AsyncControlHandle<Req, Rsp> {
     ///
     /// Must be called from within a Tokio runtime context.
     pub fn new(handle: ControlHandle<Req, Rsp>) -> std::io::Result<Self> {
-        // 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,

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,

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 &notifiers {
-        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)");