Keep epoll interest alive while a duplicate of the registered fd survives

litebox/src/fd/mod.rs

@@ -9,6 +9,7 @@
 )]
 
 use alloc::sync::Arc;
+use alloc::sync::Weak;
 use alloc::vec;
 use alloc::vec::Vec;
 use core::marker::PhantomData;
@@ -145,11 +146,33 @@ impl<Platform: RawSyncPrimitivesProvider> Descriptors<Platform> {
             // Unique, so we can just return it if allowed.
             if can_close_immediately(old.x.entry.read().as_subsystem::<Subsystem>()) {
                 fd.x.mark_as_closed();
-                let entry = Arc::into_inner(old.x)
-                    .map(|shared| RwLock::into_inner(shared.entry))
-                    .map(DescriptorEntry::into_subsystem_entry::<Subsystem>)
-                    .unwrap();
-                Some(CloseResult::Closed(entry))
+                match Arc::try_unwrap(old.x) {
+                    Ok(shared) => {
+                        let entry = DescriptorEntry::into_subsystem_entry::<Subsystem>(
+                            RwLock::into_inner(shared.entry),
+                        );
+                        Some(CloseResult::Closed(entry))
+                    }
+                    Err(x) => {
+                        // The strong count was 1 above, but a lock-free
+                        // `WeakEntryHandle::upgrade` on another thread (e.g. an
+                        // epoll re-poll of a still-registered interest) can
+                        // transiently re-share the entry before we take
+                        // ownership. Fall back to the shared path: duplicate the
+                        // descriptor so it is closed once that reference drops,
+                        // rather than dropping the entry without an orderly
+                        // close.
+                        let replaced = self.entries[idx].replace(IndividualEntry {
+                            x,
+                            metadata: old.metadata,
+                        });
+                        assert!(replaced.is_none());
+                        Some(CloseResult::Duplicated(TypedFd {
+                            _phantom: PhantomData,
+                            x: OwnedFd::new(idx),
+                        }))
+                    }
+                }
             } else {
                 // Put it back
                 let old = self.entries[idx].replace(old);
@@ -518,11 +541,20 @@ impl<Platform: RawSyncPrimitivesProvider> Descriptors<Platform> {
     }
 }
 
+/// An opaque, stable identity for a descriptor's entry (its open file
+/// description).
+///
+/// Equal keys denote the same entry. A key is stable across `dup` and for the
+/// entry's lifetime, so it can identify an entry (for example as a map key)
+/// without dereferencing anything.
+#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct EntryStableKey(usize);
+
 /// A handle to a descriptor entry (via [`Descriptors::entry_handle`]) that can be used without
 /// maintaining access to the descriptor table itself.
 pub struct EntryHandle<Platform: RawSyncPrimitivesProvider, Subsystem: FdEnabledSubsystem>(
     Arc<SharedEntry<Platform>>,
-    PhantomData<Subsystem>,
+    PhantomData<fn(Subsystem) -> Subsystem>,
 );
 impl<Platform: RawSyncPrimitivesProvider, Subsystem: FdEnabledSubsystem>
     EntryHandle<Platform, Subsystem>
@@ -556,6 +588,57 @@ impl<Platform: RawSyncPrimitivesProvider, Subsystem: FdEnabledSubsystem>
     pub fn with_entry_mut<R>(&self, f: impl FnOnce(&mut Subsystem::Entry) -> R) -> R {
         f(self.0.entry.write().as_subsystem_mut::<Subsystem>())
     }
+
+    /// Apply `f` on metadata at the entry, if it exists.
+    ///
+    /// In contrast to [`Descriptors::with_metadata`], this obtains entry-level metadata.
+    /// For FD-specific metadata, one necessarily needs the specific FD.
+    pub fn with_entry_metadata<T, R>(&self, f: impl FnOnce(&T) -> R) -> Option<R>
+    where
+        T: core::any::Any + Clone + Send + Sync,
+    {
+        self.0.entry.read().metadata.get::<T>().map(f)
+    }
+
+    /// An opaque, stable identity for this entry (see [`EntryStableKey`]).
+    #[must_use]
+    pub fn stable_key(&self) -> EntryStableKey {
+        EntryStableKey(Arc::as_ptr(&self.0).addr())
+    }
+
+    /// Obtains a non-owning [`WeakEntryHandle`] to this entry.
+    #[must_use]
+    pub fn downgrade(&self) -> WeakEntryHandle<Platform, Subsystem> {
+        WeakEntryHandle(Arc::downgrade(&self.0), PhantomData)
+    }
+}
+
+/// A weak reference to a descriptor entry.
+pub struct WeakEntryHandle<Platform: RawSyncPrimitivesProvider, Subsystem: FdEnabledSubsystem>(
+    Weak<SharedEntry<Platform>>,
+    PhantomData<fn(Subsystem) -> Subsystem>,
+);
+
+impl<Platform: RawSyncPrimitivesProvider, Subsystem: FdEnabledSubsystem>
+    WeakEntryHandle<Platform, Subsystem>
+{
+    /// Upgrades to a strong [`EntryHandle`] if the entry is still alive.
+    #[must_use]
+    pub fn upgrade(&self) -> Option<EntryHandle<Platform, Subsystem>> {
+        self.0
+            .upgrade()
+            .map(|entry| EntryHandle(entry, PhantomData))
+    }
+
+    /// An opaque, stable identity for this entry (see [`EntryStableKey`]).
+    ///
+    /// A [`WeakEntryHandle`] keeps the underlying allocation reserved even after
+    /// the entry is closed, so this key is never reused for a different entry
+    /// while this handle exists.
+    #[must_use]
+    pub fn stable_key(&self) -> EntryStableKey {
+        EntryStableKey(self.0.as_ptr().addr())
+    }
 }
 
 /// Result of a [`Descriptors::close_and_duplicate_if_shared`] operation

