tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-15 00:00:15 +02:00
Code Issues Packages Projects Releases Wiki Activity

sync: support mpsc send with &self (#2861)

Browse Source 
Updates the mpsc channel to use the intrusive waker based sempahore.
This enables using `Sender` with `&self`.

Instead of using `Sender::poll_ready` to ensure capacity and updating
the `Sender` state, `async fn Sender::reserve()` is added. This function
returns a `Permit` value representing the reserved capacity.

Fixes: #2637
Refs: #2718 (intrusive waiters)
This commit is contained in:
Carl Lerche
2020-09-24 17:26:38 -07:00
committed by GitHub
parent 4186b0aa38
commit cf025ba45f
19 changed files with 473 additions and 2575 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tokio-test/src/io.rs
+3 -1
View File
@@ -200,7 +200,9 @@ impl Inner {
}
fn poll_action(&mut self, cx: &mut task::Context<'_>) -> Poll<Option<Action>> {
self.rx.poll_recv(cx)
use futures_core::stream::Stream;
Pin::new(&mut self.rx).poll_next(cx)
}
fn read(&mut self, dst: &mut ReadBuf<'_>) -> io::Result<()> {
tokio/src/signal/unix.rs
+1 -29
View File
@@ -391,35 +391,7 @@ impl Signal {
poll_fn(|cx| self.poll_recv(cx)).await
}
/// Polls to receive the next signal notification event, outside of an
/// `async` context.
///
/// `None` is returned if no more events can be received by this stream.
///
/// # Examples
///
/// Polling from a manually implemented future
///
/// ```rust,no_run
/// use std::pin::Pin;
/// use std::future::Future;
/// use std::task::{Context, Poll};
/// use tokio::signal::unix::Signal;
///
/// struct MyFuture {
/// signal: Signal,
/// }
///
/// impl Future for MyFuture {
/// type Output = Option<()>;
///
/// fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
/// println!("polling MyFuture");
/// self.signal.poll_recv(cx)
/// }
/// }
/// ```
pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<()>> {
pub(crate) fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<()>> {
self.rx.poll_recv(cx)
}
}
tokio/src/stream/mod.rs
+2 -2
View File
@@ -270,8 +270,8 @@ pub trait StreamExt: Stream {
/// # #[tokio::main(basic_scheduler)]
/// async fn main() {
/// # time::pause();
/// let (mut tx1, rx1) = mpsc::channel(10);
/// let (mut tx2, rx2) = mpsc::channel(10);
/// let (tx1, rx1) = mpsc::channel(10);
/// let (tx2, rx2) = mpsc::channel(10);
///
/// let mut rx = rx1.merge(rx2);
///
tokio/src/stream/stream_map.rs
+2 -2
View File
@@ -57,8 +57,8 @@ use std::task::{Context, Poll};
///
/// #[tokio::main]
/// async fn main() {
/// let (mut tx1, rx1) = mpsc::channel(10);
/// let (mut tx2, rx2) = mpsc::channel(10);
/// let (tx1, rx1) = mpsc::channel(10);
/// let (tx2, rx2) = mpsc::channel(10);
///
/// tokio::spawn(async move {
/// tx1.send(1).await.unwrap();
tokio/src/sync/batch_semaphore.rs
+7 -3
View File
@@ -165,7 +165,6 @@ impl Semaphore {
/// permits and notifies all pending waiters.
// This will be used once the bounded MPSC is updated to use the new
// semaphore implementation.
#[allow(dead_code)]
pub(crate) fn close(&self) {
let mut waiters = self.waiters.lock().unwrap();
// If the semaphore's permits counter has enough permits for an
@@ -185,6 +184,11 @@ impl Semaphore {
}
}
/// Returns true if the semaphore is closed
pub(crate) fn is_closed(&self) -> bool {
self.permits.load(Acquire) & Self::CLOSED == Self::CLOSED
}
pub(crate) fn try_acquire(&self, num_permits: u32) -> Result<(), TryAcquireError> {
assert!(
num_permits as usize <= Self::MAX_PERMITS,
@@ -194,8 +198,8 @@ impl Semaphore {
let num_permits = (num_permits as usize) << Self::PERMIT_SHIFT;
let mut curr = self.permits.load(Acquire);
loop {
// Has the semaphore closed?git
if curr & Self::CLOSED > 0 {
// Has the semaphore closed?
if curr & Self::CLOSED == Self::CLOSED {
return Err(TryAcquireError::Closed);
}
tokio/src/sync/mod.rs
+4 -5
View File
@@ -106,7 +106,7 @@
//!
//! #[tokio::main]
//! async fn main() {
//! let (mut tx, mut rx) = mpsc::channel(100);
//! let (tx, mut rx) = mpsc::channel(100);
//!
//! tokio::spawn(async move {
//! for i in 0..10 {
@@ -150,7 +150,7 @@
//! for _ in 0..10 {
//! // Each task needs its own `tx` handle. This is done by cloning the
//! // original handle.
//! let mut tx = tx.clone();
//! let tx = tx.clone();
//!
//! tokio::spawn(async move {
//! tx.send(&b"data to write"[..]).await.unwrap();
@@ -213,7 +213,7 @@
//!
//! // Spawn tasks that will send the increment command.
//! for _ in 0..10 {
//! let mut cmd_tx = cmd_tx.clone();
//! let cmd_tx = cmd_tx.clone();
//!
//! join_handles.push(tokio::spawn(async move {
//! let (resp_tx, resp_rx) = oneshot::channel();
@@ -443,7 +443,6 @@ cfg_sync! {
pub mod oneshot;
pub(crate) mod batch_semaphore;
pub(crate) mod semaphore_ll;
mod semaphore;
pub use semaphore::{Semaphore, SemaphorePermit, OwnedSemaphorePermit};
@@ -473,7 +472,7 @@ cfg_not_sync! {
cfg_signal_internal! {
pub(crate) mod mpsc;
pub(crate) mod semaphore_ll;
pub(crate) mod batch_semaphore;
}
}
tokio/src/sync/mpsc/bounded.rs
+212 -144
View File
@@ -1,6 +1,6 @@
use crate::sync::batch_semaphore::{self as semaphore, TryAcquireError};
use crate::sync::mpsc::chan;
use crate::sync::mpsc::error::{ClosedError, SendError, TryRecvError, TrySendError};
use crate::sync::semaphore_ll as semaphore;
use crate::sync::mpsc::error::{SendError, TryRecvError, TrySendError};
cfg_time! {
use crate::sync::mpsc::error::SendTimeoutError;
@@ -8,6 +8,7 @@ cfg_time! {
}
use std::fmt;
#[cfg(any(feature = "signal", feature = "process", feature = "stream"))]
use std::task::{Context, Poll};
/// Send values to the associated `Receiver`.
@@ -17,20 +18,14 @@ pub struct Sender<T> {
chan: chan::Tx<T, Semaphore>,
}
impl<T> Clone for Sender<T> {
fn clone(&self) -> Self {
Sender {
chan: self.chan.clone(),
}
}
}
impl<T> fmt::Debug for Sender<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Sender")
.field("chan", &self.chan)
.finish()
}
/// Permit to send one value into the channel.
///
/// `Permit` values are returned by [`Sender::reserve()`] and are used to
/// guarantee channel capacity before generating a message to send.
///
/// [`Sender::reserve()`]: Sender::reserve
pub struct Permit<'a, T> {
chan: &'a chan::Tx<T, Semaphore>,
}
/// Receive values from the associated `Sender`.
@@ -41,14 +36,6 @@ pub struct Receiver<T> {
chan: chan::Rx<T, Semaphore>,
}
impl<T> fmt::Debug for Receiver<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Receiver")
.field("chan", &self.chan)
.finish()
}
}
/// Creates a bounded mpsc channel for communicating between asynchronous tasks
/// with backpressure.
///
@@ -77,7 +64,7 @@ impl<T> fmt::Debug for Receiver<T> {
///
/// #[tokio::main]
/// async fn main() {
/// let (mut tx, mut rx) = mpsc::channel(100);
/// let (tx, mut rx) = mpsc::channel(100);
///
/// tokio::spawn(async move {
/// for i in 0..10 {
@@ -125,7 +112,7 @@ impl<T> Receiver<T> {
///
/// #[tokio::main]
/// async fn main() {
/// let (mut tx, mut rx) = mpsc::channel(100);
/// let (tx, mut rx) = mpsc::channel(100);
///
/// tokio::spawn(async move {
/// tx.send("hello").await.unwrap();
@@ -143,7 +130,7 @@ impl<T> Receiver<T> {
///
/// #[tokio::main]
/// async fn main() {
/// let (mut tx, mut rx) = mpsc::channel(100);
/// let (tx, mut rx) = mpsc::channel(100);
///
/// tx.send("hello").await.unwrap();
/// tx.send("world").await.unwrap();
@@ -154,12 +141,11 @@ impl<T> Receiver<T> {
/// ```
pub async fn recv(&mut self) -> Option<T> {
use crate::future::poll_fn;
poll_fn(|cx| self.poll_recv(cx)).await
poll_fn(|cx| self.chan.recv(cx)).await
}
#[doc(hidden)] // TODO: document
pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<T>> {
#[cfg(any(feature = "signal", feature = "process"))]
pub(crate) fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<T>> {
self.chan.recv(cx)
}
@@ -178,7 +164,7 @@ impl<T> Receiver<T> {
/// use tokio::sync::mpsc;
///
/// fn main() {
/// let (mut tx, mut rx) = mpsc::channel::<u8>(10);
/// let (tx, mut rx) = mpsc::channel::<u8>(10);
///
/// let sync_code = thread::spawn(move || {
/// assert_eq!(Some(10), rx.blocking_recv());
@@ -215,12 +201,53 @@ impl<T> Receiver<T> {
/// Closes the receiving half of a channel, without dropping it.
///
/// This prevents any further messages from being sent on the channel while
/// still enabling the receiver to drain messages that are buffered.
/// still enabling the receiver to drain messages that are buffered. Any
/// outstanding [`Permit`] values will still be able to send messages.
///
/// In order to guarantee no messages are dropped, after calling `close()`,
/// `recv()` must be called until `None` is returned.
///
/// [`Permit`]: Permit
///
/// # Examples
///
/// ```
/// use tokio::sync::mpsc;
///
/// #[tokio::main]
/// async fn main() {
/// let (tx, mut rx) = mpsc::channel(20);
///
/// tokio::spawn(async move {
/// let mut i = 0;
/// while let Ok(permit) = tx.reserve().await {
/// permit.send(i);
/// i += 1;
/// }
/// });
///
/// rx.close();
///
/// while let Some(msg) = rx.recv().await {
/// println!("got {}", msg);
/// }
///
/// // Channel closed and no messages are lost.
/// }
/// ```
pub fn close(&mut self) {
self.chan.close();
}
}
impl<T> fmt::Debug for Receiver<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Receiver")
.field("chan", &self.chan)
.finish()
}
}
impl<T> Unpin for Receiver<T> {}
cfg_stream! {
@@ -228,7 +255,7 @@ cfg_stream! {
type Item = T;
fn poll_next(mut self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
self.poll_recv(cx)
self.chan.recv(cx)
}
}
}
@@ -267,7 +294,7 @@ impl<T> Sender<T> {
///
/// #[tokio::main]
/// async fn main() {
/// let (mut tx, mut rx) = mpsc::channel(1);
/// let (tx, mut rx) = mpsc::channel(1);
///
/// tokio::spawn(async move {
/// for i in 0..10 {
@@ -283,17 +310,13 @@ impl<T> Sender<T> {
/// }
/// }
/// ```
pub async fn send(&mut self, value: T) -> Result<(), SendError<T>> {
use crate::future::poll_fn;
if poll_fn(|cx| self.poll_ready(cx)).await.is_err() {
return Err(SendError(value));
}
match self.try_send(value) {
Ok(()) => Ok(()),
Err(TrySendError::Full(_)) => unreachable!(),
Err(TrySendError::Closed(value)) => Err(SendError(value)),
pub async fn send(&self, value: T) -> Result<(), SendError<T>> {
match self.reserve().await {
Ok(permit) => {
permit.send(value);
Ok(())
}
Err(_) => Err(SendError(value)),
}
}
@@ -304,9 +327,6 @@ impl<T> Sender<T> {
/// with [`send`], this function has two failure cases instead of one (one for
/// disconnection, one for a full buffer).
///
/// This function may be paired with [`poll_ready`] in order to wait for
/// channel capacity before trying to send a value.
///
/// # Errors
///
/// If the channel capacity has been reached, i.e., the channel has `n`
@@ -318,7 +338,6 @@ impl<T> Sender<T> {
/// an error. The error includes the value passed to `send`.
///
/// [`send`]: Sender::send
/// [`poll_ready`]: Sender::poll_ready
/// [`channel`]: channel
/// [`close`]: Receiver::close
///
@@ -330,8 +349,8 @@ impl<T> Sender<T> {
/// #[tokio::main]
/// async fn main() {
/// // Create a channel with buffer size 1
/// let (mut tx1, mut rx) = mpsc::channel(1);
/// let mut tx2 = tx1.clone();
/// let (tx1, mut rx) = mpsc::channel(1);
/// let tx2 = tx1.clone();
///
/// tokio::spawn(async move {
/// tx1.send(1).await.unwrap();
@@ -359,8 +378,15 @@ impl<T> Sender<T> {
/// }
/// }
/// ```
pub fn try_send(&mut self, message: T) -> Result<(), TrySendError<T>> {
self.chan.try_send(message)?;
pub fn try_send(&self, message: T) -> Result<(), TrySendError<T>> {
match self.chan.semaphore().0.try_acquire(1) {
Ok(_) => {}
Err(TryAcquireError::Closed) => return Err(TrySendError::Closed(message)),
Err(TryAcquireError::NoPermits) => return Err(TrySendError::Full(message)),
}
// Send the message
self.chan.send(message);
Ok(())
}
@@ -392,7 +418,7 @@ impl<T> Sender<T> {
///
/// #[tokio::main]
/// async fn main() {
/// let (mut tx, mut rx) = mpsc::channel(1);
/// let (tx, mut rx) = mpsc::channel(1);
///
/// tokio::spawn(async move {
/// for i in 0..10 {
@@ -412,27 +438,22 @@ impl<T> Sender<T> {
#[cfg(feature = "time")]
#[cfg_attr(docsrs, doc(cfg(feature = "time")))]
pub async fn send_timeout(
&mut self,
&self,
value: T,
timeout: Duration,
) -> Result<(), SendTimeoutError<T>> {
use crate::future::poll_fn;
match crate::time::timeout(timeout, poll_fn(|cx| self.poll_ready(cx))).await {
let permit = match crate::time::timeout(timeout, self.reserve()).await {
Err(_) => {
return Err(SendTimeoutError::Timeout(value));
}
Ok(Err(_)) => {
return Err(SendTimeoutError::Closed(value));
}
Ok(_) => {}
}
Ok(Ok(permit)) => permit,
};
match self.try_send(value) {
Ok(()) => Ok(()),
Err(TrySendError::Full(_)) => unreachable!(),
Err(TrySendError::Closed(value)) => Err(SendTimeoutError::Closed(value)),
}
permit.send(value);
Ok(())
}
/// Blocking send to call outside of asynchronous contexts.
@@ -450,7 +471,7 @@ impl<T> Sender<T> {
/// use tokio::sync::mpsc;
///
/// fn main() {
/// let (mut tx, mut rx) = mpsc::channel::<u8>(1);
/// let (tx, mut rx) = mpsc::channel::<u8>(1);
///
/// let sync_code = thread::spawn(move || {
/// tx.blocking_send(10).unwrap();
@@ -462,92 +483,139 @@ impl<T> Sender<T> {
/// sync_code.join().unwrap()
/// }
/// ```
pub fn blocking_send(&mut self, value: T) -> Result<(), SendError<T>> {
pub fn blocking_send(&self, value: T) -> Result<(), SendError<T>> {
let mut enter_handle = crate::runtime::enter::enter(false);
enter_handle.block_on(self.send(value)).unwrap()
}
/// Returns `Poll::Ready(Ok(()))` when the channel is able to accept another item.
/// Wait for channel capacity. Once capacity to send one message is
/// available, it is reserved for the caller.
///
/// If the channel is full, then `Poll::Pending` is returned and the task is notified when a
/// slot becomes available.
/// If the channel is full, the function waits for the number of unreceived
/// messages to become less than the channel capacity. Capacity to send one
/// message is reserved for the caller. A [`Permit`] is returned to track
/// the reserved capacity. The [`send`] function on [`Permit`] consumes the
/// reserved capacity.
///
/// Once `poll_ready` returns `Poll::Ready(Ok(()))`, a call to `try_send` will succeed unless
/// the channel has since been closed. To provide this guarantee, the channel reserves one slot
/// in the channel for the coming send. This reserved slot is not available to other `Sender`
/// instances, so you need to be careful to not end up with deadlocks by blocking after calling
/// `poll_ready` but before sending an element.
/// Dropping [`Permit`] without sending a message releases the capacity back
/// to the channel.
///
/// If, after `poll_ready` succeeds, you decide you do not wish to send an item after all, you
/// can use [`disarm`](Sender::disarm) to release the reserved slot.
/// [`Permit`]: Permit
/// [`send`]: Permit::send
///
/// Until an item is sent or [`disarm`](Sender::disarm) is called, repeated calls to
/// `poll_ready` will return either `Poll::Ready(Ok(()))` or `Poll::Ready(Err(_))` if channel
/// is closed.
pub fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), ClosedError>> {
self.chan.poll_ready(cx).map_err(|_| ClosedError::new())
}
/// # Examples
///
/// ```
/// use tokio::sync::mpsc;
///
/// #[tokio::main]
/// async fn main() {
/// let (tx, mut rx) = mpsc::channel(1);
///
/// // Reserve capacity
/// let permit = tx.reserve().await.unwrap();
///
/// // Trying to send directly on the `tx` will fail due to no
/// // available capacity.
/// assert!(tx.try_send(123).is_err());
///
/// // Sending on the permit succeeds
/// permit.send(456);
///
/// // The value sent on the permit is received
/// assert_eq!(rx.recv().await.unwrap(), 456);
/// }
/// ```
pub async fn reserve(&self) -> Result<Permit<'_, T>, SendError<()>> {
match self.chan.semaphore().0.acquire(1).await {
Ok(_) => {}
Err(_) => return Err(SendError(())),
}
/// Undo a successful call to `poll_ready`.
///
/// Once a call to `poll_ready` returns `Poll::Ready(Ok(()))`, it holds up one slot in the
/// channel to make room for the coming send. `disarm` allows you to give up that slot if you
/// decide you do not wish to send an item after all. After calling `disarm`, you must call
/// `poll_ready` until it returns `Poll::Ready(Ok(()))` before attempting to send again.
///
/// Returns `false` if no slot is reserved for this sender (usually because `poll_ready` was
/// not previously called, or did not succeed).
///
/// # Motivation
///
/// Since `poll_ready` takes up one of the finite number of slots in a bounded channel, callers
/// need to send an item shortly after `poll_ready` succeeds. If they do not, idle senders may
/// take up all the slots of the channel, and prevent active senders from getting any requests
/// through. Consider this code that forwards from one channel to another:
///
/// ```rust,ignore
/// loop {
/// ready!(tx.poll_ready(cx))?;
/// if let Some(item) = ready!(rx.poll_recv(cx)) {
/// tx.try_send(item)?;
/// } else {
/// break;
/// }
/// }
/// ```
///
/// If many such forwarders exist, and they all forward into a single (cloned) `Sender`, then
/// any number of forwarders may be waiting for `rx.poll_recv` at the same time. While they do,
/// they are effectively each reducing the channel's capacity by 1. If enough of these
/// forwarders are idle, forwarders whose `rx` _do_ have elements will be unable to find a spot
/// for them through `poll_ready`, and the system will deadlock.
///
/// `disarm` solves this problem by allowing you to give up the reserved slot if you find that
/// you have to block. We can then fix the code above by writing:
///
/// ```rust,ignore
/// loop {
/// ready!(tx.poll_ready(cx))?;
/// let item = rx.poll_recv(cx);
/// if let Poll::Ready(Ok(_)) = item {
/// // we're going to send the item below, so don't disarm
/// } else {
/// // give up our send slot, we won't need it for a while
/// tx.disarm();
/// }
/// if let Some(item) = ready!(item) {
/// tx.try_send(item)?;
/// } else {
/// break;
/// }
/// }
/// ```
pub fn disarm(&mut self) -> bool {
if self.chan.is_ready() {
self.chan.disarm();
true
} else {
false
Ok(Permit { chan: &self.chan })
}
}
impl<T> Clone for Sender<T> {
fn clone(&self) -> Self {
Sender {
chan: self.chan.clone(),
}
}
}
impl<T> fmt::Debug for Sender<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Sender")
.field("chan", &self.chan)
.finish()
}
}
// ===== impl Permit =====
impl<T> Permit<'_, T> {
/// Sends a value using the reserved capacity.
///
/// Capacity for the message has already been reserved. The message is sent
/// to the receiver and the permit is consumed. The operation will succeed
/// even if the receiver half has been closed. See [`Receiver::close`] for
/// more details on performing a clean shutdown.
///
/// [`Receiver::close`]: Receiver::close
///
/// # Examples
///
/// ```
/// use tokio::sync::mpsc;
///
/// #[tokio::main]
/// async fn main() {
/// let (tx, mut rx) = mpsc::channel(1);
///
/// // Reserve capacity
/// let permit = tx.reserve().await.unwrap();
///
/// // Trying to send directly on the `tx` will fail due to no
/// // available capacity.
/// assert!(tx.try_send(123).is_err());
///
/// // Send a message on the permit
/// permit.send(456);
///
/// // The value sent on the permit is received
/// assert_eq!(rx.recv().await.unwrap(), 456);
/// }
/// ```
pub fn send(self, value: T) {
use std::mem;
self.chan.send(value);
// Avoid the drop logic
mem::forget(self);
}
}
impl<T> Drop for Permit<'_, T> {
fn drop(&mut self) {
use chan::Semaphore;
let semaphore = self.chan.semaphore();
// Add the permit back to the semaphore
semaphore.add_permit();
if semaphore.is_closed() && semaphore.is_idle() {
self.chan.wake_rx();
}
}
}
impl<T> fmt::Debug for Permit<'_, T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Permit")
.field("chan", &self.chan)
.finish()
}
}
tokio/src/sync/mpsc/chan.rs
+37 -235
View File
@@ -2,8 +2,8 @@ use crate::loom::cell::UnsafeCell;
use crate::loom::future::AtomicWaker;
use crate::loom::sync::atomic::AtomicUsize;
use crate::loom::sync::Arc;
use crate::sync::mpsc::error::{ClosedError, TryRecvError};
use crate::sync::mpsc::{error, list};
use crate::sync::mpsc::error::TryRecvError;
use crate::sync::mpsc::list;
use std::fmt;
use std::process;
@@ -12,21 +12,13 @@ use std::task::Poll::{Pending, Ready};
use std::task::{Context, Poll};
/// Channel sender
pub(crate) struct Tx<T, S: Semaphore> {
pub(crate) struct Tx<T, S> {
inner: Arc<Chan<T, S>>,
permit: S::Permit,
}
impl<T, S: Semaphore> fmt::Debug for Tx<T, S>
where
S::Permit: fmt::Debug,
S: fmt::Debug,
{
impl<T, S: fmt::Debug> fmt::Debug for Tx<T, S> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Tx")
.field("inner", &self.inner)
.field("permit", &self.permit)
.finish()
fmt.debug_struct("Tx").field("inner", &self.inner).finish()
}
}
@@ -35,71 +27,20 @@ pub(crate) struct Rx<T, S: Semaphore> {
inner: Arc<Chan<T, S>>,
}
impl<T, S: Semaphore> fmt::Debug for Rx<T, S>
where
S: fmt::Debug,
{
impl<T, S: Semaphore + fmt::Debug> fmt::Debug for Rx<T, S> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Rx").field("inner", &self.inner).finish()
}
}
#[derive(Debug, Eq, PartialEq)]
pub(crate) enum TrySendError {
Closed,
Full,
}
impl<T> From<(T, TrySendError)> for error::SendError<T> {
fn from(src: (T, TrySendError)) -> error::SendError<T> {
match src.1 {
TrySendError::Closed => error::SendError(src.0),
TrySendError::Full => unreachable!(),
}
}
}
impl<T> From<(T, TrySendError)> for error::TrySendError<T> {
fn from(src: (T, TrySendError)) -> error::TrySendError<T> {
match src.1 {
TrySendError::Closed => error::TrySendError::Closed(src.0),
TrySendError::Full => error::TrySendError::Full(src.0),
}
}
}
pub(crate) trait Semaphore {
type Permit;
fn new_permit() -> Self::Permit;
/// The permit is dropped without a value being sent. In this case, the
/// permit must be returned to the semaphore.
///
/// # Return
///
/// Returns true if the permit was acquired.
fn drop_permit(&self, permit: &mut Self::Permit) -> bool;
fn is_idle(&self) -> bool;
fn add_permit(&self);
fn poll_acquire(
&self,
cx: &mut Context<'_>,
permit: &mut Self::Permit,
) -> Poll<Result<(), ClosedError>>;
fn try_acquire(&self, permit: &mut Self::Permit) -> Result<(), TrySendError>;
/// A value was sent into the channel and the permit held by `tx` is
/// dropped. In this case, the permit should not immeditely be returned to
/// the semaphore. Instead, the permit is returnred to the semaphore once
/// the sent value is read by the rx handle.
fn forget(&self, permit: &mut Self::Permit);
fn close(&self);
fn is_closed(&self) -> bool;
}
struct Chan<T, S> {
@@ -157,10 +98,7 @@ impl<T> fmt::Debug for RxFields<T> {
unsafe impl<T: Send, S: Send> Send for Chan<T, S> {}
unsafe impl<T: Send, S: Sync> Sync for Chan<T, S> {}
pub(crate) fn channel<T, S>(semaphore: S) -> (Tx<T, S>, Rx<T, S>)
where
S: Semaphore,
{
pub(crate) fn channel<T, S: Semaphore>(semaphore: S) -> (Tx<T, S>, Rx<T, S>) {
let (tx, rx) = list::channel();
let chan = Arc::new(Chan {
@@ -179,48 +117,27 @@ where
// ===== impl Tx =====
impl<T, S> Tx<T, S>
where
S: Semaphore,
{
impl<T, S> Tx<T, S> {
fn new(chan: Arc<Chan<T, S>>) -> Tx<T, S> {
Tx {
inner: chan,
permit: S::new_permit(),
}
Tx { inner: chan }
}
pub(crate) fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), ClosedError>> {
self.inner.semaphore.poll_acquire(cx, &mut self.permit)
}
pub(crate) fn disarm(&mut self) {
// TODO: should this error if not acquired?
self.inner.semaphore.drop_permit(&mut self.permit);
pub(super) fn semaphore(&self) -> &S {
&self.inner.semaphore
}
/// Send a message and notify the receiver.
pub(crate) fn try_send(&mut self, value: T) -> Result<(), (T, TrySendError)> {
self.inner.try_send(value, &mut self.permit)
pub(crate) fn send(&self, value: T) {
self.inner.send(value);
}
/// Wake the receive half
pub(crate) fn wake_rx(&self) {
self.inner.rx_waker.wake();
}
}
impl<T> Tx<T, (crate::sync::semaphore_ll::Semaphore, usize)> {
pub(crate) fn is_ready(&self) -> bool {
self.permit.is_acquired()
}
}
impl<T> Tx<T, AtomicUsize> {
pub(crate) fn send_unbounded(&self, value: T) -> Result<(), (T, TrySendError)> {
self.inner.try_send(value, &mut ())
}
}
impl<T, S> Clone for Tx<T, S>
where
S: Semaphore,
{
impl<T, S> Clone for Tx<T, S> {
fn clone(&self) -> Tx<T, S> {
// Using a Relaxed ordering here is sufficient as the caller holds a
// strong ref to `self`, preventing a concurrent decrement to zero.
@@ -228,22 +145,12 @@ where
Tx {
inner: self.inner.clone(),
permit: S::new_permit(),
}
}
}
impl<T, S> Drop for Tx<T, S>
where
S: Semaphore,
{
impl<T, S> Drop for Tx<T, S> {
fn drop(&mut self) {
let notify = self.inner.semaphore.drop_permit(&mut self.permit);
if notify && self.inner.semaphore.is_idle() {
self.inner.rx_waker.wake();
}
if self.inner.tx_count.fetch_sub(1, AcqRel) != 1 {
return;
}
@@ -252,16 +159,13 @@ where
self.inner.tx.close();
// Notify the receiver
self.inner.rx_waker.wake();
self.wake_rx();
}
}
// ===== impl Rx =====
impl<T, S> Rx<T, S>
where
S: Semaphore,
{
impl<T, S: Semaphore> Rx<T, S> {
fn new(chan: Arc<Chan<T, S>>) -> Rx<T, S> {
Rx { inner: chan }
}
@@ -349,10 +253,7 @@ where
}
}
impl<T, S> Drop for Rx<T, S>
where
S: Semaphore,
{
impl<T, S: Semaphore> Drop for Rx<T, S> {
fn drop(&mut self) {
use super::block::Read::Value;
@@ -370,25 +271,13 @@ where
// ===== impl Chan =====
impl<T, S> Chan<T, S>
where
S: Semaphore,
{
fn try_send(&self, value: T, permit: &mut S::Permit) -> Result<(), (T, TrySendError)> {
if let Err(e) = self.semaphore.try_acquire(permit) {
return Err((value, e));
}
impl<T, S> Chan<T, S> {
fn send(&self, value: T) {
// Push the value
self.tx.push(value);
// Notify the rx task
self.rx_waker.wake();
// Release the permit
self.semaphore.forget(permit);
Ok(())
}
}
@@ -407,74 +296,24 @@ impl<T, S> Drop for Chan<T, S> {
}
}
use crate::sync::semaphore_ll::TryAcquireError;
impl From<TryAcquireError> for TrySendError {
fn from(src: TryAcquireError) -> TrySendError {
if src.is_closed() {
TrySendError::Closed
} else if src.is_no_permits() {
TrySendError::Full
} else {
unreachable!();
}
}
}
// ===== impl Semaphore for (::Semaphore, capacity) =====
use crate::sync::semaphore_ll::Permit;
impl Semaphore for (crate::sync::semaphore_ll::Semaphore, usize) {
type Permit = Permit;
fn new_permit() -> Permit {
Permit::new()
}
fn drop_permit(&self, permit: &mut Permit) -> bool {
let ret = permit.is_acquired();
permit.release(1, &self.0);
ret
}
impl Semaphore for (crate::sync::batch_semaphore::Semaphore, usize) {
fn add_permit(&self) {
self.0.add_permits(1)
self.0.release(1)
}
fn is_idle(&self) -> bool {
self.0.available_permits() == self.1
}
fn poll_acquire(
&self,
cx: &mut Context<'_>,
permit: &mut Permit,
) -> Poll<Result<(), ClosedError>> {
// Keep track of task budget
let coop = ready!(crate::coop::poll_proceed(cx));
permit
.poll_acquire(cx, 1, &self.0)
.map_err(|_| ClosedError::new())
.map(move |r| {
coop.made_progress();
r
})
}
fn try_acquire(&self, permit: &mut Permit) -> Result<(), TrySendError> {
permit.try_acquire(1, &self.0)?;
Ok(())
}
fn forget(&self, permit: &mut Self::Permit) {
permit.forget(1);
}
fn close(&self) {
self.0.close();
}
fn is_closed(&self) -> bool {
self.0.is_closed()
}
}
// ===== impl Semaphore for AtomicUsize =====
@@ -483,14 +322,6 @@ use std::sync::atomic::Ordering::{Acquire, Release};
use std::usize;
impl Semaphore for AtomicUsize {
type Permit = ();
fn new_permit() {}
fn drop_permit(&self, _permit: &mut ()) -> bool {
false
}
fn add_permit(&self) {
let prev = self.fetch_sub(2, Release);
@@ -504,40 +335,11 @@ impl Semaphore for AtomicUsize {
self.load(Acquire) >> 1 == 0
}
fn poll_acquire(
&self,
_cx: &mut Context<'_>,
permit: &mut (),
) -> Poll<Result<(), ClosedError>> {
Ready(self.try_acquire(permit).map_err(|_| ClosedError::new()))
}
fn try_acquire(&self, _permit: &mut ()) -> Result<(), TrySendError> {
let mut curr = self.load(Acquire);
loop {
if curr & 1 == 1 {
return Err(TrySendError::Closed);
}
if curr == usize::MAX ^ 1 {
// Overflowed the ref count. There is no safe way to recover, so
// abort the process. In practice, this should never happen.
process::abort()
}
match self.compare_exchange(curr, curr + 2, AcqRel, Acquire) {
Ok(_) => return Ok(()),
Err(actual) => {
curr = actual;
}
}
}
}
fn forget(&self, _permit: &mut ()) {}
fn close(&self) {
self.fetch_or(1, Release);
}
fn is_closed(&self) -> bool {
self.load(Acquire) & 1 == 1
}
}
tokio/src/sync/mpsc/error.rs
-20
View File
@@ -94,26 +94,6 @@ impl fmt::Display for TryRecvError {
impl Error for TryRecvError {}
// ===== ClosedError =====
/// Error returned by [`Sender::poll_ready`](super::Sender::poll_ready).
#[derive(Debug)]
pub struct ClosedError(());
impl ClosedError {
pub(crate) fn new() -> ClosedError {
ClosedError(())
}
}
impl fmt::Display for ClosedError {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(fmt, "channel closed")
}
}
impl Error for ClosedError {}
cfg_time! {
// ===== SendTimeoutError =====
tokio/src/sync/mpsc/mod.rs
+1 -1
View File
@@ -76,7 +76,7 @@
pub(super) mod block;
mod bounded;
pub use self::bounded::{channel, Receiver, Sender};
pub use self::bounded::{channel, Permit, Receiver, Sender};
mod chan;
tokio/src/sync/mpsc/unbounded.rs
+36 -3
View File
@@ -73,8 +73,7 @@ impl<T> UnboundedReceiver<T> {
UnboundedReceiver { chan }
}
#[doc(hidden)] // TODO: doc
pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<T>> {
fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<T>> {
self.chan.recv(cx)
}
@@ -174,7 +173,41 @@ impl<T> UnboundedSender<T> {
/// [`close`]: UnboundedReceiver::close
/// [`UnboundedReceiver`]: UnboundedReceiver
pub fn send(&self, message: T) -> Result<(), SendError<T>> {
self.chan.send_unbounded(message)?;
if !self.inc_num_messages() {
return Err(SendError(message));
}
self.chan.send(message);
Ok(())
}
fn inc_num_messages(&self) -> bool {
use std::process;
use std::sync::atomic::Ordering::{AcqRel, Acquire};
let mut curr = self.chan.semaphore().load(Acquire);
loop {
if curr & 1 == 1 {
return false;
}
if curr == usize::MAX ^ 1 {
// Overflowed the ref count. There is no safe way to recover, so
// abort the process. In practice, this should never happen.
process::abort()
}
match self
.chan
.semaphore()
.compare_exchange(curr, curr + 2, AcqRel, Acquire)
{
Ok(_) => return true,
Err(actual) => {
curr = actual;
}
}
}
}
}
tokio/src/sync/semaphore_ll.rs
-1221
View File
File diff suppressed because it is too large Load Diff
tokio/src/sync/tests/loom_mpsc.rs
+7 -7
View File
@@ -7,17 +7,17 @@ use loom::thread;
#[test]
fn closing_tx() {
loom::model(|| {
let (mut tx, mut rx) = mpsc::channel(16);
let (tx, mut rx) = mpsc::channel(16);
thread::spawn(move || {
tx.try_send(()).unwrap();
drop(tx);
});
let v = block_on(poll_fn(|cx| rx.poll_recv(cx)));
let v = block_on(rx.recv());
assert!(v.is_some());
let v = block_on(poll_fn(|cx| rx.poll_recv(cx)));
let v = block_on(rx.recv());
assert!(v.is_none());
});
}
@@ -32,10 +32,10 @@ fn closing_unbounded_tx() {
drop(tx);
});
let v = block_on(poll_fn(|cx| rx.poll_recv(cx)));
let v = block_on(rx.recv());
assert!(v.is_some());
let v = block_on(poll_fn(|cx| rx.poll_recv(cx)));
let v = block_on(rx.recv());
assert!(v.is_none());
});
}
@@ -53,7 +53,7 @@ fn dropping_tx() {
}
drop(tx);
let v = block_on(poll_fn(|cx| rx.poll_recv(cx)));
let v = block_on(rx.recv());
assert!(v.is_none());
});
}
@@ -71,7 +71,7 @@ fn dropping_unbounded_tx() {
}
drop(tx);
let v = block_on(poll_fn(|cx| rx.poll_recv(cx)));
let v = block_on(rx.recv());
assert!(v.is_none());
});
}
tokio/src/sync/tests/loom_semaphore_ll.rs
-192
View File
@@ -1,192 +0,0 @@
use crate::sync::semaphore_ll::*;
use futures::future::poll_fn;
use loom::future::block_on;
use loom::thread;
use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering::SeqCst;
use std::sync::Arc;
use std::task::Poll::Ready;
use std::task::{Context, Poll};
#[test]
fn basic_usage() {
const NUM: usize = 2;
struct Actor {
waiter: Permit,
shared: Arc<Shared>,
}
struct Shared {
semaphore: Semaphore,
active: AtomicUsize,
}
impl Future for Actor {
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
let me = &mut *self;
ready!(me.waiter.poll_acquire(cx, 1, &me.shared.semaphore)).unwrap();
let actual = me.shared.active.fetch_add(1, SeqCst);
assert!(actual <= NUM - 1);
let actual = me.shared.active.fetch_sub(1, SeqCst);
assert!(actual <= NUM);
me.waiter.release(1, &me.shared.semaphore);
Ready(())
}
}
loom::model(|| {
let shared = Arc::new(Shared {
semaphore: Semaphore::new(NUM),
active: AtomicUsize::new(0),
});
for _ in 0..NUM {
let shared = shared.clone();
thread::spawn(move || {
block_on(Actor {
waiter: Permit::new(),
shared,
});
});
}
block_on(Actor {
waiter: Permit::new(),
shared,
});
});
}
#[test]
fn release() {
loom::model(|| {
let semaphore = Arc::new(Semaphore::new(1));
{
let semaphore = semaphore.clone();
thread::spawn(move || {
let mut permit = Permit::new();
block_on(poll_fn(|cx| permit.poll_acquire(cx, 1, &semaphore))).unwrap();
permit.release(1, &semaphore);
});
}
let mut permit = Permit::new();
block_on(poll_fn(|cx| permit.poll_acquire(cx, 1, &semaphore))).unwrap();
permit.release(1, &semaphore);
});
}
#[test]
fn basic_closing() {
const NUM: usize = 2;
loom::model(|| {
let semaphore = Arc::new(Semaphore::new(1));
for _ in 0..NUM {
let semaphore = semaphore.clone();
thread::spawn(move || {
let mut permit = Permit::new();
for _ in 0..2 {
block_on(poll_fn(|cx| {
permit.poll_acquire(cx, 1, &semaphore).map_err(|_| ())
}))?;
permit.release(1, &semaphore);
}
Ok::<(), ()>(())
});
}
semaphore.close();
});
}
#[test]
fn concurrent_close() {
const NUM: usize = 3;
loom::model(|| {
let semaphore = Arc::new(Semaphore::new(1));
for _ in 0..NUM {
let semaphore = semaphore.clone();
thread::spawn(move || {
let mut permit = Permit::new();
block_on(poll_fn(|cx| {
permit.poll_acquire(cx, 1, &semaphore).map_err(|_| ())
}))?;
permit.release(1, &semaphore);
semaphore.close();
Ok::<(), ()>(())
});
}
});
}
#[test]
fn batch() {
let mut b = loom::model::Builder::new();
b.preemption_bound = Some(1);
b.check(|| {
let semaphore = Arc::new(Semaphore::new(10));
let active = Arc::new(AtomicUsize::new(0));
let mut ths = vec![];
for _ in 0..2 {
let semaphore = semaphore.clone();
let active = active.clone();
ths.push(thread::spawn(move || {
let mut permit = Permit::new();
for n in &[4, 10, 8] {
block_on(poll_fn(|cx| permit.poll_acquire(cx, *n, &semaphore))).unwrap();
active.fetch_add(*n as usize, SeqCst);
let num_active = active.load(SeqCst);
assert!(num_active <= 10);
thread::yield_now();
active.fetch_sub(*n as usize, SeqCst);
permit.release(*n, &semaphore);
}
}));
}
for th in ths.into_iter() {
th.join().unwrap();
}
assert_eq!(10, semaphore.available_permits());
});
}
tokio/src/sync/tests/mod.rs
-2
View File
@@ -1,6 +1,5 @@
cfg_not_loom! {
mod atomic_waker;
mod semaphore_ll;
mod semaphore_batch;
}
@@ -12,6 +11,5 @@ cfg_loom! {
mod loom_notify;
mod loom_oneshot;
mod loom_semaphore_batch;
mod loom_semaphore_ll;
mod loom_watch;
}
tokio/src/sync/tests/semaphore_ll.rs
-470
View File
@@ -1,470 +0,0 @@
use crate::sync::semaphore_ll::{Permit, Semaphore};
use tokio_test::*;
#[test]
fn poll_acquire_one_available() {
let s = Semaphore::new(100);
assert_eq!(s.available_permits(), 100);
// Polling for a permit succeeds immediately
let mut permit = task::spawn(Permit::new());
assert!(!permit.is_acquired());
assert_ready_ok!(permit.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
assert_eq!(s.available_permits(), 99);
assert!(permit.is_acquired());
// Polling again on the same waiter does not claim a new permit
assert_ready_ok!(permit.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
assert_eq!(s.available_permits(), 99);
assert!(permit.is_acquired());
}
#[test]
fn poll_acquire_many_available() {
let s = Semaphore::new(100);
assert_eq!(s.available_permits(), 100);
// Polling for a permit succeeds immediately
let mut permit = task::spawn(Permit::new());
assert!(!permit.is_acquired());
assert_ready_ok!(permit.enter(|cx, mut p| p.poll_acquire(cx, 5, &s)));
assert_eq!(s.available_permits(), 95);
assert!(permit.is_acquired());
// Polling again on the same waiter does not claim a new permit
assert_ready_ok!(permit.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
assert_eq!(s.available_permits(), 95);
assert!(permit.is_acquired());
assert_ready_ok!(permit.enter(|cx, mut p| p.poll_acquire(cx, 5, &s)));
assert_eq!(s.available_permits(), 95);
assert!(permit.is_acquired());
// Polling for a larger number of permits acquires more
assert_ready_ok!(permit.enter(|cx, mut p| p.poll_acquire(cx, 8, &s)));
assert_eq!(s.available_permits(), 92);
assert!(permit.is_acquired());
}
#[test]
fn try_acquire_one_available() {
let s = Semaphore::new(100);
assert_eq!(s.available_permits(), 100);
// Polling for a permit succeeds immediately
let mut permit = Permit::new();
assert!(!permit.is_acquired());
assert_ok!(permit.try_acquire(1, &s));
assert_eq!(s.available_permits(), 99);
assert!(permit.is_acquired());
// Polling again on the same waiter does not claim a new permit
assert_ok!(permit.try_acquire(1, &s));
assert_eq!(s.available_permits(), 99);
assert!(permit.is_acquired());
}
#[test]
fn try_acquire_many_available() {
let s = Semaphore::new(100);
assert_eq!(s.available_permits(), 100);
// Polling for a permit succeeds immediately
let mut permit = Permit::new();
assert!(!permit.is_acquired());
assert_ok!(permit.try_acquire(5, &s));
assert_eq!(s.available_permits(), 95);
assert!(permit.is_acquired());
// Polling again on the same waiter does not claim a new permit
assert_ok!(permit.try_acquire(5, &s));
assert_eq!(s.available_permits(), 95);
assert!(permit.is_acquired());
}
#[test]
fn poll_acquire_one_unavailable() {
let s = Semaphore::new(1);
let mut permit_1 = task::spawn(Permit::new());
let mut permit_2 = task::spawn(Permit::new());
// Acquire the first permit
assert_ready_ok!(permit_1.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
assert_eq!(s.available_permits(), 0);
permit_2.enter(|cx, mut p| {
// Try to acquire the second permit
assert_pending!(p.poll_acquire(cx, 1, &s));
});
permit_1.release(1, &s);
assert_eq!(s.available_permits(), 0);
assert!(permit_2.is_woken());
assert_ready_ok!(permit_2.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
permit_2.release(1, &s);
assert_eq!(s.available_permits(), 1);
}
#[test]
fn forget_acquired() {
let s = Semaphore::new(1);
// Polling for a permit succeeds immediately
let mut permit = task::spawn(Permit::new());
assert_ready_ok!(permit.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
assert_eq!(s.available_permits(), 0);
permit.forget(1);
assert_eq!(s.available_permits(), 0);
}
#[test]
fn forget_waiting() {
let s = Semaphore::new(0);
// Polling for a permit succeeds immediately
let mut permit = task::spawn(Permit::new());
assert_pending!(permit.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
assert_eq!(s.available_permits(), 0);
permit.forget(1);
s.add_permits(1);
assert!(!permit.is_woken());
assert_eq!(s.available_permits(), 1);
}
#[test]
fn poll_acquire_many_unavailable() {
let s = Semaphore::new(5);
let mut permit_1 = task::spawn(Permit::new());
let mut permit_2 = task::spawn(Permit::new());
let mut permit_3 = task::spawn(Permit::new());
// Acquire the first permit
assert_ready_ok!(permit_1.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
assert_eq!(s.available_permits(), 4);
permit_2.enter(|cx, mut p| {
// Try to acquire the second permit
assert_pending!(p.poll_acquire(cx, 5, &s));
});
assert_eq!(s.available_permits(), 0);
permit_3.enter(|cx, mut p| {
// Try to acquire the third permit
assert_pending!(p.poll_acquire(cx, 3, &s));
});
permit_1.release(1, &s);
assert_eq!(s.available_permits(), 0);
assert!(permit_2.is_woken());
assert_ready_ok!(permit_2.enter(|cx, mut p| p.poll_acquire(cx, 5, &s)));
assert!(!permit_3.is_woken());
assert_eq!(s.available_permits(), 0);
permit_2.release(1, &s);
assert!(!permit_3.is_woken());
assert_eq!(s.available_permits(), 0);
permit_2.release(2, &s);
assert!(permit_3.is_woken());
assert_ready_ok!(permit_3.enter(|cx, mut p| p.poll_acquire(cx, 3, &s)));
}
#[test]
fn try_acquire_one_unavailable() {
let s = Semaphore::new(1);
let mut permit_1 = Permit::new();
let mut permit_2 = Permit::new();
// Acquire the first permit
assert_ok!(permit_1.try_acquire(1, &s));
assert_eq!(s.available_permits(), 0);
assert_err!(permit_2.try_acquire(1, &s));
permit_1.release(1, &s);
assert_eq!(s.available_permits(), 1);
assert_ok!(permit_2.try_acquire(1, &s));
permit_2.release(1, &s);
assert_eq!(s.available_permits(), 1);
}
#[test]
fn try_acquire_many_unavailable() {
let s = Semaphore::new(5);
let mut permit_1 = Permit::new();
let mut permit_2 = Permit::new();
// Acquire the first permit
assert_ok!(permit_1.try_acquire(1, &s));
assert_eq!(s.available_permits(), 4);
assert_err!(permit_2.try_acquire(5, &s));
permit_1.release(1, &s);
assert_eq!(s.available_permits(), 5);
assert_ok!(permit_2.try_acquire(5, &s));
permit_2.release(1, &s);
assert_eq!(s.available_permits(), 1);
permit_2.release(1, &s);
assert_eq!(s.available_permits(), 2);
}
#[test]
fn poll_acquire_one_zero_permits() {
let s = Semaphore::new(0);
assert_eq!(s.available_permits(), 0);
let mut permit = task::spawn(Permit::new());
// Try to acquire the permit
permit.enter(|cx, mut p| {
assert_pending!(p.poll_acquire(cx, 1, &s));
});
s.add_permits(1);
assert!(permit.is_woken());
assert_ready_ok!(permit.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
}
#[test]
#[should_panic]
fn validates_max_permits() {
use std::usize;
Semaphore::new((usize::MAX >> 2) + 1);
}
#[test]
fn close_semaphore_prevents_acquire() {
let s = Semaphore::new(5);
s.close();
assert_eq!(5, s.available_permits());
let mut permit_1 = task::spawn(Permit::new());
let mut permit_2 = task::spawn(Permit::new());
assert_ready_err!(permit_1.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
assert_eq!(5, s.available_permits());
assert_ready_err!(permit_2.enter(|cx, mut p| p.poll_acquire(cx, 2, &s)));
assert_eq!(5, s.available_permits());
}
#[test]
fn close_semaphore_notifies_permit1() {
let s = Semaphore::new(0);
let mut permit = task::spawn(Permit::new());
assert_pending!(permit.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
s.close();
assert!(permit.is_woken());
assert_ready_err!(permit.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
}
#[test]
fn close_semaphore_notifies_permit2() {
let s = Semaphore::new(2);
let mut permit1 = task::spawn(Permit::new());
let mut permit2 = task::spawn(Permit::new());
let mut permit3 = task::spawn(Permit::new());
let mut permit4 = task::spawn(Permit::new());
// Acquire a couple of permits
assert_ready_ok!(permit1.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
assert_ready_ok!(permit2.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
assert_pending!(permit3.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
assert_pending!(permit4.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
s.close();
assert!(permit3.is_woken());
assert!(permit4.is_woken());
assert_ready_err!(permit3.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
assert_ready_err!(permit4.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
assert_eq!(0, s.available_permits());
permit1.release(1, &s);
assert_eq!(1, s.available_permits());
assert_ready_err!(permit1.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
permit2.release(1, &s);
assert_eq!(2, s.available_permits());
}
#[test]
fn poll_acquire_additional_permits_while_waiting_before_assigned() {
let s = Semaphore::new(1);
let mut permit = task::spawn(Permit::new());
assert_pending!(permit.enter(|cx, mut p| p.poll_acquire(cx, 2, &s)));
assert_pending!(permit.enter(|cx, mut p| p.poll_acquire(cx, 3, &s)));
s.add_permits(1);
assert!(!permit.is_woken());
s.add_permits(1);
assert!(permit.is_woken());
assert_ready_ok!(permit.enter(|cx, mut p| p.poll_acquire(cx, 3, &s)));
}
#[test]
fn try_acquire_additional_permits_while_waiting_before_assigned() {
let s = Semaphore::new(1);
let mut permit = task::spawn(Permit::new());
assert_pending!(permit.enter(|cx, mut p| p.poll_acquire(cx, 2, &s)));
assert_err!(permit.enter(|_, mut p| p.try_acquire(3, &s)));
s.add_permits(1);
assert!(permit.is_woken());
assert_ok!(permit.enter(|_, mut p| p.try_acquire(2, &s)));
}
#[test]
fn poll_acquire_additional_permits_while_waiting_after_assigned_success() {
let s = Semaphore::new(1);
let mut permit = task::spawn(Permit::new());
assert_pending!(permit.enter(|cx, mut p| p.poll_acquire(cx, 2, &s)));
s.add_permits(2);
assert!(permit.is_woken());
assert_ready_ok!(permit.enter(|cx, mut p| p.poll_acquire(cx, 3, &s)));
}
#[test]
fn poll_acquire_additional_permits_while_waiting_after_assigned_requeue() {
let s = Semaphore::new(1);
let mut permit = task::spawn(Permit::new());
assert_pending!(permit.enter(|cx, mut p| p.poll_acquire(cx, 2, &s)));
s.add_permits(2);
assert!(permit.is_woken());
assert_pending!(permit.enter(|cx, mut p| p.poll_acquire(cx, 4, &s)));
s.add_permits(1);
assert!(permit.is_woken());
assert_ready_ok!(permit.enter(|cx, mut p| p.poll_acquire(cx, 4, &s)));
}
#[test]
fn poll_acquire_fewer_permits_while_waiting() {
let s = Semaphore::new(1);
let mut permit = task::spawn(Permit::new());
assert_pending!(permit.enter(|cx, mut p| p.poll_acquire(cx, 2, &s)));
assert_eq!(s.available_permits(), 0);
assert_ready_ok!(permit.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
assert_eq!(s.available_permits(), 0);
}
#[test]
fn poll_acquire_fewer_permits_after_assigned() {
let s = Semaphore::new(1);
let mut permit1 = task::spawn(Permit::new());
let mut permit2 = task::spawn(Permit::new());
assert_pending!(permit1.enter(|cx, mut p| p.poll_acquire(cx, 5, &s)));
assert_eq!(s.available_permits(), 0);
assert_pending!(permit2.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
s.add_permits(4);
assert!(permit1.is_woken());
assert!(!permit2.is_woken());
assert_ready_ok!(permit1.enter(|cx, mut p| p.poll_acquire(cx, 3, &s)));
assert!(permit2.is_woken());
assert_eq!(s.available_permits(), 1);
assert_ready_ok!(permit2.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
}
#[test]
fn forget_partial_1() {
let s = Semaphore::new(0);
let mut permit = task::spawn(Permit::new());
assert_pending!(permit.enter(|cx, mut p| p.poll_acquire(cx, 2, &s)));
s.add_permits(1);
assert_eq!(0, s.available_permits());
permit.release(1, &s);
assert_ready_ok!(permit.enter(|cx, mut p| p.poll_acquire(cx, 1, &s)));
assert_eq!(s.available_permits(), 0);
}
#[test]
fn forget_partial_2() {
let s = Semaphore::new(0);
let mut permit = task::spawn(Permit::new());
assert_pending!(permit.enter(|cx, mut p| p.poll_acquire(cx, 2, &s)));
s.add_permits(1);
assert_eq!(0, s.available_permits());
permit.release(1, &s);
s.add_permits(1);
assert_ready_ok!(permit.enter(|cx, mut p| p.poll_acquire(cx, 2, &s)));
assert_eq!(s.available_permits(), 0);
}
tokio/src/util/linked_list.rs
+8 -12
View File
@@ -126,7 +126,6 @@ impl<L: Link> LinkedList<L, L::Target> {
}
/// Returns whether the linked list doesn not contain any node
#[cfg_attr(any(feature = "udp", feature = "uds"), allow(unused))]
pub(crate) fn is_empty(&self) -> bool {
if self.head.is_some() {
return false;
@@ -182,20 +181,17 @@ impl<L: Link> fmt::Debug for LinkedList<L, L::Target> {
}
}
impl<L: Link> Default for LinkedList<L, L::Target> {
fn default() -> Self {
Self::new()
#[cfg(any(feature = "sync", feature = "signal", feature = "process"))]
impl<L: Link> LinkedList<L, L::Target> {
pub(crate) fn last(&self) -> Option<&L::Target> {
let tail = self.tail.as_ref()?;
unsafe { Some(&*tail.as_ptr()) }
}
}
cfg_sync! {
impl<L: Link> LinkedList<L, L::Target> {
pub(crate) fn last(&self) -> Option<&L::Target> {
let tail = self.tail.as_ref()?;
unsafe {
Some(&*tail.as_ptr())
}
}
impl<L: Link> Default for LinkedList<L, L::Target> {
fn default() -> Self {
Self::new()
}
}
tokio/tests/rt_threaded.rs
+5 -5
View File
@@ -70,7 +70,7 @@ fn many_multishot_futures() {
let (start_tx, mut chain_rx) = tokio::sync::mpsc::channel(10);
for _ in 0..CHAIN {
let (mut next_tx, next_rx) = tokio::sync::mpsc::channel(10);
let (next_tx, next_rx) = tokio::sync::mpsc::channel(10);
// Forward all the messages
rt.spawn(async move {
@@ -83,8 +83,8 @@ fn many_multishot_futures() {
}
// This final task cycles if needed
let (mut final_tx, final_rx) = tokio::sync::mpsc::channel(10);
let mut cycle_tx = start_tx.clone();
let (final_tx, final_rx) = tokio::sync::mpsc::channel(10);
let cycle_tx = start_tx.clone();
let mut rem = CYCLES;
rt.spawn(async move {
@@ -107,7 +107,7 @@ fn many_multishot_futures() {
{
rt.block_on(async move {
for mut start_tx in start_txs {
for start_tx in start_txs {
start_tx.send("ping").await.unwrap();
}
@@ -340,7 +340,7 @@ fn coop_and_block_in_place() {
.unwrap();
rt.block_on(async move {
let (mut tx, mut rx) = tokio::sync::mpsc::channel(1024);
let (tx, mut rx) = tokio::sync::mpsc::channel(1024);
// Fill the channel
for _ in 0..1024 {
tokio/tests/sync_mpsc.rs
+148 -221
View File
@@ -17,74 +17,72 @@ trait AssertSend: Send {}
impl AssertSend for mpsc::Sender<i32> {}
impl AssertSend for mpsc::Receiver<i32> {}
#[test]
fn send_recv_with_buffer() {
let (tx, rx) = mpsc::channel::<i32>(16);
let mut tx = task::spawn(tx);
let mut rx = task::spawn(rx);
#[tokio::test]
async fn send_recv_with_buffer() {
let (tx, mut rx) = mpsc::channel::<i32>(16);
// Using poll_ready / try_send
assert_ready_ok!(tx.enter(|cx, mut tx| tx.poll_ready(cx)));
tx.try_send(1).unwrap();
// let permit assert_ready_ok!(tx.reserve());
let permit = tx.reserve().await.unwrap();
permit.send(1);
// Without poll_ready
tx.try_send(2).unwrap();
drop(tx);
let val = assert_ready!(rx.enter(|cx, mut rx| rx.poll_recv(cx)));
let val = rx.recv().await;
assert_eq!(val, Some(1));
let val = assert_ready!(rx.enter(|cx, mut rx| rx.poll_recv(cx)));
let val = rx.recv().await;
assert_eq!(val, Some(2));
let val = assert_ready!(rx.enter(|cx, mut rx| rx.poll_recv(cx)));
let val = rx.recv().await;
assert!(val.is_none());
}
#[test]
fn disarm() {
let (tx, rx) = mpsc::channel::<i32>(2);
let mut tx1 = task::spawn(tx.clone());
let mut tx2 = task::spawn(tx.clone());
let mut tx3 = task::spawn(tx.clone());
let mut tx4 = task::spawn(tx);
let mut rx = task::spawn(rx);
#[tokio::test]
async fn reserve_disarm() {
let (tx, mut rx) = mpsc::channel::<i32>(2);
let tx1 = tx.clone();
let tx2 = tx.clone();
let tx3 = tx.clone();
let tx4 = tx;
// We should be able to `poll_ready` two handles without problem
assert_ready_ok!(tx1.enter(|cx, mut tx| tx.poll_ready(cx)));
assert_ready_ok!(tx2.enter(|cx, mut tx| tx.poll_ready(cx)));
let permit1 = assert_ok!(tx1.reserve().await);
let permit2 = assert_ok!(tx2.reserve().await);
// But a third should not be ready
assert_pending!(tx3.enter(|cx, mut tx| tx.poll_ready(cx)));
let mut r3 = task::spawn(tx3.reserve());
assert_pending!(r3.poll());
let mut r4 = task::spawn(tx4.reserve());
assert_pending!(r4.poll());
// Using one of the reserved slots should allow a new handle to become ready
tx1.try_send(1).unwrap();
permit1.send(1);
// We also need to receive for the slot to be free
let _ = assert_ready!(rx.enter(|cx, mut rx| rx.poll_recv(cx))).unwrap();
assert!(!r3.is_woken());
rx.recv().await.unwrap();
// Now there's a free slot!
assert_ready_ok!(tx3.enter(|cx, mut tx| tx.poll_ready(cx)));
assert_pending!(tx4.enter(|cx, mut tx| tx.poll_ready(cx)));
assert!(r3.is_woken());
assert!(!r4.is_woken());
// Dropping a ready handle should also open up a slot
drop(tx2);
assert_ready_ok!(tx4.enter(|cx, mut tx| tx.poll_ready(cx)));
assert_pending!(tx1.enter(|cx, mut tx| tx.poll_ready(cx)));
// Dropping a permit should also open up a slot
drop(permit2);
assert!(r4.is_woken());
// Explicitly disarming a handle should also open a slot
assert!(tx3.disarm());
assert_ready_ok!(tx1.enter(|cx, mut tx| tx.poll_ready(cx)));
// Disarming a non-armed sender does not free up a slot
assert!(!tx3.disarm());
assert_pending!(tx3.enter(|cx, mut tx| tx.poll_ready(cx)));
let mut r1 = task::spawn(tx1.reserve());
assert_pending!(r1.poll());
}
#[tokio::test]
async fn send_recv_stream_with_buffer() {
use tokio::stream::StreamExt;
let (mut tx, mut rx) = mpsc::channel::<i32>(16);
let (tx, mut rx) = mpsc::channel::<i32>(16);
tokio::spawn(async move {
assert_ok!(tx.send(1).await);
@@ -98,7 +96,7 @@ async fn send_recv_stream_with_buffer() {
#[tokio::test]
async fn async_send_recv_with_buffer() {
let (mut tx, mut rx) = mpsc::channel(16);
let (tx, mut rx) = mpsc::channel(16);
tokio::spawn(async move {
assert_ok!(tx.send(1).await);
@@ -110,37 +108,36 @@ async fn async_send_recv_with_buffer() {
assert_eq!(None, rx.recv().await);
}
#[test]
fn start_send_past_cap() {
let mut t1 = task::spawn(());
let mut t2 = task::spawn(());
let mut t3 = task::spawn(());
#[tokio::test]
async fn start_send_past_cap() {
use std::future::Future;
let (mut tx1, mut rx) = mpsc::channel(1);
let mut tx2 = tx1.clone();
let mut t1 = task::spawn(());
let (tx1, mut rx) = mpsc::channel(1);
let tx2 = tx1.clone();
assert_ok!(tx1.try_send(()));
t1.enter(|cx, _| {
assert_pending!(tx1.poll_ready(cx));
});
let mut r1 = Box::pin(tx1.reserve());
t1.enter(|cx, _| assert_pending!(r1.as_mut().poll(cx)));
t2.enter(|cx, _| {
assert_pending!(tx2.poll_ready(cx));
});
{
let mut r2 = task::spawn(tx2.reserve());
assert_pending!(r2.poll());
drop(r1);
assert!(rx.recv().await.is_some());
assert!(r2.is_woken());
assert!(!t1.is_woken());
}
drop(tx1);
let val = t3.enter(|cx, _| assert_ready!(rx.poll_recv(cx)));
assert!(val.is_some());
assert!(t2.is_woken());
assert!(!t1.is_woken());
drop(tx2);
let val = t3.enter(|cx, _| assert_ready!(rx.poll_recv(cx)));
assert!(val.is_none());
assert!(rx.recv().await.is_none());
}
#[test]
@@ -149,26 +146,20 @@ fn buffer_gteq_one() {
mpsc::channel::<i32>(0);
}
#[test]
fn send_recv_unbounded() {
let mut t1 = task::spawn(());
#[tokio::test]
async fn send_recv_unbounded() {
let (tx, mut rx) = mpsc::unbounded_channel::<i32>();
// Using `try_send`
assert_ok!(tx.send(1));
assert_ok!(tx.send(2));
let val = assert_ready!(t1.enter(|cx, _| rx.poll_recv(cx)));
assert_eq!(val, Some(1));
let val = assert_ready!(t1.enter(|cx, _| rx.poll_recv(cx)));
assert_eq!(val, Some(2));
assert_eq!(rx.recv().await, Some(1));
assert_eq!(rx.recv().await, Some(2));
drop(tx);
let val = assert_ready!(t1.enter(|cx, _| rx.poll_recv(cx)));
assert!(val.is_none());
assert!(rx.recv().await.is_none());
}
#[tokio::test]
@@ -201,11 +192,10 @@ async fn send_recv_stream_unbounded() {
assert_eq!(None, rx.next().await);
}
#[test]
fn no_t_bounds_buffer() {
#[tokio::test]
async fn no_t_bounds_buffer() {
struct NoImpls;
let mut t1 = task::spawn(());
let (tx, mut rx) = mpsc::channel(100);
// sender should be Debug even though T isn't Debug
@@ -215,15 +205,13 @@ fn no_t_bounds_buffer() {
// and sender should be Clone even though T isn't Clone
assert!(tx.clone().try_send(NoImpls).is_ok());
let val = assert_ready!(t1.enter(|cx, _| rx.poll_recv(cx)));
assert!(val.is_some());
assert!(rx.recv().await.is_some());
}
#[test]
fn no_t_bounds_unbounded() {
#[tokio::test]
async fn no_t_bounds_unbounded() {
struct NoImpls;
let mut t1 = task::spawn(());
let (tx, mut rx) = mpsc::unbounded_channel();
// sender should be Debug even though T isn't Debug
@@ -233,133 +221,87 @@ fn no_t_bounds_unbounded() {
// and sender should be Clone even though T isn't Clone
assert!(tx.clone().send(NoImpls).is_ok());
let val = assert_ready!(t1.enter(|cx, _| rx.poll_recv(cx)));
assert!(val.is_some());
assert!(rx.recv().await.is_some());
}
#[test]
fn send_recv_buffer_limited() {
let mut t1 = task::spawn(());
let mut t2 = task::spawn(());
#[tokio::test]
async fn send_recv_buffer_limited() {
let (tx, mut rx) = mpsc::channel::<i32>(1);
let (mut tx, mut rx) = mpsc::channel::<i32>(1);
// Reserve capacity
let p1 = assert_ok!(tx.reserve().await);
// Run on a task context
t1.enter(|cx, _| {
assert_ready_ok!(tx.poll_ready(cx));
// Send first message
p1.send(1);
// Send first message
assert_ok!(tx.try_send(1));
// Not ready
let mut p2 = task::spawn(tx.reserve());
assert_pending!(p2.poll());
// Not ready
assert_pending!(tx.poll_ready(cx));
// Take the value
assert!(rx.recv().await.is_some());
// Send second message
assert_err!(tx.try_send(1337));
});
// Notified
assert!(p2.is_woken());
t2.enter(|cx, _| {
// Take the value
let val = assert_ready!(rx.poll_recv(cx));
assert_eq!(Some(1), val);
});
// Trying to send fails
assert_err!(tx.try_send(1337));
assert!(t1.is_woken());
// Send second
let permit = assert_ready_ok!(p2.poll());
permit.send(2);
t1.enter(|cx, _| {
assert_ready_ok!(tx.poll_ready(cx));
assert_ok!(tx.try_send(2));
// Not ready
assert_pending!(tx.poll_ready(cx));
});
t2.enter(|cx, _| {
// Take the value
let val = assert_ready!(rx.poll_recv(cx));
assert_eq!(Some(2), val);
});
t1.enter(|cx, _| {
assert_ready_ok!(tx.poll_ready(cx));
});
assert!(rx.recv().await.is_some());
}
#[test]
fn recv_close_gets_none_idle() {
let mut t1 = task::spawn(());
let (mut tx, mut rx) = mpsc::channel::<i32>(10);
#[tokio::test]
async fn recv_close_gets_none_idle() {
let (tx, mut rx) = mpsc::channel::<i32>(10);
rx.close();
t1.enter(|cx, _| {
let val = assert_ready!(rx.poll_recv(cx));
assert!(val.is_none());
assert_ready_err!(tx.poll_ready(cx));
});
assert!(rx.recv().await.is_none());
assert_err!(tx.send(1).await);
}
#[test]
fn recv_close_gets_none_reserved() {
let mut t1 = task::spawn(());
let mut t2 = task::spawn(());
let mut t3 = task::spawn(());
#[tokio::test]
async fn recv_close_gets_none_reserved() {
let (tx1, mut rx) = mpsc::channel::<i32>(1);
let tx2 = tx1.clone();
let (mut tx1, mut rx) = mpsc::channel::<i32>(1);
let mut tx2 = tx1.clone();
assert_ready_ok!(t1.enter(|cx, _| tx1.poll_ready(cx)));
t2.enter(|cx, _| {
assert_pending!(tx2.poll_ready(cx));
});
let permit1 = assert_ok!(tx1.reserve().await);
let mut permit2 = task::spawn(tx2.reserve());
assert_pending!(permit2.poll());
rx.close();
assert!(t2.is_woken());
assert!(permit2.is_woken());
assert_ready_err!(permit2.poll());
t2.enter(|cx, _| {
assert_ready_err!(tx2.poll_ready(cx));
});
{
let mut recv = task::spawn(rx.recv());
assert_pending!(recv.poll());
t3.enter(|cx, _| assert_pending!(rx.poll_recv(cx)));
permit1.send(123);
assert!(recv.is_woken());
assert!(!t1.is_woken());
assert!(!t2.is_woken());
assert_ok!(tx1.try_send(123));
assert!(t3.is_woken());
t3.enter(|cx, _| {
let v = assert_ready!(rx.poll_recv(cx));
let v = assert_ready!(recv.poll());
assert_eq!(v, Some(123));
}
let v = assert_ready!(rx.poll_recv(cx));
assert!(v.is_none());
});
assert!(rx.recv().await.is_none());
}
#[test]
fn tx_close_gets_none() {
let mut t1 = task::spawn(());
#[tokio::test]
async fn tx_close_gets_none() {
let (_, mut rx) = mpsc::channel::<i32>(10);
// Run on a task context
t1.enter(|cx, _| {
let v = assert_ready!(rx.poll_recv(cx));
assert!(v.is_none());
});
assert!(rx.recv().await.is_none());
}
#[test]
fn try_send_fail() {
let mut t1 = task::spawn(());
let (mut tx, mut rx) = mpsc::channel(1);
#[tokio::test]
async fn try_send_fail() {
let (tx, mut rx) = mpsc::channel(1);
tx.try_send("hello").unwrap();
@@ -369,60 +311,48 @@ fn try_send_fail() {
_ => panic!(),
}
let val = assert_ready!(t1.enter(|cx, _| rx.poll_recv(cx)));
assert_eq!(val, Some("hello"));
assert_eq!(rx.recv().await, Some("hello"));
assert_ok!(tx.try_send("goodbye"));
drop(tx);
let val = assert_ready!(t1.enter(|cx, _| rx.poll_recv(cx)));
assert_eq!(val, Some("goodbye"));
let val = assert_ready!(t1.enter(|cx, _| rx.poll_recv(cx)));
assert!(val.is_none());
assert_eq!(rx.recv().await, Some("goodbye"));
assert!(rx.recv().await.is_none());
}
#[test]
fn drop_tx_with_permit_releases_permit() {
let mut t1 = task::spawn(());
let mut t2 = task::spawn(());
#[tokio::test]
async fn drop_permit_releases_permit() {
// poll_ready reserves capacity, ensure that the capacity is released if tx
// is dropped w/o sending a value.
let (mut tx1, _rx) = mpsc::channel::<i32>(1);
let mut tx2 = tx1.clone();
let (tx1, _rx) = mpsc::channel::<i32>(1);
let tx2 = tx1.clone();
assert_ready_ok!(t1.enter(|cx, _| tx1.poll_ready(cx)));
let permit = assert_ok!(tx1.reserve().await);
t2.enter(|cx, _| {
assert_pending!(tx2.poll_ready(cx));
});
let mut reserve2 = task::spawn(tx2.reserve());
assert_pending!(reserve2.poll());
drop(tx1);
drop(permit);
assert!(t2.is_woken());
assert_ready_ok!(t2.enter(|cx, _| tx2.poll_ready(cx)));
assert!(reserve2.is_woken());
assert_ready_ok!(reserve2.poll());
}
#[test]
fn dropping_rx_closes_channel() {
let mut t1 = task::spawn(());
let (mut tx, rx) = mpsc::channel(100);
#[tokio::test]
async fn dropping_rx_closes_channel() {
let (tx, rx) = mpsc::channel(100);
let msg = Arc::new(());
assert_ok!(tx.try_send(msg.clone()));
drop(rx);
assert_ready_err!(t1.enter(|cx, _| tx.poll_ready(cx)));
assert_err!(tx.reserve().await);
assert_eq!(1, Arc::strong_count(&msg));
}
#[test]
fn dropping_rx_closes_channel_for_try() {
let (mut tx, rx) = mpsc::channel(100);
let (tx, rx) = mpsc::channel(100);
let msg = Arc::new(());
tx.try_send(msg.clone()).unwrap();
@@ -444,7 +374,7 @@ fn dropping_rx_closes_channel_for_try() {
fn unconsumed_messages_are_dropped() {
let msg = Arc::new(());
let (mut tx, rx) = mpsc::channel(100);
let (tx, rx) = mpsc::channel(100);
tx.try_send(msg.clone()).unwrap();
@@ -457,7 +387,7 @@ fn unconsumed_messages_are_dropped() {
#[test]
fn try_recv() {
let (mut tx, mut rx) = mpsc::channel(1);
let (tx, mut rx) = mpsc::channel(1);
match rx.try_recv() {
Err(TryRecvError::Empty) => {}
_ => panic!(),
@@ -495,7 +425,7 @@ fn try_recv_unbounded() {
#[test]
fn blocking_recv() {
let (mut tx, mut rx) = mpsc::channel::<u8>(1);
let (tx, mut rx) = mpsc::channel::<u8>(1);
let sync_code = thread::spawn(move || {
assert_eq!(Some(10), rx.blocking_recv());
@@ -516,7 +446,7 @@ async fn blocking_recv_async() {
#[test]
fn blocking_send() {
let (mut tx, mut rx) = mpsc::channel::<u8>(1);
let (tx, mut rx) = mpsc::channel::<u8>(1);
let sync_code = thread::spawn(move || {
tx.blocking_send(10).unwrap();
@@ -531,28 +461,25 @@ fn blocking_send() {
#[tokio::test]
#[should_panic]
async fn blocking_send_async() {
let (mut tx, _rx) = mpsc::channel::<()>(1);
let (tx, _rx) = mpsc::channel::<()>(1);
let _ = tx.blocking_send(());
}
#[test]
fn ready_close_cancel_bounded() {
use futures::future::poll_fn;
let (mut tx, mut rx) = mpsc::channel::<()>(100);
#[tokio::test]
async fn ready_close_cancel_bounded() {
let (tx, mut rx) = mpsc::channel::<()>(100);
let _tx2 = tx.clone();
{
let mut ready = task::spawn(async { poll_fn(|cx| tx.poll_ready(cx)).await });
assert_ready_ok!(ready.poll());
}
let permit = assert_ok!(tx.reserve().await);
rx.close();
let mut recv = task::spawn(async { rx.recv().await });
let mut recv = task::spawn(rx.recv());
assert_pending!(recv.poll());
drop(tx);
drop(permit);
assert!(recv.is_woken());
let val = assert_ready!(recv.poll());
assert!(val.is_none());
}