From ee0d6f05e23b4ba43c99046aac5d82090e21a644 Mon Sep 17 00:00:00 2001 From: Rhett Griggs Date: Thu, 30 Jul 2026 18:09:46 -0400 Subject: [PATCH 1/2] Release owned queue handles after grant completion Terminal grant methods forgot the entire grant to bypass its default Drop behavior. With ArcBBQueue, that retained an Arc after every commit or release and kept the queue allocation alive indefinitely. Release the owned handle while preserving terminal coordination, and cover each affected method with an Arc liveness regression test. --- bbqueue/src/lib.rs | 95 +++++++++++++++++++++++++++++---- bbqueue/src/prod_cons/framed.rs | 17 ++++-- bbqueue/src/prod_cons/stream.rs | 19 ++----- 3 files changed, 100 insertions(+), 31 deletions(-) diff --git a/bbqueue/src/lib.rs b/bbqueue/src/lib.rs index 74bfbf4..ffd51b0 100644 --- a/bbqueue/src/lib.rs +++ b/bbqueue/src/lib.rs @@ -143,13 +143,14 @@ pub mod export { #[cfg(all(test, feature = "alloc"))] mod test { + use alloc::sync::Arc; use core::{ops::Deref, time::Duration}; use crate::{ queue::{ArcBBQueue, BBQueue}, traits::{ coordination::cas::AtomicCoord, - notifier::maitake::MaiNotSpsc, + notifier::{maitake::MaiNotSpsc, polling::Polling}, storage::{BoxedSlice, Inline}, }, }; @@ -157,8 +158,6 @@ mod test { #[cfg(all(target_has_atomic = "ptr", feature = "alloc"))] #[test] fn ux() { - use crate::traits::{notifier::polling::Polling, storage::BoxedSlice}; - static BBQ: BBQueue, AtomicCoord, Polling> = BBQueue::new(); let _ = BBQ.stream_producer(); let _ = BBQ.stream_consumer(); @@ -177,9 +176,6 @@ mod test { #[cfg(target_has_atomic = "ptr")] #[test] fn smoke() { - use crate::traits::notifier::polling::Polling; - use core::ops::Deref; - static BBQ: BBQueue, AtomicCoord, Polling> = BBQueue::new(); let prod = BBQ.stream_producer(); let cons = BBQ.stream_consumer(); @@ -203,9 +199,6 @@ mod test { #[cfg(target_has_atomic = "ptr")] #[test] fn smoke_framed() { - use crate::traits::notifier::polling::Polling; - use core::ops::Deref; - static BBQ: BBQueue, AtomicCoord, Polling> = BBQueue::new(); let prod = BBQ.framed_producer(); let cons = BBQ.framed_consumer(); @@ -225,8 +218,6 @@ mod test { #[cfg(target_has_atomic = "ptr")] #[test] fn framed_misuse() { - use crate::traits::notifier::polling::Polling; - static BBQ: BBQueue, AtomicCoord, Polling> = BBQueue::new(); let prod = BBQ.stream_producer(); let cons = BBQ.framed_consumer(); @@ -255,6 +246,88 @@ mod test { assert!(cons.read().is_err()); } + #[test] + fn framed_commit_releases_owned_queue_handle() { + let bbq: ArcBBQueue, AtomicCoord, Polling> = + ArcBBQueue::new_with_storage(Inline::new()); + let weak = Arc::downgrade(&bbq.0); + let prod = bbq.framed_producer(); + let baseline = Arc::strong_count(&bbq.0); + + let wgr = prod.grant(1).expect("space should be available"); + assert_eq!(Arc::strong_count(&bbq.0), baseline + 1); + wgr.commit(1); + assert_eq!(Arc::strong_count(&bbq.0), baseline); + + drop(prod); + drop(bbq); + assert!(weak.upgrade().is_none()); + } + + #[test] + fn framed_release_releases_owned_queue_handle() { + let bbq: ArcBBQueue, AtomicCoord, Polling> = + ArcBBQueue::new_with_storage(Inline::new()); + let weak = Arc::downgrade(&bbq.0); + let seed_prod = bbq.0.as_ref().framed_producer(); + seed_prod + .grant(1) + .expect("space should be available") + .commit(1); + let cons = bbq.framed_consumer(); + let baseline = Arc::strong_count(&bbq.0); + + let rgr = cons.read().expect("seeded frame should be available"); + assert_eq!(Arc::strong_count(&bbq.0), baseline + 1); + rgr.release(); + assert_eq!(Arc::strong_count(&bbq.0), baseline); + + drop(cons); + drop(bbq); + assert!(weak.upgrade().is_none()); + } + + #[test] + fn stream_commit_releases_owned_queue_handle() { + let bbq: ArcBBQueue, AtomicCoord, Polling> = + ArcBBQueue::new_with_storage(Inline::new()); + let weak = Arc::downgrade(&bbq.0); + let prod = bbq.stream_producer(); + let baseline = Arc::strong_count(&bbq.0); + + let wgr = prod.grant_exact(1).expect("space should be available"); + assert_eq!(Arc::strong_count(&bbq.0), baseline + 1); + wgr.commit(1); + assert_eq!(Arc::strong_count(&bbq.0), baseline); + + drop(prod); + drop(bbq); + assert!(weak.upgrade().is_none()); + } + + #[test] + fn stream_release_releases_owned_queue_handle() { + let bbq: ArcBBQueue, AtomicCoord, Polling> = + ArcBBQueue::new_with_storage(Inline::new()); + let weak = Arc::downgrade(&bbq.0); + let seed_prod = bbq.0.as_ref().stream_producer(); + seed_prod + .grant_exact(1) + .expect("space should be available") + .commit(1); + let cons = bbq.stream_consumer(); + let baseline = Arc::strong_count(&bbq.0); + + let rgr = cons.read().expect("seeded byte should be available"); + assert_eq!(Arc::strong_count(&bbq.0), baseline + 1); + rgr.release(1); + assert_eq!(Arc::strong_count(&bbq.0), baseline); + + drop(cons); + drop(bbq); + assert!(weak.upgrade().is_none()); + } + #[tokio::test] async fn asink() { static BBQ: BBQueue, AtomicCoord, MaiNotSpsc> = BBQueue::new(); diff --git a/bbqueue/src/prod_cons/framed.rs b/bbqueue/src/prod_cons/framed.rs index 9418e16..4458d8e 100644 --- a/bbqueue/src/prod_cons/framed.rs +++ b/bbqueue/src/prod_cons/framed.rs @@ -4,8 +4,9 @@ use core::{ marker::PhantomData, + mem::ManuallyDrop, ops::{Deref, DerefMut}, - ptr::NonNull, + ptr::{self, NonNull}, }; use crate::traits::{ @@ -277,8 +278,11 @@ where } self.bbq.cor.commit_inner(cap, grant_len, used_len); - self.bbq.not.wake_one_consumer(); - core::mem::forget(self); + let this = ManuallyDrop::new(self); + // SAFETY: Read once; the ManuallyDrop grant is not accessed or dropped afterward. + let bbq = unsafe { ptr::read(&this.bbq) }; + bbq.not.wake_one_consumer(); + drop(bbq); } /// Aborts the grant, making no frame available to the consumer @@ -359,8 +363,11 @@ where let hdrlen: usize = const { core::mem::size_of::() }; let used = len + hdrlen; self.bbq.cor.release_inner(used); - self.bbq.not.wake_one_producer(); - core::mem::forget(self); + let this = ManuallyDrop::new(self); + // SAFETY: Read once; the ManuallyDrop grant is not accessed or dropped afterward. + let bbq = unsafe { ptr::read(&this.bbq) }; + bbq.not.wake_one_producer(); + drop(bbq); } /// Drop the grant WITHOUT releasing the message from the queue. diff --git a/bbqueue/src/prod_cons/stream.rs b/bbqueue/src/prod_cons/stream.rs index 543afbf..622911c 100644 --- a/bbqueue/src/prod_cons/stream.rs +++ b/bbqueue/src/prod_cons/stream.rs @@ -208,14 +208,8 @@ where /// Make `used` bytes available to be read. /// /// `used` is capped to the length of the grant. - pub fn commit(self, used: usize) { - let (_, cap) = unsafe { self.bbq.sto.ptr_len() }; - let used = used.min(self.len); - self.bbq.cor.commit_inner(cap, self.len, used); - if used != 0 { - self.bbq.not.wake_one_consumer(); - } - core::mem::forget(self); + pub fn commit(mut self, used: usize) { + self.to_commit = used; } } @@ -271,13 +265,8 @@ where /// Make `used` bytes available for writing. /// /// `used` is capped to the length of the grant - pub fn release(self, used: usize) { - let used = used.min(self.len); - self.bbq.cor.release_inner(used); - if used != 0 { - self.bbq.not.wake_one_producer(); - } - core::mem::forget(self); + pub fn release(mut self, used: usize) { + self.to_release = used; } } From fbbb3ef3a125e0b99cc794acd300d2edb34ff550 Mon Sep 17 00:00:00 2001 From: Rhett Griggs Date: Thu, 30 Jul 2026 18:31:31 -0400 Subject: [PATCH 2/2] Enable Miri leak checking The existing Arc-backed tests expose owned handle leaks, but the Miri script disabled leak reports. Keep the checker enabled now that the suite is leak-free so CI catches future regressions. --- bbqueue/miri.sh | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/bbqueue/miri.sh b/bbqueue/miri.sh index 5d13344..5a83dab 100755 --- a/bbqueue/miri.sh +++ b/bbqueue/miri.sh @@ -1,10 +1,7 @@ #!/bin/bash # We disable isolation because we use tokio sleep -# We disable leaks because ??? -# -# TODO: Can we eliminate some of these limitations for testing? -MIRIFLAGS="-Zmiri-disable-isolation -Zmiri-ignore-leaks" \ +MIRIFLAGS="-Zmiri-disable-isolation" \ cargo +nightly miri test \ --target x86_64-unknown-linux-gnu \ --features=std