mirror of
https://github.com/tokio-rs/tokio.git
synced 2026-08-21 00:00:10 +02:00
sync: clarify broadcast lagging semantics and pin them with tests (#8239)
Expand module and API docs around RecvError::Lagged / TryRecvError::Lagged so capacity rounding, miss counts, and post-lag resume behavior are explicit. Add integration tests covering slow receivers within capacity, overflow, per-receiver lag, async recv, and cursor advancement after Lagged.
This commit is contained in:
yoda77777
+79
-39
@@ -32,15 +32,26 @@
|
||||
//! This broadcast channel implementation handles this case by setting a hard
|
||||
//! upper bound on the number of values the channel may retain at any given
|
||||
//! time. This upper bound is passed to the [`channel`] function as an argument.
|
||||
//! The provided capacity is rounded **up** to the next power of two; that
|
||||
//! rounded size is the number of messages the ring buffer can hold, and is what
|
||||
//! lag detection is based on. For example, `channel(3)` allocates a buffer of
|
||||
//! length 4, so a receiver only lags once it falls more than 4 messages behind
|
||||
//! the sender.
|
||||
//!
|
||||
//! If a value is sent when the channel is at capacity, the oldest value
|
||||
//! currently held by the channel is released. This frees up space for the new
|
||||
//! value. Any receiver that has not yet seen the released value will return
|
||||
//! [`RecvError::Lagged`] the next time [`recv`] is called.
|
||||
//! currently held by the channel is overwritten. This frees up space for the
|
||||
//! new value. Any receiver that has not yet seen the overwritten value will
|
||||
//! return [`RecvError::Lagged`] the next time [`recv`] (or
|
||||
//! [`try_recv`](Receiver::try_recv)) is called. The error carries the number of
|
||||
//! messages that were dropped before the receiver's cursor and are therefore
|
||||
//! no longer available.
|
||||
//!
|
||||
//! Once [`RecvError::Lagged`] is returned, the lagging receiver's position is
|
||||
//! updated to the oldest value contained by the channel. The next call to
|
||||
//! [`recv`] will return this value.
|
||||
//! Returning [`RecvError::Lagged`] does **not** close or disconnect the
|
||||
//! receiver. The lagging receiver's internal cursor is advanced to the oldest
|
||||
//! value still retained by the channel. The **next** successful call to
|
||||
//! [`recv`] / [`try_recv`](Receiver::try_recv) returns that oldest retained
|
||||
//! value (unless further sends overwrite it again before the receiver reads
|
||||
//! it). Subsequent receives then continue in send order from there.
|
||||
//!
|
||||
//! This behavior enables a receiver to detect when it has lagged so far behind
|
||||
//! that data has been dropped. The caller may decide how to respond to this:
|
||||
@@ -97,20 +108,23 @@
|
||||
//!
|
||||
//! ```
|
||||
//! use tokio::sync::broadcast;
|
||||
//! use tokio::sync::broadcast::error::RecvError;
|
||||
//!
|
||||
//! # #[tokio::main(flavor = "current_thread")]
|
||||
//! # async fn main() {
|
||||
//! // Capacity 2 → ring buffer of length 2.
|
||||
//! let (tx, mut rx) = broadcast::channel(2);
|
||||
//!
|
||||
//! tx.send(10).unwrap();
|
||||
//! tx.send(20).unwrap();
|
||||
//! // Overwrites 10; receiver has not read it yet.
|
||||
//! tx.send(30).unwrap();
|
||||
//!
|
||||
//! // The receiver lagged behind
|
||||
//! assert!(rx.recv().await.is_err());
|
||||
//!
|
||||
//! // At this point, we can abort or continue with lost messages
|
||||
//! // One message (10) was dropped; cursor moves to the oldest retained value (20).
|
||||
//! assert!(matches!(rx.recv().await, Err(RecvError::Lagged(1))));
|
||||
//!
|
||||
//! // At this point, we can abort or continue with lost messages.
|
||||
//! // Continuing resumes from the oldest retained message.
|
||||
//! assert_eq!(20, rx.recv().await.unwrap());
|
||||
//! assert_eq!(30, rx.recv().await.unwrap());
|
||||
//! # }
|
||||
@@ -278,10 +292,18 @@ pub mod error {
|
||||
/// be sent.
|
||||
Closed,
|
||||
|
||||
/// The receiver lagged too far behind. Attempting to receive again will
|
||||
/// return the oldest message still retained by the channel.
|
||||
/// The receiver lagged too far behind: one or more messages were
|
||||
/// overwritten in the ring buffer before this receiver could read them.
|
||||
///
|
||||
/// Includes the number of skipped messages.
|
||||
/// The receiver remains subscribed. Its internal cursor has been advanced
|
||||
/// to the oldest message still retained by the channel; the next
|
||||
/// successful [`recv`] call returns that message (unless further sends
|
||||
/// overwrite it first).
|
||||
///
|
||||
/// The `u64` is the number of messages that were skipped (dropped before
|
||||
/// the receiver's previous cursor position).
|
||||
///
|
||||
/// [`recv`]: crate::sync::broadcast::Receiver::recv
|
||||
Lagged(u64),
|
||||
}
|
||||
|
||||
@@ -312,11 +334,18 @@ pub mod error {
|
||||
/// be sent.
|
||||
Closed,
|
||||
|
||||
/// The receiver lagged too far behind and has been forcibly disconnected.
|
||||
/// Attempting to receive again will return the oldest message still
|
||||
/// retained by the channel.
|
||||
/// The receiver lagged too far behind: one or more messages were
|
||||
/// overwritten in the ring buffer before this receiver could read them.
|
||||
///
|
||||
/// Includes the number of skipped messages.
|
||||
/// The receiver remains subscribed. Its internal cursor has been advanced
|
||||
/// to the oldest message still retained by the channel; the next
|
||||
/// successful [`try_recv`] call returns that message (unless further sends
|
||||
/// overwrite it first).
|
||||
///
|
||||
/// The `u64` is the number of messages that were skipped (dropped before
|
||||
/// the receiver's previous cursor position).
|
||||
///
|
||||
/// [`try_recv`]: crate::sync::broadcast::Receiver::try_recv
|
||||
Lagged(u64),
|
||||
}
|
||||
|
||||
@@ -453,7 +482,10 @@ const MAX_RECEIVERS: usize = usize::MAX >> 2;
|
||||
/// Create a bounded, multi-producer, multi-consumer channel where each sent
|
||||
/// value is broadcasted to all active receivers.
|
||||
///
|
||||
/// **Note:** The actual capacity may be greater than the provided `capacity`.
|
||||
/// **Note:** The provided `capacity` is rounded **up** to the next power of
|
||||
/// two. That rounded size is the number of messages the internal ring buffer
|
||||
/// can retain, and is what [lag detection](self#lagging) uses. For example,
|
||||
/// `channel(3)` behaves as if the capacity were 4.
|
||||
///
|
||||
/// All data sent on [`Sender`] will become available on every active
|
||||
/// [`Receiver`] in the same order as it was sent.
|
||||
@@ -1130,12 +1162,16 @@ impl<T> Receiver<T> {
|
||||
/// Returns the number of messages that were sent into the channel and that
|
||||
/// this [`Receiver`] has yet to receive.
|
||||
///
|
||||
/// If the returned value from `len` is larger than the next largest power of 2
|
||||
/// of the capacity of the channel any call to [`recv`] will return an
|
||||
/// `Err(RecvError::Lagged)` and any call to [`try_recv`] will return an
|
||||
/// `Err(TryRecvError::Lagged)`, e.g. if the capacity of the channel is 10,
|
||||
/// [`recv`] will start to return `Err(RecvError::Lagged)` once `len` returns
|
||||
/// values larger than 16.
|
||||
/// This count includes messages that have already been overwritten in the
|
||||
/// ring buffer and are no longer readable. If `len` is **greater than** the
|
||||
/// channel's effective capacity (the provided capacity rounded up to the
|
||||
/// next power of two), the next call to [`recv`] returns
|
||||
/// `Err(RecvError::Lagged)` and the next call to [`try_recv`] returns
|
||||
/// `Err(TryRecvError::Lagged)`. For example, with `channel(10)` the buffer
|
||||
/// length is 16, so lagging begins once `len` is larger than 16.
|
||||
///
|
||||
/// After a successful receive (including after handling `Lagged` and then
|
||||
/// reading retained messages), `len` decreases accordingly.
|
||||
///
|
||||
/// [`Receiver`]: crate::sync::broadcast::Receiver
|
||||
/// [`recv`]: crate::sync::broadcast::Receiver::recv
|
||||
@@ -1399,11 +1435,13 @@ impl<T: Clone> Receiver<T> {
|
||||
/// dropped, indicating that no further values can be sent on the channel.
|
||||
///
|
||||
/// If the [`Receiver`] handle falls behind, once the channel is full, newly
|
||||
/// sent values will overwrite old values. At this point, a call to [`recv`]
|
||||
/// will return with `Err(RecvError::Lagged)` and the [`Receiver`]'s
|
||||
/// internal cursor is updated to point to the oldest value still held by
|
||||
/// the channel. A subsequent call to [`recv`] will return this value
|
||||
/// **unless** it has been since overwritten.
|
||||
/// sent values overwrite old values in the ring buffer. The next call to
|
||||
/// [`recv`] then returns `Err(RecvError::Lagged(n))`, where `n` is the
|
||||
/// number of overwritten messages the receiver missed. The receiver stays
|
||||
/// subscribed; its internal cursor is advanced to the oldest value still
|
||||
/// held by the channel. A subsequent call to [`recv`] returns that value,
|
||||
/// unless further sends overwrite it before the receiver reads it. See
|
||||
/// [lagging](self#lagging) for details.
|
||||
///
|
||||
/// # Cancel safety
|
||||
///
|
||||
@@ -1444,6 +1482,7 @@ impl<T: Clone> Receiver<T> {
|
||||
///
|
||||
/// ```
|
||||
/// use tokio::sync::broadcast;
|
||||
/// use tokio::sync::broadcast::error::RecvError;
|
||||
///
|
||||
/// # #[tokio::main(flavor = "current_thread")]
|
||||
/// # async fn main() {
|
||||
@@ -1453,11 +1492,10 @@ impl<T: Clone> Receiver<T> {
|
||||
/// tx.send(20).unwrap();
|
||||
/// tx.send(30).unwrap();
|
||||
///
|
||||
/// // The receiver lagged behind
|
||||
/// assert!(rx.recv().await.is_err());
|
||||
///
|
||||
/// // At this point, we can abort or continue with lost messages
|
||||
/// // One message was overwritten before this receiver could read it.
|
||||
/// assert!(matches!(rx.recv().await, Err(RecvError::Lagged(1))));
|
||||
///
|
||||
/// // Resume from the oldest retained message, or abort the task instead.
|
||||
/// assert_eq!(20, rx.recv().await.unwrap());
|
||||
/// assert_eq!(30, rx.recv().await.unwrap());
|
||||
/// # }
|
||||
@@ -1478,12 +1516,14 @@ impl<T: Clone> Receiver<T> {
|
||||
/// dropped, indicating that no further values can be sent on the channel.
|
||||
///
|
||||
/// If the [`Receiver`] handle falls behind, once the channel is full, newly
|
||||
/// sent values will overwrite old values. At this point, a call to [`recv`]
|
||||
/// will return with `Err(TryRecvError::Lagged)` and the [`Receiver`]'s
|
||||
/// internal cursor is updated to point to the oldest value still held by
|
||||
/// the channel. A subsequent call to [`try_recv`] will return this value
|
||||
/// **unless** it has been since overwritten. If there are no values to
|
||||
/// receive, `Err(TryRecvError::Empty)` is returned.
|
||||
/// sent values overwrite old values in the ring buffer. The next call to
|
||||
/// [`try_recv`] then returns `Err(TryRecvError::Lagged(n))`, where `n` is
|
||||
/// the number of overwritten messages the receiver missed. The receiver
|
||||
/// stays subscribed; its internal cursor is advanced to the oldest value
|
||||
/// still held by the channel. A subsequent call to [`try_recv`] returns
|
||||
/// that value, unless further sends overwrite it before the receiver reads
|
||||
/// it. If there are no values to receive, `Err(TryRecvError::Empty)` is
|
||||
/// returned. See [lagging](self#lagging) for details.
|
||||
///
|
||||
/// [`recv`]: crate::sync::broadcast::Receiver::recv
|
||||
/// [`try_recv`]: crate::sync::broadcast::Receiver::try_recv
|
||||
|
||||
@@ -720,3 +720,240 @@ async fn broadcast_sender_new_must_be_closed() {
|
||||
let mut task2 = task::spawn(tx.closed());
|
||||
assert_pending!(task2.poll());
|
||||
}
|
||||
|
||||
// --- Lagging semantics -------------------------------------------------------
|
||||
//
|
||||
// Capacity is rounded up to the next power of two. That buffer length is the
|
||||
// maximum number of messages retained; exceeding it overwrites the oldest
|
||||
// message and causes RecvError::Lagged / TryRecvError::Lagged on the slow
|
||||
// receiver. After Lagged, the cursor points at the oldest retained message.
|
||||
|
||||
/// A slow receiver that stays within the ring buffer receives every message
|
||||
/// in order; lagging only begins once the buffer wraps past the receiver.
|
||||
#[test]
|
||||
fn slow_receiver_within_capacity_does_not_lag() {
|
||||
// capacity 2 → buffer length 2
|
||||
let (tx, mut slow) = broadcast::channel(2);
|
||||
let mut fast = tx.subscribe();
|
||||
|
||||
assert_ok!(tx.send(1));
|
||||
assert_ok!(tx.send(2));
|
||||
|
||||
// Fast keeps up; slow has not read yet but both messages are still retained.
|
||||
assert_eq!(assert_recv!(fast), 1);
|
||||
assert_eq!(assert_recv!(fast), 2);
|
||||
assert_empty!(fast);
|
||||
|
||||
assert_eq!(assert_recv!(slow), 1);
|
||||
assert_eq!(assert_recv!(slow), 2);
|
||||
assert_empty!(slow);
|
||||
}
|
||||
|
||||
/// When sends overwrite the oldest retained values, the slow receiver sees
|
||||
/// Lagged(n) with the correct miss count, then resumes from the oldest
|
||||
/// retained message in send order.
|
||||
#[test]
|
||||
fn capacity_overflow_reports_lagged_then_oldest_retained() {
|
||||
// capacity 2 → buffer length 2; third send overwrites the first.
|
||||
let (tx, mut rx) = broadcast::channel(2);
|
||||
|
||||
assert_ok!(tx.send(10));
|
||||
assert_ok!(tx.send(20));
|
||||
assert_ok!(tx.send(30));
|
||||
|
||||
assert_lagged!(rx.try_recv(), 1);
|
||||
// After Lagged, cursor is at oldest retained (20).
|
||||
assert_eq!(assert_recv!(rx), 20);
|
||||
assert_eq!(assert_recv!(rx), 30);
|
||||
assert_empty!(rx);
|
||||
}
|
||||
|
||||
/// Lagged(n) counts every overwritten message since the receiver's previous
|
||||
/// cursor, not merely "at least one".
|
||||
#[test]
|
||||
fn lagged_count_matches_number_of_overwritten_messages() {
|
||||
// capacity 4 → buffer length 4
|
||||
let (tx, mut rx) = broadcast::channel(4);
|
||||
|
||||
for i in 1..=4 {
|
||||
assert_ok!(tx.send(i));
|
||||
}
|
||||
// Still within capacity: no lag.
|
||||
assert_eq!(assert_recv!(rx), 1);
|
||||
|
||||
// Send four more without reading: overwrites 2, 3, 4, and then the slot
|
||||
// that held 1 (already consumed). Receiver still owed 2,3,4 — all gone —
|
||||
// so it missed 3 messages; oldest retained is 5.
|
||||
for i in 5..=8 {
|
||||
assert_ok!(tx.send(i));
|
||||
}
|
||||
|
||||
assert_lagged!(rx.try_recv(), 3);
|
||||
assert_eq!(assert_recv!(rx), 5);
|
||||
assert_eq!(assert_recv!(rx), 6);
|
||||
assert_eq!(assert_recv!(rx), 7);
|
||||
assert_eq!(assert_recv!(rx), 8);
|
||||
assert_empty!(rx);
|
||||
}
|
||||
|
||||
/// Non-power-of-two capacity is rounded up; lag detection uses the rounded
|
||||
/// buffer length (e.g. capacity 3 → buffer 4).
|
||||
#[test]
|
||||
fn lag_uses_power_of_two_buffer_length() {
|
||||
// capacity 3 → buffer length 4
|
||||
let (tx, mut rx) = broadcast::channel(3);
|
||||
|
||||
for i in 1..=4 {
|
||||
assert_ok!(tx.send(i));
|
||||
}
|
||||
// Four messages fit in the rounded buffer; no lag yet.
|
||||
assert_eq!(rx.len(), 4);
|
||||
assert_eq!(assert_recv!(rx), 1);
|
||||
|
||||
// After reading 1, next points at 2. Buffer holds 2,3,4,5 — still no lag.
|
||||
assert_ok!(tx.send(5));
|
||||
assert_eq!(rx.len(), 4);
|
||||
|
||||
// Overwrites 2; receiver still wants 2 → Lagged(1), resume at 3.
|
||||
assert_ok!(tx.send(6));
|
||||
assert_eq!(rx.len(), 5);
|
||||
assert_lagged!(rx.try_recv(), 1);
|
||||
assert_eq!(assert_recv!(rx), 3);
|
||||
assert_eq!(assert_recv!(rx), 4);
|
||||
assert_eq!(assert_recv!(rx), 5);
|
||||
assert_eq!(assert_recv!(rx), 6);
|
||||
assert_empty!(rx);
|
||||
}
|
||||
|
||||
/// Async `recv` reports the same Lagged semantics as `try_recv`.
|
||||
#[tokio::test]
|
||||
async fn async_recv_lagged_then_resumes_from_oldest_retained() {
|
||||
use broadcast::error::RecvError;
|
||||
|
||||
let (tx, mut rx) = broadcast::channel(2);
|
||||
|
||||
assert_ok!(tx.send(10));
|
||||
assert_ok!(tx.send(20));
|
||||
assert_ok!(tx.send(30));
|
||||
|
||||
assert!(matches!(rx.recv().await, Err(RecvError::Lagged(1))));
|
||||
assert_eq!(rx.recv().await.unwrap(), 20);
|
||||
assert_eq!(rx.recv().await.unwrap(), 30);
|
||||
}
|
||||
|
||||
/// A receiver that lags, catches up, then lags again reports a fresh miss
|
||||
/// count based on the new gap only.
|
||||
#[test]
|
||||
fn lag_catch_up_then_lag_again() {
|
||||
let (tx, mut rx) = broadcast::channel(2);
|
||||
|
||||
assert_ok!(tx.send(1));
|
||||
assert_ok!(tx.send(2));
|
||||
assert_ok!(tx.send(3));
|
||||
|
||||
assert_lagged!(rx.try_recv(), 1);
|
||||
assert_eq!(assert_recv!(rx), 2);
|
||||
assert_eq!(assert_recv!(rx), 3);
|
||||
assert_empty!(rx);
|
||||
|
||||
// Fully caught up; another overflow lags again from a clean cursor.
|
||||
assert_ok!(tx.send(4));
|
||||
assert_ok!(tx.send(5));
|
||||
assert_ok!(tx.send(6));
|
||||
|
||||
assert_lagged!(rx.try_recv(), 1);
|
||||
assert_eq!(assert_recv!(rx), 5);
|
||||
assert_eq!(assert_recv!(rx), 6);
|
||||
assert_empty!(rx);
|
||||
}
|
||||
|
||||
/// Only the slow receiver lags; a caught-up receiver is unaffected.
|
||||
#[test]
|
||||
fn lag_is_per_receiver() {
|
||||
let (tx, mut slow) = broadcast::channel(2);
|
||||
let mut fast = tx.subscribe();
|
||||
|
||||
assert_ok!(tx.send(1));
|
||||
assert_ok!(tx.send(2));
|
||||
|
||||
assert_eq!(assert_recv!(fast), 1);
|
||||
assert_eq!(assert_recv!(fast), 2);
|
||||
|
||||
assert_ok!(tx.send(3));
|
||||
assert_ok!(tx.send(4));
|
||||
// Fast has read through 2; buffer holds 3,4 — no lag.
|
||||
assert_eq!(assert_recv!(fast), 3);
|
||||
assert_eq!(assert_recv!(fast), 4);
|
||||
|
||||
// Slow never read; 1 and 2 were overwritten → Lagged(2), oldest is 3.
|
||||
assert_lagged!(slow.try_recv(), 2);
|
||||
assert_eq!(assert_recv!(slow), 3);
|
||||
assert_eq!(assert_recv!(slow), 4);
|
||||
assert_empty!(slow);
|
||||
assert_empty!(fast);
|
||||
}
|
||||
|
||||
/// If the receiver lags and more messages are sent before it reads the
|
||||
/// Lagged error's "resume" position, a second Lagged reflects the additional
|
||||
/// overwrites since the cursor was advanced.
|
||||
#[test]
|
||||
fn lag_again_before_reading_retained_messages() {
|
||||
let (tx, mut rx) = broadcast::channel(2);
|
||||
|
||||
assert_ok!(tx.send(1));
|
||||
assert_ok!(tx.send(2));
|
||||
assert_ok!(tx.send(3));
|
||||
|
||||
// First lag: miss 1, cursor advances to oldest retained (value 2).
|
||||
assert_lagged!(rx.try_recv(), 1);
|
||||
|
||||
// Before reading 2/3, send enough to overwrite them (and one more).
|
||||
assert_ok!(tx.send(4));
|
||||
assert_ok!(tx.send(5));
|
||||
assert_ok!(tx.send(6));
|
||||
|
||||
// Cursor was at value 2; values 2, 3, and 4 are gone; oldest retained is 5.
|
||||
assert_lagged!(rx.try_recv(), 3);
|
||||
assert_eq!(assert_recv!(rx), 5);
|
||||
assert_eq!(assert_recv!(rx), 6);
|
||||
assert_empty!(rx);
|
||||
}
|
||||
|
||||
/// With capacity 1 (buffer length 1), every send after the first unread one
|
||||
/// causes a lag of exactly one when the receiver finally reads.
|
||||
#[test]
|
||||
fn single_slot_capacity_lag_semantics() {
|
||||
let (tx, mut rx) = broadcast::channel(1);
|
||||
|
||||
assert_ok!(tx.send(1));
|
||||
assert_eq!(assert_recv!(rx), 1);
|
||||
|
||||
assert_ok!(tx.send(2));
|
||||
assert_ok!(tx.send(3));
|
||||
|
||||
assert_lagged!(rx.try_recv(), 1);
|
||||
assert_eq!(assert_recv!(rx), 3);
|
||||
assert_empty!(rx);
|
||||
}
|
||||
|
||||
/// `len` after lagging still counts from the old cursor until Lagged is
|
||||
/// observed and the cursor advances; then `len` reflects retained messages.
|
||||
#[test]
|
||||
fn receiver_len_after_lag_error_advances_cursor() {
|
||||
let (tx, mut rx) = broadcast::channel(2);
|
||||
|
||||
assert_ok!(tx.send(1));
|
||||
assert_ok!(tx.send(2));
|
||||
assert_ok!(tx.send(3));
|
||||
assert_ok!(tx.send(4));
|
||||
|
||||
// Missed 1 and 2; buffer holds 3,4. len counts from old cursor.
|
||||
assert_eq!(rx.len(), 4);
|
||||
assert_lagged!(rx.try_recv(), 2);
|
||||
// Cursor now at oldest retained (3); two messages remain.
|
||||
assert_eq!(rx.len(), 2);
|
||||
assert_eq!(assert_recv!(rx), 3);
|
||||
assert_eq!(rx.len(), 1);
|
||||
assert_eq!(assert_recv!(rx), 4);
|
||||
assert_eq!(rx.len(), 0);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user