diff --git a/glommio/Cargo.toml b/glommio/Cargo.toml index 8b6c381fc..443258c79 100755 --- a/glommio/Cargo.toml +++ b/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"] } diff --git a/glommio/src/executor/mod.rs b/glommio/src/executor/mod.rs index 5cc2d67ee..52e64b209 100644 --- a/glommio/src/executor/mod.rs +++ b/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)); diff --git a/glommio/src/task/raw.rs b/glommio/src/task/raw.rs index a42a8e237..67759aa9e 100644 --- a/glommio/src/task/raw.rs +++ b/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. diff --git a/valgrind.supp b/valgrind.supp new file mode 100644 index 000000000..8078b0fd6 --- /dev/null +++ b/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* +}