From a6509b09896c617e26e45d0a81b43605d022e883 Mon Sep 17 00:00:00 2001 From: tison Date: Fri, 28 Aug 2026 22:13:19 +0800 Subject: [PATCH] test(mpsc): cover bounded sender wakeups --- tests-integration/tests/mpsc_test.rs | 72 ++++++++++++++++++++++++++++ 1 file changed, 72 insertions(+) diff --git a/tests-integration/tests/mpsc_test.rs b/tests-integration/tests/mpsc_test.rs index fab703ea..72256a7a 100644 --- a/tests-integration/tests/mpsc_test.rs +++ b/tests-integration/tests/mpsc_test.rs @@ -15,15 +15,24 @@ // specific language governing permissions and limitations // under the License. +use std::task::Poll; use std::time::Instant; use asyncband::mpsc; use asyncband::mpsc::RecvError; use asyncband::mpsc::TryRecvError; use asyncband::mpsc::TrySendError; +use tests_integration::poll_once; use tests_integration::test_runtime; use tokio_test::assert_ok; +fn expect_ready(poll: Poll) -> T { + match poll { + Poll::Ready(value) => value, + Poll::Pending => panic!("future should be ready"), + } +} + #[test] fn test_unbounded_pressure() { let n = 1024 * 1024; @@ -269,6 +278,69 @@ async fn send_after_close_bounded() { assert_eq!(error.into_inner(), 2); } +#[test] +fn bounded_wakes_blocked_senders_one_at_a_time() { + let (tx, mut rx) = mpsc::bounded(1); + tx.try_send(0).unwrap(); + + let first_tx = tx.clone(); + let second_tx = tx.clone(); + let mut first = Box::pin(first_tx.send(1)); + let mut second = Box::pin(second_tx.send(2)); + + assert!(poll_once(first.as_mut()).is_pending()); + assert!(poll_once(second.as_mut()).is_pending()); + + assert_eq!(rx.try_recv(), Ok(0)); + assert_eq!(expect_ready(poll_once(first.as_mut())), Ok(())); + assert!(poll_once(second.as_mut()).is_pending()); + + assert_eq!(rx.try_recv(), Ok(1)); + assert_eq!(expect_ready(poll_once(second.as_mut())), Ok(())); + assert_eq!(rx.try_recv(), Ok(2)); +} + +#[test] +fn bounded_cancelled_notified_sender_passes_slot_to_next_sender() { + let (tx, mut rx) = mpsc::bounded(1); + tx.try_send(0).unwrap(); + + let first_tx = tx.clone(); + let second_tx = tx.clone(); + let mut first = Box::pin(first_tx.send(1)); + let mut second = Box::pin(second_tx.send(2)); + + assert!(poll_once(first.as_mut()).is_pending()); + assert!(poll_once(second.as_mut()).is_pending()); + + assert_eq!(rx.try_recv(), Ok(0)); + drop(first); + + assert_eq!(expect_ready(poll_once(second.as_mut())), Ok(())); + assert_eq!(rx.try_recv(), Ok(2)); +} + +#[test] +fn bounded_receiver_drop_returns_values_to_all_blocked_senders() { + let (tx, rx) = mpsc::bounded(1); + tx.try_send(0).unwrap(); + + let first_tx = tx.clone(); + let second_tx = tx.clone(); + let mut first = Box::pin(first_tx.send(1)); + let mut second = Box::pin(second_tx.send(2)); + + assert!(poll_once(first.as_mut()).is_pending()); + assert!(poll_once(second.as_mut()).is_pending()); + + drop(rx); + + let first_error = expect_ready(poll_once(first.as_mut())).unwrap_err(); + let second_error = expect_ready(poll_once(second.as_mut())).unwrap_err(); + assert_eq!(first_error.into_inner(), 1); + assert_eq!(second_error.into_inner(), 2); +} + #[test] fn test_bounded_pressure() { let n = 1024 * 1024;