Fix memory leak in task destruction (header not deallocated).

glommio/Cargo.toml

@@ -51,6 +51,7 @@ futures = "0"
 hdrhistogram = "7"
 pretty_env_logger = "0"
 rand = "0"
+rusty-fork = "0.3"
 tokio = { version = "1", default-features = false, features = ["rt", "macros", "rt-multi-thread", "net", "io-util", "time", "sync"] }
 tracing-subscriber = { version = "0", features = ["env-filter"] }
 

glommio/src/executor/mod.rs

@@ -2864,9 +2864,11 @@ impl ExecutorProxy {
 #[cfg(test)]
 mod test {
     use core::mem::MaybeUninit;
+    use rusty_fork::rusty_fork_test;
     use std::{
         cell::Cell,
         collections::HashMap,
+        fs,
         sync::{
             atomic::{AtomicUsize, Ordering},
             Arc, Mutex,
@@ -2881,12 +2883,90 @@ mod test {
 
     use crate::{
         enclose,
+        io::{DmaFile, OpenOptions, OwnedDmaFile},
         timer::{self, sleep, Timer},
         SharesManager,
     };
 
     use super::*;
 
+    fn eventfd_count() -> usize {
+        fs::read_dir("/proc/self/fd")
+            .expect("failed to enumerate this process's file descriptors")
+            .filter_map(|entry| {
+                let entry = entry.ok()?;
+                fs::read_link(entry.path()).ok()
+            })
+            .filter(|target| target == std::path::Path::new("anon_inode:[eventfd]"))
+            .count()
+    }
+
+    fn run_shared_channel_round() {
+        let (sender, receiver) = crate::channels::shared_channel::new_bounded(1);
+
+        let sender = LocalExecutorBuilder::default()
+            .io_memory(0)
+            .spawn(move || async move {
+                let sender = sender.connect().await;
+                let file = OpenOptions::new()
+                    .create_new(true)
+                    .read(true)
+                    .write(true)
+                    .tmpfile(true)
+                    .dma_open(std::env::temp_dir())
+                    .await
+                    .unwrap();
+                let file: OwnedDmaFile = file.into();
+                sender.send(file).await.unwrap();
+            })
+            .unwrap();
+
+        let receiver = LocalExecutorBuilder::default()
+            .io_memory(0)
+            .spawn(move || async move {
+                let receiver = receiver.connect().await;
+                let file: DmaFile = receiver.recv().await.unwrap().into();
+                assert!(file.read_at(0, 1).await.unwrap().is_empty());
+                file.close().await.unwrap();
+            })
+            .unwrap();
+
+        sender.join().unwrap();
+        receiver.join().unwrap();
+    }
+
+    // The fork is critical here as it makes sure that the eventfd_count check works regardless of other tests
+    // running in the same process (which is what happens when running with cargo test instead of cargo nextest).
+    rusty_fork_test! {
+        #[test]
+        fn executor_shutdown_does_not_leak_eventfds() {
+            // The disconnected notifier is a process-wide singleton. Initialize it before
+            // measuring so the baseline contains every eventfd that is expected to persist.
+            let _ = crate::sys::get_sleep_notifier_for(usize::MAX);
+            let initial_eventfds = eventfd_count();
+
+            // Run enough rounds to make the leak from #448 unambiguous, then check that
+            // additional executor shutdowns do not accumulate descriptors either.
+            for _ in 0..10 {
+                run_shared_channel_round();
+            }
+            assert_eq!(
+                eventfd_count(),
+                initial_eventfds,
+                "eventfds leaked after 10 rounds"
+            );
+
+            for _ in 0..90 {
+                run_shared_channel_round();
+            }
+            assert_eq!(
+                eventfd_count(),
+                initial_eventfds,
+                "eventfds leaked after 100 rounds"
+            );
+        }
+    }
+
     #[test]
     fn create_and_destroy_executor() {
         let mut var = Rc::new(RefCell::new(0));

glommio/src/task/raw.rs

@@ -407,8 +407,8 @@ where
 
     /// Cleans up task's resources and deallocates it.
     ///
-    /// The schedule function will be dropped, and the task will then get
-    /// deallocated. The task must be closed before this function is called.
+    /// The schedule function and header will be dropped, and the task will then
+    /// get deallocated. The task must be closed before this function is called.
     #[inline]
     unsafe fn destroy(ptr: *const ()) {
         dbg_context!(ptr, "destroy", {
@@ -422,6 +422,10 @@ where
             abort_on_panic(|| {
                 // Drop the schedule function.
                 (raw.schedule as *mut S).drop_in_place();
+
+                // Drop the header so resources owned by it, such as the executor's
+                // sleep notifier, are released before the task allocation is freed.
+                (raw.header as *mut Header).drop_in_place();
             });
 
             // Finally, deallocate the memory reserved by the task.

valgrind.supp

@@ -0,0 +1,26 @@
+# The process-global reactor registry removes each SleepNotifier entry, but the
+# hash map retains its bucket allocation for reuse until process exit.
+{
+   glommio_reactor_notifier_registry_capacity
+   Memcheck:Leak
+   match-leak-kinds: possible
+   fun:malloc
+   ...
+   fun:*hashbrown*RawTableInner*new_uninitialized*
+   ...
+   fun:*glommio*ReactorGlobalState*new_local_state
+}
+
+# The Rust test harness retains the std::sync::mpmc context for its main
+# thread-local test-event receiver until process exit.
+{
+   rust_test_harness_mpmc_context_tls
+   Memcheck:Leak
+   match-leak-kinds: possible
+   fun:malloc
+   ...
+   fun:current_or_unnamed
+   fun:*std4sync4mpmc7context*Context3new
+   ...
+   fun:*test*run_tests*
+}