litebox/src/pipes.rs

@@ -729,3 +729,13 @@ crate::fd::enable_fds_for_subsystem! {
     PipeEnd<Platform>;
     -> PipeFd<Platform>;
 }
+
+impl<Platform: RawSyncPrimitivesProvider + TimeProvider> DescriptorEntry<Platform> {
+    /// Runs `f` with the [`IOPollable`] backing this pipe end.
+    pub fn with_iopollable<R>(&self, f: impl FnOnce(&dyn IOPollable) -> R) -> R {
+        match &self.entry {
+            PipeEnd::Receiver(receiver) => f(receiver.as_ref()),
+            PipeEnd::Sender(sender) => f(sender.as_ref()),
+        }
+    }
+}

litebox_runner_linux_userland/tests/epoll_dup.c

@@ -0,0 +1,81 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+// Tests: an epoll interest survives closing the registered fd as long as a
+// duplicate referring to the same open file description remains open, matching
+// Linux epoll(7) semantics ("a file descriptor is removed from an interest
+// list only after all the file descriptors referring to the underlying open
+// file description have been closed").
+
+#define _GNU_SOURCE
+#include <errno.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <string.h>
+#include <sys/epoll.h>
+#include <sys/eventfd.h>
+#include <unistd.h>
+
+#define TEST_ASSERT(cond, msg)                                                \
+    do {                                                                      \
+        if (!(cond)) {                                                        \
+            fprintf(stderr, "FAIL: %s (line %d): %s (errno=%d: %s)\n",        \
+                    __func__, __LINE__, msg, errno, strerror(errno));         \
+            return 1;                                                         \
+        }                                                                     \
+    } while (0)
+
+int main(void) {
+    int efd = eventfd(0, EFD_CLOEXEC);
+    TEST_ASSERT(efd >= 0, "eventfd failed");
+
+    int epfd = epoll_create1(EPOLL_CLOEXEC);
+    TEST_ASSERT(epfd >= 0, "epoll_create1 failed");
+
+    struct epoll_event ev;
+    memset(&ev, 0, sizeof(ev));
+    ev.events = EPOLLIN;
+    ev.data.u64 = 0x42;
+    TEST_ASSERT(epoll_ctl(epfd, EPOLL_CTL_ADD, efd, &ev) == 0,
+                "epoll_ctl ADD failed");
+
+    // The duplicate shares the same open file description as efd.
+    int dupfd = dup(efd);
+    TEST_ASSERT(dupfd >= 0, "dup failed");
+
+    // Close the originally-registered fd. The interest must survive because
+    // dupfd still refers to the same open file description.
+    TEST_ASSERT(close(efd) == 0, "close original failed");
+
+    // Make the description readable through the surviving duplicate.
+    uint64_t one = 1;
+    TEST_ASSERT(write(dupfd, &one, sizeof(one)) == (ssize_t)sizeof(one),
+                "write via dup failed");
+
+    // The registration must still be reported, carrying its original data.
+    struct epoll_event out[4];
+    memset(out, 0, sizeof(out));
+    int n = epoll_wait(epfd, out, 4, 1000);
+    TEST_ASSERT(n == 1, "epoll_wait should report the surviving registration");
+    TEST_ASSERT((out[0].events & EPOLLIN) != 0, "expected EPOLLIN");
+    TEST_ASSERT(out[0].data.u64 == 0x42, "event data mismatch");
+
+    // The registration is durable: after draining and re-arming through the
+    // duplicate, a second wait still reports it.
+    uint64_t val = 0;
+    TEST_ASSERT(read(dupfd, &val, sizeof(val)) == (ssize_t)sizeof(val),
+                "read via dup failed");
+    TEST_ASSERT(val == 1, "unexpected eventfd value");
+    TEST_ASSERT(write(dupfd, &one, sizeof(one)) == (ssize_t)sizeof(one),
+                "second write via dup failed");
+    memset(out, 0, sizeof(out));
+    n = epoll_wait(epfd, out, 4, 1000);
+    TEST_ASSERT(n == 1, "epoll_wait should still report after re-arm");
+    TEST_ASSERT(out[0].data.u64 == 0x42, "event data mismatch after re-arm");
+
+    TEST_ASSERT(close(dupfd) == 0, "close dup failed");
+    TEST_ASSERT(close(epfd) == 0, "close epoll failed");
+
+    printf("epoll dup-survival: PASS\n");
+    return 0;
+}