From d5bfbfba3a15bc891bd964cb12c04c99ebc3c6a2 Mon Sep 17 00:00:00 2001 From: Rhett Griggs Date: Fri, 31 Jul 2026 01:10:00 -0400 Subject: [PATCH 1/2] Check queue readiness before waiting --- bbqueue/src/prod_cons/framed.rs | 6 +++ bbqueue/src/prod_cons/mod.rs | 65 +++++++++++++++++++++++++++++++++ bbqueue/src/prod_cons/stream.rs | 9 +++++ 3 files changed, 80 insertions(+) diff --git a/bbqueue/src/prod_cons/framed.rs b/bbqueue/src/prod_cons/framed.rs index 9418e16..d800891 100644 --- a/bbqueue/src/prod_cons/framed.rs +++ b/bbqueue/src/prod_cons/framed.rs @@ -169,6 +169,9 @@ where /// a smaller size may be committed. Dropping the grant without calling /// commit means that no data will be made visible to the consumer. pub async fn wait_grant(&self, sz: H) -> FramedGrantW { + if let Ok(grant) = self.grant(sz) { + return grant; + } self.bbq.not.wait_for_not_full(|| self.grant(sz).ok()).await } } @@ -247,6 +250,9 @@ where /// /// The returned grant must be released to free the space in the buffer. pub async fn wait_read(&self) -> FramedGrantR { + if let Ok(grant) = self.read() { + return grant; + } self.bbq.not.wait_for_not_empty(|| self.read().ok()).await } } diff --git a/bbqueue/src/prod_cons/mod.rs b/bbqueue/src/prod_cons/mod.rs index 9e9b34e..40fa11b 100644 --- a/bbqueue/src/prod_cons/mod.rs +++ b/bbqueue/src/prod_cons/mod.rs @@ -18,3 +18,68 @@ pub mod framed; pub mod stream; + +#[cfg(test)] +mod tests { + use core::sync::atomic::{AtomicUsize, Ordering}; + + use const_init::ConstInit; + + use crate::{ + BBQueue, + traits::{ + coordination::cas::AtomicCoord, + notifier::{AsyncNotifier, Notifier}, + storage::Inline, + }, + }; + + static WAIT_CALLS: AtomicUsize = AtomicUsize::new(0); + + struct CountingNotifier; + + impl ConstInit for CountingNotifier { + const INIT: Self = Self; + } + + impl Notifier for CountingNotifier { + fn wake_one_consumer(&self) {} + + fn wake_one_producer(&self) {} + } + + impl AsyncNotifier for CountingNotifier { + async fn wait_for_not_empty Option>(&self, mut f: F) -> T { + WAIT_CALLS.fetch_add(1, Ordering::Relaxed); + f().expect("test queue has readable data") + } + + async fn wait_for_not_full Option>(&self, mut f: F) -> T { + WAIT_CALLS.fetch_add(1, Ordering::Relaxed); + f().expect("test queue has writable space") + } + } + + #[tokio::test] + async fn ready_waits_do_not_arm_notifier() { + WAIT_CALLS.store(0, Ordering::Relaxed); + + let stream: BBQueue, AtomicCoord, CountingNotifier> = BBQueue::new(); + let stream_producer = stream.stream_producer(); + let stream_consumer = stream.stream_consumer(); + + stream_producer.wait_grant_exact(1).await.commit(1); + stream_consumer.wait_read().await.release(1); + stream_producer.wait_grant_max_remaining(1).await.commit(1); + stream_consumer.wait_read().await.release(1); + + let framed: BBQueue, AtomicCoord, CountingNotifier> = BBQueue::new(); + let framed_producer = framed.framed_producer(); + let framed_consumer = framed.framed_consumer(); + + framed_producer.wait_grant(1).await.commit(1); + framed_consumer.wait_read().await.release(); + + assert_eq!(WAIT_CALLS.load(Ordering::Relaxed), 0); + } +} diff --git a/bbqueue/src/prod_cons/stream.rs b/bbqueue/src/prod_cons/stream.rs index 543afbf..bff8cb4 100644 --- a/bbqueue/src/prod_cons/stream.rs +++ b/bbqueue/src/prod_cons/stream.rs @@ -134,6 +134,9 @@ where { /// Wait for a grant of any size, up to `max`, to become available pub async fn wait_grant_max_remaining(&self, max: usize) -> StreamGrantW { + if let Ok(grant) = self.grant_max_remaining(max) { + return grant; + } self.bbq .not .wait_for_not_full(|| self.grant_max_remaining(max).ok()) @@ -144,6 +147,9 @@ where /// /// If `sz` exceeds the capacity of the buffer, this method will never return. pub async fn wait_grant_exact(&self, sz: usize) -> StreamGrantW { + if let Ok(grant) = self.grant_exact(sz) { + return grant; + } self.bbq .not .wait_for_not_full(|| self.grant_exact(sz).ok()) @@ -193,6 +199,9 @@ where { /// Wait for any read data to become available pub async fn wait_read(&self) -> StreamGrantR { + if let Ok(grant) = self.read() { + return grant; + } self.bbq.not.wait_for_not_empty(|| self.read().ok()).await } } From 7d6e7738e19a8aea81a2dd7045de1cd342681dd2 Mon Sep 17 00:00:00 2001 From: Rhett Griggs Date: Fri, 31 Jul 2026 14:07:50 -0400 Subject: [PATCH 2/2] Report an unsatisfiable wait grant instead of waiting forever `FramedProducer::wait_grant` and `StreamProducer::wait_grant_exact` park until a request can be met, but a request larger than the buffer itself cannot be met by any amount of consumer progress. Both documented that as "will never return" and left the caller hung on a size the queue could have rejected outright. The synchronous `grant` and `grant_exact` already answer this exact input with `WriteGrantError::InsufficientSize`, so the waiting variants now return the same `Result` rather than growing a second convention for the same condition. The capacity comparison happens once, before the wait: capacity is fixed for the lifetime of the queue, so a request that does not fit an empty buffer will never fit a fuller one. `wait_grant_max_remaining` is unaffected; it caps its ask at what is available. --- bbqueue/src/prod_cons/framed.rs | 62 +++++++++++++++++++++++++++++---- bbqueue/src/prod_cons/mod.rs | 12 +++++-- bbqueue/src/prod_cons/stream.rs | 56 ++++++++++++++++++++++++++--- 3 files changed, 117 insertions(+), 13 deletions(-) diff --git a/bbqueue/src/prod_cons/framed.rs b/bbqueue/src/prod_cons/framed.rs index d800891..04f8d0f 100644 --- a/bbqueue/src/prod_cons/framed.rs +++ b/bbqueue/src/prod_cons/framed.rs @@ -162,17 +162,28 @@ where { /// Wait for the given write grant to become available /// - /// If `sz` is larger than the storage buffer, this method will never - /// return. - /// /// The returned grant can be used to write up to `sz` bytes, though /// a smaller size may be committed. Dropping the grant without calling /// commit means that no data will be made visible to the consumer. - pub async fn wait_grant(&self, sz: H) -> FramedGrantW { + /// + /// # Errors + /// + /// Returns [`WriteGrantError::InsufficientSize`] when a frame of `sz` cannot + /// fit the buffer even when empty. Releasing every outstanding read grant + /// would not satisfy such a request, so it is reported rather than awaited. + pub async fn wait_grant(&self, sz: H) -> Result, WriteGrantError> { + let needed = sz + .into() + .checked_add(core::mem::size_of::()) + .ok_or(WriteGrantError::InsufficientSize)?; + if needed > self.capacity() { + return Err(WriteGrantError::InsufficientSize); + } + if let Ok(grant) = self.grant(sz) { - return grant; + return Ok(grant); } - self.bbq.not.wait_for_not_full(|| self.grant(sz).ok()).await + Ok(self.bbq.not.wait_for_not_full(|| self.grant(sz).ok()).await) } } @@ -419,3 +430,42 @@ where H: LenHeader + Send, { } + +#[cfg(all(test, feature = "maitake-sync-0_3"))] +mod async_tests { + use core::{ + future::Future, + pin::pin, + task::{Context, Poll, Waker}, + }; + + use crate::{ + BBQueue, + prod_cons::framed::FramedProducer, + traits::{ + bbqhdl::BbqHandle, + coordination::{WriteGrantError, cas::AtomicCoord}, + notifier::maitake::MaiNotSpsc, + storage::Inline, + }, + }; + + type AsyncRing = BBQueue, AtomicCoord, MaiNotSpsc>; + + /// A frame too large for an empty buffer can never be granted, so the wait + /// resolves instead of parking on a consumer that cannot help. + #[test] + fn wait_grant_rejects_a_frame_larger_than_the_buffer() { + let bbq: AsyncRing = BBQueue::new(); + let prod: FramedProducer<&AsyncRing, usize> = BbqHandle::framed_producer::(&&bbq); + + // The payload alone fits; the header pushes it past capacity. + let sz = prod.capacity() - 1; + + let mut wait = pin!(prod.wait_grant(sz)); + let Poll::Ready(res) = wait.as_mut().poll(&mut Context::from_waker(Waker::noop())) else { + panic!("wait_grant parked on a request no consumer can satisfy"); + }; + assert_eq!(res.err(), Some(WriteGrantError::InsufficientSize)); + } +} diff --git a/bbqueue/src/prod_cons/mod.rs b/bbqueue/src/prod_cons/mod.rs index 40fa11b..6e5465e 100644 --- a/bbqueue/src/prod_cons/mod.rs +++ b/bbqueue/src/prod_cons/mod.rs @@ -68,7 +68,11 @@ mod tests { let stream_producer = stream.stream_producer(); let stream_consumer = stream.stream_consumer(); - stream_producer.wait_grant_exact(1).await.commit(1); + stream_producer + .wait_grant_exact(1) + .await + .expect("1 byte fits a 64 byte buffer") + .commit(1); stream_consumer.wait_read().await.release(1); stream_producer.wait_grant_max_remaining(1).await.commit(1); stream_consumer.wait_read().await.release(1); @@ -77,7 +81,11 @@ mod tests { let framed_producer = framed.framed_producer(); let framed_consumer = framed.framed_consumer(); - framed_producer.wait_grant(1).await.commit(1); + framed_producer + .wait_grant(1) + .await + .expect("a 1 byte frame fits a 64 byte buffer") + .commit(1); framed_consumer.wait_read().await.release(); assert_eq!(WAIT_CALLS.load(Ordering::Relaxed), 0); diff --git a/bbqueue/src/prod_cons/stream.rs b/bbqueue/src/prod_cons/stream.rs index bff8cb4..f66901c 100644 --- a/bbqueue/src/prod_cons/stream.rs +++ b/bbqueue/src/prod_cons/stream.rs @@ -145,15 +145,24 @@ where /// Wait for a grant of EXACTLY `sz` to become available. /// - /// If `sz` exceeds the capacity of the buffer, this method will never return. - pub async fn wait_grant_exact(&self, sz: usize) -> StreamGrantW { + /// # Errors + /// + /// Returns [`WriteGrantError::InsufficientSize`] when `sz` cannot fit the + /// buffer even when empty. Releasing every outstanding read grant would not + /// satisfy such a request, so it is reported rather than awaited. + pub async fn wait_grant_exact(&self, sz: usize) -> Result, WriteGrantError> { + if sz > self.capacity() { + return Err(WriteGrantError::InsufficientSize); + } + if let Ok(grant) = self.grant_exact(sz) { - return grant; + return Ok(grant); } - self.bbq + Ok(self + .bbq .not .wait_for_not_full(|| self.grant_exact(sz).ok()) - .await + .await) } } @@ -331,3 +340,40 @@ where } unsafe impl Send for StreamGrantR {} + +#[cfg(all(test, feature = "maitake-sync-0_3"))] +mod async_tests { + use core::{ + future::Future, + pin::pin, + task::{Context, Poll, Waker}, + }; + + use crate::{ + BBQueue, + prod_cons::stream::StreamProducer, + traits::{ + coordination::{WriteGrantError, cas::AtomicCoord}, + notifier::maitake::MaiNotSpsc, + storage::Inline, + }, + }; + + type AsyncRing = BBQueue, AtomicCoord, MaiNotSpsc>; + + /// A request larger than the whole buffer can never be granted, so the wait + /// resolves instead of parking on a consumer that cannot help. + #[test] + fn wait_grant_exact_rejects_a_size_larger_than_the_buffer() { + let bbq: AsyncRing = BBQueue::new(); + let prod: StreamProducer<&AsyncRing> = bbq.stream_producer(); + + let sz = prod.capacity() + 1; + + let mut wait = pin!(prod.wait_grant_exact(sz)); + let Poll::Ready(res) = wait.as_mut().poll(&mut Context::from_waker(Waker::noop())) else { + panic!("wait_grant_exact parked on a request no consumer can satisfy"); + }; + assert_eq!(res.err(), Some(WriteGrantError::InsufficientSize)); + } +}