Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

channel: add max_capacity method #4904

Merged
merged 10 commits into from Aug 14, 2022
Merged
41 changes: 41 additions & 0 deletions tokio/src/sync/mpsc/bounded.rs
Expand Up @@ -998,6 +998,8 @@ impl<T> Sender<T> {
///
/// The capacity goes down when sending a value by calling [`send`] or by reserving capacity
/// with [`reserve`]. The capacity goes up when values are received by the [`Receiver`].
/// This is distinct from [`max_capacity`], which always returns buffer capacity initially
/// specified when calling [`channel`]
///
/// # Examples
///
Expand All @@ -1023,6 +1025,8 @@ impl<T> Sender<T> {
///
/// [`send`]: Sender::send
/// [`reserve`]: Sender::reserve
/// [`channel`]: channel
/// [`max_capacity`]: Sender::max_capacity
pub fn capacity(&self) -> usize {
self.chan.semaphore().0.available_permits()
}
Expand All @@ -1036,6 +1040,43 @@ impl<T> Sender<T> {
chan: self.chan.downgrade(),
}
}

/// Returns the maximum buffer capacity of the channel.
///
/// The maximum capacity is the buffer capacity initially specified when calling
/// [`channel`]. This is distinct from [`capacity`], which returns the *current*
/// available buffer capacity: as messages are sent and received, the
/// value returned by [`capacity`] will go up or down, while `max_capacity`
/// will always return maximum capacity that was set when the channel was
/// created.
Daksh14 marked this conversation as resolved.
Show resolved Hide resolved
///
/// # Examples
///
/// ```
/// use tokio::sync::mpsc;
///
/// #[tokio::main]
/// async fn main() {
/// let (tx, _rx) = mpsc::channel::<()>(5);
///
/// // both max capacity and capacity are the same at first
/// assert_eq!(tx.max_capacity(), 5);
/// assert_eq!(tx.capacity(), 5);
///
/// // Making a reservation doesn't change the max capacity.
/// let permit = tx.reserve().await.unwrap();
/// assert_eq!(tx.max_capacity(), 5);
/// // but drops the capacity by one
/// assert_eq!(tx.capacity(), 4);
hawkw marked this conversation as resolved.
Show resolved Hide resolved
/// }
Daksh14 marked this conversation as resolved.
Show resolved Hide resolved
/// ```
///
/// [`channel`]: channel
/// [`max_capacity`]: Sender::max_capacity
/// [`capacity`]: Sender::capacity
pub fn max_capacity(&self) -> usize {
self.chan.semaphore().1
}
Daksh14 marked this conversation as resolved.
Show resolved Hide resolved
}

impl<T> Clone for Sender<T> {
Expand Down
19 changes: 19 additions & 0 deletions tokio/tests/sync_mpsc.rs
Expand Up @@ -921,3 +921,22 @@ async fn test_tx_count_weak_sender() {

assert!(tx_weak.upgrade().is_none() && tx_weak2.upgrade().is_none());
}

// Tests that channel `capacity` changes and `max_capacity` stays the same
#[tokio::test]
async fn test_tx_capacity() {
let (tx, _rx) = mpsc::channel::<()>(10);
// both capacities are same before
assert_eq!(tx.capacity(), 10);
assert_eq!(tx.max_capacity(), 10);

let _permit = tx.reserve().await.unwrap();
// after reserve, only capacity should drop by one
assert_eq!(tx.capacity(), 9);
assert_eq!(tx.max_capacity(), 10);
Daksh14 marked this conversation as resolved.
Show resolved Hide resolved

let _sent = tx.send(()).await.unwrap();
// after send, capacity should drop by one again
assert_eq!(tx.capacity(), 8);
assert_eq!(tx.max_capacity(), 10);
}