sync: move into tokio crate (#1705)

A step towards collapsing Tokio sub crates into a single `tokio`
crate (#1318).

The sync implementation is now provided by the main `tokio` crate.
Functionality can be opted out of by using the various net related
feature flags.
This commit is contained in:
Carl Lerche
2019-10-29 15:11:31 -07:00
committed by GitHub
parent c62ef2d232
commit 2b909d6805
58 changed files with 574 additions and 1195 deletions
+5 -4
View File
@@ -63,7 +63,7 @@ signal = [
"net-driver",
"signal-hook-registry"
]
sync = ["tokio-sync"]
sync = ["fnv"]
tcp = ["io", "net-driver"]
timer = ["crossbeam-utils", "slab"]
udp = ["io", "net-driver"]
@@ -81,6 +81,8 @@ process = [
]
[dependencies]
tokio-macros = { version = "=0.2.0-alpha.6", optional = true, path = "../tokio-macros" }
futures-core-preview = "=0.3.0-alpha.19"
futures-sink-preview = "=0.3.0-alpha.19"
futures-util-preview = { version = "=0.3.0-alpha.19", features = ["sink", "channel"] }
@@ -89,6 +91,7 @@ futures-util-preview = { version = "=0.3.0-alpha.19", features = ["sink", "chann
bytes = { version = "0.4", optional = true }
crossbeam-channel = { version = "0.3.8", optional = true }
crossbeam-utils = { version = "0.6.0", optional = true }
fnv = { version = "1.0.6", optional = true }
iovec = { version = "0.1", optional = true }
lazy_static = { version = "1.0.2", optional = true }
memchr = { version = "2.2", optional = true }
@@ -97,8 +100,6 @@ num_cpus = { version = "1.8.0", optional = true }
pin-project = { version = "0.4", optional = true }
# Backs `DelayQueue`
slab = { version = "0.4.1", optional = true }
tokio-macros = { version = "=0.2.0-alpha.6", optional = true, path = "../tokio-macros" }
tokio-sync = { version = "=0.2.0-alpha.6", optional = true, path = "../tokio-sync", features = ["async-traits"] }
[target.'cfg(unix)'.dependencies]
crossbeam-queue = { version = "0.1.2", optional = true }
@@ -124,7 +125,7 @@ flate2 = { version = "1", features = ["tokio"] }
http = "0.1"
httparse = "1.0"
libc = "0.2"
loom = { version = "0.2.11", features = ["futures", "checkpoint"] }
loom = { version = "0.2.12", features = ["futures", "checkpoint"] }
num_cpus = "1.0"
rand = "0.7.2"
serde = { version = "1.0", features = ["derive"] }
+270
View File
@@ -0,0 +1,270 @@
#![feature(test)]
#![warn(rust_2018_idioms)]
extern crate test;
use tokio::sync::mpsc::*;
use futures::{future, Async, Future, Sink, Stream};
use std::thread;
use test::Bencher;
type Medium = [usize; 64];
type Large = [Medium; 64];
#[bench]
fn bounded_new_medium(b: &mut Bencher) {
b.iter(|| {
let _ = test::black_box(&channel::<Medium>(1_000));
})
}
#[bench]
fn unbounded_new_medium(b: &mut Bencher) {
b.iter(|| {
let _ = test::black_box(&unbounded_channel::<Medium>());
})
}
#[bench]
fn bounded_new_large(b: &mut Bencher) {
b.iter(|| {
let _ = test::black_box(&channel::<Large>(1_000));
})
}
#[bench]
fn unbounded_new_large(b: &mut Bencher) {
b.iter(|| {
let _ = test::black_box(&unbounded_channel::<Large>());
})
}
#[bench]
fn send_one_message(b: &mut Bencher) {
b.iter(|| {
let (mut tx, mut rx) = channel(1_000);
// Send
tx.try_send(1).unwrap();
// Receive
assert_eq!(Async::Ready(Some(1)), rx.poll().unwrap());
})
}
#[bench]
fn send_one_message_large(b: &mut Bencher) {
b.iter(|| {
let (mut tx, mut rx) = channel::<Large>(1_000);
// Send
let _ = tx.try_send([[0; 64]; 64]);
// Receive
let _ = test::black_box(&rx.poll());
})
}
#[bench]
fn bounded_rx_not_ready(b: &mut Bencher) {
let (_tx, mut rx) = channel::<i32>(1_000);
b.iter(|| {
future::lazy(|| {
assert!(rx.poll().unwrap().is_not_ready());
Ok::<_, ()>(())
})
.wait()
.unwrap();
})
}
#[bench]
fn bounded_tx_poll_ready(b: &mut Bencher) {
let (mut tx, _rx) = channel::<i32>(1);
b.iter(|| {
future::lazy(|| {
assert!(tx.poll_ready().unwrap().is_ready());
Ok::<_, ()>(())
})
.wait()
.unwrap();
})
}
#[bench]
fn bounded_tx_poll_not_ready(b: &mut Bencher) {
let (mut tx, _rx) = channel::<i32>(1);
tx.try_send(1).unwrap();
b.iter(|| {
future::lazy(|| {
assert!(tx.poll_ready().unwrap().is_not_ready());
Ok::<_, ()>(())
})
.wait()
.unwrap();
})
}
#[bench]
fn unbounded_rx_not_ready(b: &mut Bencher) {
let (_tx, mut rx) = unbounded_channel::<i32>();
b.iter(|| {
future::lazy(|| {
assert!(rx.poll().unwrap().is_not_ready());
Ok::<_, ()>(())
})
.wait()
.unwrap();
})
}
#[bench]
fn unbounded_rx_not_ready_x5(b: &mut Bencher) {
let (_tx, mut rx) = unbounded_channel::<i32>();
b.iter(|| {
future::lazy(|| {
assert!(rx.poll().unwrap().is_not_ready());
assert!(rx.poll().unwrap().is_not_ready());
assert!(rx.poll().unwrap().is_not_ready());
assert!(rx.poll().unwrap().is_not_ready());
assert!(rx.poll().unwrap().is_not_ready());
Ok::<_, ()>(())
})
.wait()
.unwrap();
})
}
#[bench]
fn bounded_uncontended_1(b: &mut Bencher) {
b.iter(|| {
let (mut tx, mut rx) = channel(1_000);
for i in 0..1000 {
tx.try_send(i).unwrap();
// No need to create a task, because poll is not going to park.
assert_eq!(Async::Ready(Some(i)), rx.poll().unwrap());
}
})
}
#[bench]
fn bounded_uncontended_1_large(b: &mut Bencher) {
b.iter(|| {
let (mut tx, mut rx) = channel::<Large>(1_000);
for i in 0..1000 {
let _ = tx.try_send([[i; 64]; 64]);
// No need to create a task, because poll is not going to park.
let _ = test::black_box(&rx.poll());
}
})
}
#[bench]
fn bounded_uncontended_2(b: &mut Bencher) {
b.iter(|| {
let (mut tx, mut rx) = channel(1000);
for i in 0..1000 {
tx.try_send(i).unwrap();
}
for i in 0..1000 {
// No need to create a task, because poll is not going to park.
assert_eq!(Async::Ready(Some(i)), rx.poll().unwrap());
}
})
}
#[bench]
fn contended_unbounded_tx(b: &mut Bencher) {
let mut threads = vec![];
let mut txs = vec![];
for _ in 0..4 {
let (tx, rx) = ::std::sync::mpsc::channel::<Sender<i32>>();
txs.push(tx);
threads.push(thread::spawn(move || {
for mut tx in rx.iter() {
for i in 0..1_000 {
tx.try_send(i).unwrap();
}
}
}));
}
b.iter(|| {
// TODO make unbounded
let (tx, rx) = channel::<i32>(1_000_000);
for th in &txs {
th.send(tx.clone()).unwrap();
}
drop(tx);
let rx = rx.wait().take(4 * 1_000);
for v in rx {
let _ = test::black_box(v);
}
});
drop(txs);
for th in threads {
th.join().unwrap();
}
}
#[bench]
fn contended_bounded_tx(b: &mut Bencher) {
const THREADS: usize = 4;
const ITERS: usize = 100;
let mut threads = vec![];
let mut txs = vec![];
for _ in 0..THREADS {
let (tx, rx) = ::std::sync::mpsc::channel::<Sender<i32>>();
txs.push(tx);
threads.push(thread::spawn(move || {
for tx in rx.iter() {
let mut tx = tx.wait();
for i in 0..ITERS {
tx.send(i as i32).unwrap();
}
}
}));
}
b.iter(|| {
let (tx, rx) = channel::<i32>(1);
for th in &txs {
th.send(tx.clone()).unwrap();
}
drop(tx);
let rx = rx.wait().take(THREADS * ITERS);
for v in rx {
let _ = test::black_box(v);
}
});
drop(txs);
for th in threads {
th.join().unwrap();
}
}
+120
View File
@@ -0,0 +1,120 @@
#![feature(test)]
#![warn(rust_2018_idioms)]
extern crate test;
use tokio::sync::oneshot;
use futures::{future, Async, Future};
use test::Bencher;
#[bench]
fn new(b: &mut Bencher) {
b.iter(|| {
let _ = ::test::black_box(&oneshot::channel::<i32>());
})
}
#[bench]
fn same_thread_send_recv(b: &mut Bencher) {
b.iter(|| {
let (tx, mut rx) = oneshot::channel();
let _ = tx.send(1);
assert_eq!(Async::Ready(1), rx.poll().unwrap());
});
}
#[bench]
fn same_thread_recv_multi_send_recv(b: &mut Bencher) {
b.iter(|| {
let (tx, mut rx) = oneshot::channel();
future::lazy(|| {
let _ = rx.poll();
let _ = rx.poll();
let _ = rx.poll();
let _ = rx.poll();
let _ = tx.send(1);
assert_eq!(Async::Ready(1), rx.poll().unwrap());
Ok::<_, ()>(())
})
.wait()
.unwrap();
});
}
#[bench]
fn multi_thread_send_recv(b: &mut Bencher) {
const MAX: usize = 10_000_000;
use std::thread;
fn spin<F: Future>(mut f: F) -> Result<F::Item, F::Error> {
use futures::Async::Ready;
loop {
match f.poll() {
Ok(Ready(v)) => return Ok(v),
Ok(_) => {}
Err(e) => return Err(e),
}
}
}
let mut ping_txs = vec![];
let mut ping_rxs = vec![];
let mut pong_txs = vec![];
let mut pong_rxs = vec![];
for _ in 0..MAX {
let (tx, rx) = oneshot::channel::<()>();
ping_txs.push(Some(tx));
ping_rxs.push(Some(rx));
let (tx, rx) = oneshot::channel::<()>();
pong_txs.push(Some(tx));
pong_rxs.push(Some(rx));
}
thread::spawn(move || {
future::lazy(|| {
for i in 0..MAX {
let ping_rx = ping_rxs[i].take().unwrap();
let pong_tx = pong_txs[i].take().unwrap();
if spin(ping_rx).is_err() {
return Ok(());
}
pong_tx.send(()).unwrap();
}
Ok::<(), ()>(())
})
.wait()
.unwrap();
});
future::lazy(|| {
let mut i = 0;
b.iter(|| {
let ping_tx = ping_txs[i].take().unwrap();
let pong_rx = pong_rxs[i].take().unwrap();
ping_tx.send(()).unwrap();
spin(pong_rx).unwrap();
i += 1;
});
Ok::<(), ()>(())
})
.wait()
.unwrap();
}
+1 -1
View File
@@ -3,7 +3,7 @@
extern crate test;
use tokio::executor::thread_pool::{Builder, Spawner};
use tokio_sync::oneshot;
use tokio::sync::oneshot;
use std::future::Future;
use std::pin::Pin;
+1 -1
View File
@@ -3,7 +3,7 @@
use crate::executor::loom::sync::{Arc, Condvar, Mutex};
use crate::executor::loom::thread;
#[cfg(feature = "blocking")]
use tokio_sync::oneshot;
use crate::sync::oneshot;
use std::cell::Cell;
use std::collections::VecDeque;
+1 -2
View File
@@ -4,8 +4,7 @@
//! dropped, the `Receiver` receives a notification.
use crate::executor::loom::sync::Arc;
use tokio_sync::oneshot;
use crate::sync::oneshot;
#[derive(Debug, Clone)]
pub(super) struct Sender {
@@ -3,8 +3,7 @@ use crate::loom::{
atomic::{AtomicUsize, Ordering},
CausalCell,
};
use tokio_sync::AtomicWaker;
use crate::sync::AtomicWaker;
#[derive(Debug)]
pub(crate) struct ScheduledIo {
-2
View File
@@ -1,5 +1,3 @@
#![cfg(feature = "async-traits")]
use super::{UnixListener, UnixStream};
use futures_core::ready;
-1
View File
@@ -90,7 +90,6 @@ impl UnixListener {
///
/// This method returns an implementation of the `Stream` trait which
/// resolves to the sockets the are accepted on this listener.
#[cfg(feature = "async-traits")]
pub fn incoming(self) -> super::Incoming {
super::Incoming::new(self)
}
-1
View File
@@ -6,7 +6,6 @@ mod datagram;
pub use self::datagram::UnixDatagram;
mod incoming;
#[cfg(feature = "async-traits")]
pub use self::incoming::Incoming;
mod listener;
+1 -1
View File
@@ -1,6 +1,6 @@
use crate::signal::os::{OsExtraData, OsStorage};
use tokio_sync::mpsc::Sender;
use crate::sync::mpsc::Sender;
use lazy_static::lazy_static;
use std::ops;
+1 -2
View File
@@ -8,8 +8,7 @@
use crate::io::AsyncRead;
use crate::net::util::PollEvented;
use crate::signal::registry::{globals, EventId, EventInfo, Globals, Init, Storage};
use tokio_sync::mpsc::{channel, Receiver};
use crate::sync::mpsc::{channel, Receiver};
use futures_core::stream::Stream;
use libc::c_int;
+2 -3
View File
@@ -7,9 +7,8 @@
#![cfg(windows)]
use super::registry::{globals, EventId, EventInfo, Init, Storage};
use tokio_sync::mpsc::{channel, Receiver};
use crate::signal::registry::{globals, EventId, EventInfo, Init, Storage};
use crate::sync::mpsc::{channel, Receiver};
use futures_core::stream::Stream;
use std::convert::TryFrom;
+135
View File
@@ -0,0 +1,135 @@
use crate::sync::watch;
use std::sync::Mutex;
/// A barrier enables multiple threads to synchronize the beginning of some computation.
///
/// ```
/// # #[tokio::main]
/// # async fn main() {
/// use tokio::sync::Barrier;
/// use std::sync::Arc;
/// use futures_util::future::join_all;
///
/// let mut handles = Vec::with_capacity(10);
/// let barrier = Arc::new(Barrier::new(10));
/// for _ in 0..10 {
/// let c = barrier.clone();
/// // The same messages will be printed together.
/// // You will NOT see any interleaving.
/// handles.push(async move {
/// println!("before wait");
/// let wr = c.wait().await;
/// println!("after wait");
/// wr
/// });
/// }
/// // Will not resolve until all "before wait" messages have been printed
/// let wrs = join_all(handles).await;
/// // Exactly one barrier will resolve as the "leader"
/// assert_eq!(wrs.into_iter().filter(|wr| wr.is_leader()).count(), 1);
/// # }
/// ```
#[derive(Debug)]
pub struct Barrier {
state: Mutex<BarrierState>,
wait: watch::Receiver<usize>,
n: usize,
}
#[derive(Debug)]
struct BarrierState {
waker: watch::Sender<usize>,
arrived: usize,
generation: usize,
}
impl Barrier {
/// Creates a new barrier that can block a given number of threads.
///
/// A barrier will block `n`-1 threads which call [`Barrier::wait`] and then wake up all
/// threads at once when the `n`th thread calls `wait`.
pub fn new(mut n: usize) -> Barrier {
let (waker, wait) = crate::sync::watch::channel(0);
if n == 0 {
// if n is 0, it's not clear what behavior the user wants.
// in std::sync::Barrier, an n of 0 exhibits the same behavior as n == 1, where every
// .wait() immediately unblocks, so we adopt that here as well.
n = 1;
}
Barrier {
state: Mutex::new(BarrierState {
waker,
arrived: 0,
generation: 1,
}),
n,
wait,
}
}
/// Does not resolve until all tasks have rendezvoused here.
///
/// Barriers are re-usable after all threads have rendezvoused once, and can
/// be used continuously.
///
/// A single (arbitrary) future will receive a [`BarrierWaitResult`] that returns `true` from
/// [`BarrierWaitResult::is_leader`] when returning from this function, and all other threads
/// will receive a result that will return `false` from `is_leader`.
pub async fn wait(&self) -> BarrierWaitResult {
// NOTE: we are taking a _synchronous_ lock here.
// It is okay to do so because the critical section is fast and never yields, so it cannot
// deadlock even if another future is concurrently holding the lock.
// It is _desireable_ to do so as synchronous Mutexes are, at least in theory, faster than
// the asynchronous counter-parts, so we should use them where possible [citation needed].
// NOTE: the extra scope here is so that the compiler doesn't think `state` is held across
// a yield point, and thus marks the returned future as !Send.
let generation = {
let mut state = self.state.lock().unwrap();
let generation = state.generation;
state.arrived += 1;
if state.arrived == self.n {
// we are the leader for this generation
// wake everyone, increment the generation, and return
state
.waker
.broadcast(state.generation)
.expect("there is at least one receiver");
state.arrived = 0;
state.generation += 1;
return BarrierWaitResult(true);
}
generation
};
// we're going to have to wait for the last of the generation to arrive
let mut wait = self.wait.clone();
loop {
// note that the first time through the loop, this _will_ yield a generation
// immediately, since we cloned a receiver that has never seen any values.
if wait.recv().await.expect("sender hasn't been closed") >= generation {
break;
}
}
BarrierWaitResult(false)
}
}
/// A `BarrierWaitResult` is returned by `wait` when all threads in the `Barrier` have rendezvoused.
#[derive(Debug, Clone)]
pub struct BarrierWaitResult(bool);
impl BarrierWaitResult {
/// Returns true if this thread from wait is the "leader thread".
///
/// Only one thread will have `true` returned from their result, all other threads will have
/// `false` returned.
pub fn is_leader(&self) -> bool {
self.0
}
}
+48
View File
@@ -0,0 +1,48 @@
#[cfg(not(all(test, loom)))]
mod imp {
pub(crate) mod future {
pub(crate) use crate::sync::task::AtomicWaker;
}
pub(crate) mod sync {
pub(crate) use std::sync::atomic;
pub(crate) use std::sync::Arc;
use std::cell::UnsafeCell;
pub(crate) struct CausalCell<T>(UnsafeCell<T>);
impl<T> CausalCell<T> {
pub(crate) fn new(data: T) -> CausalCell<T> {
CausalCell(UnsafeCell::new(data))
}
pub(crate) fn with<F, R>(&self, f: F) -> R
where
F: FnOnce(*const T) -> R,
{
f(self.0.get())
}
pub(crate) fn with_mut<F, R>(&self, f: F) -> R
where
F: FnOnce(*mut T) -> R,
{
f(self.0.get())
}
}
}
pub(crate) mod thread {
pub(crate) fn yield_now() {
::std::sync::atomic::spin_loop_hint();
}
}
}
#[cfg(all(test, loom))]
mod imp {
pub(crate) use loom::*;
}
pub(crate) use self::imp::*;
+43 -3
View File
@@ -13,6 +13,46 @@
//! - [watch](watch/index.html), a single-producer, multi-consumer channel that
//! only stores the **most recently** sent value.
pub use tokio_sync::Barrier;
pub use tokio_sync::{mpsc, oneshot, watch};
pub use tokio_sync::{Mutex, MutexGuard};
macro_rules! debug {
($($t:tt)*) => {
if false {
println!($($t)*);
}
}
}
macro_rules! if_loom {
($($t:tt)*) => {{
#[cfg(loom)]
const LOOM: bool = true;
#[cfg(not(loom))]
const LOOM: bool = false;
if LOOM {
$($t)*
}
}}
}
mod barrier;
pub use barrier::{Barrier, BarrierWaitResult};
mod loom;
pub mod mpsc;
mod mutex;
pub use mutex::{Mutex, MutexGuard};
pub mod oneshot;
pub mod semaphore;
mod task;
pub use task::AtomicWaker;
pub mod watch;
/// Unit tests
#[cfg(test)]
mod tests;
+387
View File
@@ -0,0 +1,387 @@
use crate::sync::loom::{
sync::atomic::{AtomicPtr, AtomicUsize},
sync::CausalCell,
thread,
};
use std::mem::MaybeUninit;
use std::ops;
use std::ptr::{self, NonNull};
use std::sync::atomic::Ordering::{self, AcqRel, Acquire, Release};
/// A block in a linked list.
///
/// Each block in the list can hold up to `BLOCK_CAP` messages.
pub(crate) struct Block<T> {
/// The start index of this block.
///
/// Slots in this block have indices in `start_index .. start_index + BLOCK_CAP`.
start_index: usize,
/// The next block in the linked list.
next: AtomicPtr<Block<T>>,
/// Bitfield tracking slots that are ready to have their values consumed.
ready_slots: AtomicUsize,
/// The observed `tail_position` value *after* the block has been passed by
/// `block_tail`.
observed_tail_position: CausalCell<usize>,
/// Array containing values pushed into the block. Values are stored in a
/// continuous array in order to improve cache line behavior when reading.
/// The values must be manually dropped.
values: Values<T>,
}
pub(crate) enum Read<T> {
Value(T),
Closed,
}
struct Values<T>([CausalCell<MaybeUninit<T>>; BLOCK_CAP]);
use super::BLOCK_CAP;
/// Masks an index to get the block identifier
const BLOCK_MASK: usize = !(BLOCK_CAP - 1);
/// Masks an index to get the value offset in a block.
const SLOT_MASK: usize = BLOCK_CAP - 1;
/// Flag tracking that a block has gone through the sender's release routine.
///
/// When this is set, the receiver may consider freeing the block.
const RELEASED: usize = 1 << BLOCK_CAP;
/// Flag tracking all senders dropped.
///
/// When this flag is set, the send half of the channel has closed.
const TX_CLOSED: usize = RELEASED << 1;
/// Mask covering all bits used to track slot readiness.
const READY_MASK: usize = RELEASED - 1;
/// Returns the index of the first slot in the block referenced by `slot_index`.
#[inline(always)]
pub(crate) fn start_index(slot_index: usize) -> usize {
BLOCK_MASK & slot_index
}
/// Returns the offset into the block referenced by `slot_index`.
#[inline(always)]
pub(crate) fn offset(slot_index: usize) -> usize {
SLOT_MASK & slot_index
}
impl<T> Block<T> {
pub(crate) fn new(start_index: usize) -> Block<T> {
Block {
// The absolute index in the channel of the first slot in the block.
start_index,
// Pointer to the next block in the linked list.
next: AtomicPtr::new(ptr::null_mut()),
ready_slots: AtomicUsize::new(0),
observed_tail_position: CausalCell::new(0),
// Value storage
values: unsafe { Values::uninitialized() },
}
}
/// Returns `true` if the block matches the given index
pub(crate) fn is_at_index(&self, index: usize) -> bool {
debug_assert!(offset(index) == 0);
self.start_index == index
}
/// Returns the number of blocks between `self` and the block at the
/// specified index.
///
/// `start_index` must represent a block *after* `self`.
pub(crate) fn distance(&self, other_index: usize) -> usize {
debug_assert!(offset(other_index) == 0);
other_index.wrapping_sub(self.start_index) / BLOCK_CAP
}
/// Read the value at the given offset.
///
/// Returns `None` if the slot is empty.
///
/// # Safety
///
/// To maintain safety, the caller must ensure:
///
/// * No concurrent access to the slot.
pub(crate) unsafe fn read(&self, slot_index: usize) -> Option<Read<T>> {
let offset = offset(slot_index);
let ready_bits = self.ready_slots.load(Acquire);
if !is_ready(ready_bits, offset) {
if is_tx_closed(ready_bits) {
return Some(Read::Closed);
}
return None;
}
// Get the value
let value = self.values[offset].with(|ptr| ptr::read(ptr));
Some(Read::Value(value.assume_init()))
}
/// Write a value to the block at the given offset.
///
/// # Safety
///
/// To maintain safety, the caller must ensure:
///
/// * The slot is empty.
/// * No concurrent access to the slot.
pub(crate) unsafe fn write(&self, slot_index: usize, value: T) {
// Get the offset into the block
let slot_offset = offset(slot_index);
self.values[slot_offset].with_mut(|ptr| {
ptr::write(ptr, MaybeUninit::new(value));
});
// Release the value. After this point, the slot ref may no longer
// be used. It is possible for the receiver to free the memory at
// any point.
self.set_ready(slot_offset);
}
/// Signal to the receiver that the sender half of the list is closed.
pub(crate) unsafe fn tx_close(&self) {
self.ready_slots.fetch_or(TX_CLOSED, Release);
}
/// Reset the block to a blank state. This enables reusing blocks in the
/// channel.
///
/// # Safety
///
/// To maintain safety, the caller must ensure:
///
/// * All slots are empty.
/// * The caller holds a unique pointer to the block.
pub(crate) unsafe fn reclaim(&mut self) {
self.start_index = 0;
self.next = AtomicPtr::new(ptr::null_mut());
self.ready_slots = AtomicUsize::new(0);
}
/// Release the block to the rx half for freeing.
///
/// This function is called by the tx half once it can be guaranteed that no
/// more senders will attempt to access the block.
///
/// # Safety
///
/// To maintain safety, the caller must ensure:
///
/// * The block will no longer be accessed by any sender.
pub(crate) unsafe fn tx_release(&self, tail_position: usize) {
// Track the observed tail_position. Any sender targetting a greater
// tail_position is guaranteed to not access this block.
self.observed_tail_position
.with_mut(|ptr| *ptr = tail_position);
// Set the released bit, signalling to the receiver that it is safe to
// free the block's memory as soon as all slots **prior** to
// `observed_tail_position` have been filled.
self.ready_slots.fetch_or(RELEASED, Release);
}
/// Mark a slot as ready
fn set_ready(&self, slot: usize) {
let mask = 1 << slot;
self.ready_slots.fetch_or(mask, Release);
}
/// Returns `true` when all slots have their `ready` bits set.
///
/// This indicates that the block is in its final state and will no longer
/// be mutated.
///
/// # Implementation
///
/// The implementation walks each slot checking the `ready` flag. It might
/// be that it would make more sense to coalesce ready flags as bits in a
/// single atomic cell. However, this could have negative impact on cache
/// behavior as there would be many more mutations to a single slot.
pub(crate) fn is_final(&self) -> bool {
self.ready_slots.load(Acquire) & READY_MASK == READY_MASK
}
/// Returns the `observed_tail_position` value, if set
pub(crate) fn observed_tail_position(&self) -> Option<usize> {
if 0 == RELEASED & self.ready_slots.load(Acquire) {
None
} else {
Some(self.observed_tail_position.with(|ptr| unsafe { *ptr }))
}
}
/// Load the next block
pub(crate) fn load_next(&self, ordering: Ordering) -> Option<NonNull<Block<T>>> {
let ret = NonNull::new(self.next.load(ordering));
debug_assert!(unsafe {
ret.map(|block| block.as_ref().start_index == self.start_index.wrapping_add(BLOCK_CAP))
.unwrap_or(true)
});
ret
}
/// Push `block` as the next block in the link.
///
/// Returns Ok if successful, otherwise, a pointer to the next block in
/// the list is returned.
///
/// This requires that the next pointer is null.
///
/// # Ordering
///
/// This performs a compare-and-swap on `next` using AcqRel ordering.
///
/// # Safety
///
/// To maintain safety, the caller must ensure:
///
/// * `block` is not freed until it has been removed from the list.
pub(crate) unsafe fn try_push(
&self,
block: &mut NonNull<Block<T>>,
ordering: Ordering,
) -> Result<(), NonNull<Block<T>>> {
block.as_mut().start_index = self.start_index.wrapping_add(BLOCK_CAP);
let next_ptr = self
.next
.compare_and_swap(ptr::null_mut(), block.as_ptr(), ordering);
match NonNull::new(next_ptr) {
Some(next_ptr) => Err(next_ptr),
None => Ok(()),
}
}
/// Grow the `Block` linked list by allocating and appending a new block.
///
/// The next block in the linked list is returned. This may or may not be
/// the one allocated by the function call.
///
/// # Implementation
///
/// It is assumed that `self.next` is null. A new block is allocated with
/// `start_index` set to be the next block. A compare-and-swap is performed
/// with AcqRel memory ordering. If the compare-and-swap is successful, the
/// newly allocated block is released to other threads walking the block
/// linked list. If the compare-and-swap fails, the current thread acquires
/// the next block in the linked list, allowing the current thread to access
/// the slots.
pub(crate) fn grow(&self) -> NonNull<Block<T>> {
// Create the new block. It is assumed that the block will become the
// next one after `&self`. If this turns out to not be the case,
// `start_index` is updated accordingly.
let new_block = Box::new(Block::new(self.start_index + BLOCK_CAP));
let mut new_block = unsafe { NonNull::new_unchecked(Box::into_raw(new_block)) };
// Attempt to store the block. The first compare-and-swap attempt is
// "unrolled" due to minor differences in logic
//
// `AcqRel` is used as the ordering **only** when attempting the
// compare-and-swap on self.next.
//
// If the compare-and-swap fails, then the actual value of the cell is
// returned from this function and accessed by the caller. Given this,
// the memory must be acquired.
//
// `Release` ensures that the newly allocated block is available to
// other threads acquiring the next pointer.
let next = NonNull::new(self.next.compare_and_swap(
ptr::null_mut(),
new_block.as_ptr(),
AcqRel,
));
let next = match next {
Some(next) => next,
None => {
// The compare-and-swap succeeded and the newly allocated block
// is successfully pushed.
return new_block;
}
};
// There already is a next block in the linked list. The newly allocated
// block could be dropped and the discovered next block returned;
// however, that would be wasteful. Instead, the linked list is walked
// by repeatedly attempting to compare-and-swap the pointer into the
// `next` register until the compare-and-swap succeed.
//
// Care is taken to update new_block's start_index field as appropriate.
let mut curr = next;
// TODO: Should this iteration be capped?
loop {
let actual = unsafe { curr.as_ref().try_push(&mut new_block, AcqRel) };
curr = match actual {
Ok(_) => {
return next;
}
Err(curr) => curr,
};
// When running outside of loom, this calls `spin_loop_hint`.
thread::yield_now();
}
}
}
/// Returns `true` if the specificed slot has a value ready to be consumed.
fn is_ready(bits: usize, slot: usize) -> bool {
let mask = 1 << slot;
mask == mask & bits
}
/// Returns `true` if the closed flag has been set.
fn is_tx_closed(bits: usize) -> bool {
TX_CLOSED == bits & TX_CLOSED
}
impl<T> Values<T> {
unsafe fn uninitialized() -> Values<T> {
let mut vals = MaybeUninit::uninit();
// When fuzzing, `CausalCell` needs to be initialized.
if_loom! {
let p = vals.as_mut_ptr() as *mut CausalCell<MaybeUninit<T>>;
for i in 0..BLOCK_CAP {
p.add(i)
.write(CausalCell::new(MaybeUninit::uninit()));
}
}
Values(vals.assume_init())
}
}
impl<T> ops::Index<usize> for Values<T> {
type Output = CausalCell<MaybeUninit<T>>;
fn index(&self, index: usize) -> &Self::Output {
self.0.index(index)
}
}
+337
View File
@@ -0,0 +1,337 @@
use crate::sync::mpsc::chan;
use crate::sync::semaphore;
use std::fmt;
use std::pin::Pin;
use std::task::{Context, Poll};
/// Send values to the associated `Receiver`.
///
/// Instances are created by the [`channel`](fn.channel.html) function.
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()
}
}
/// Receive values from the associated `Sender`.
///
/// Instances are created by the [`channel`](fn.channel.html) function.
pub struct Receiver<T> {
/// The channel receiver
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()
}
}
/// Error returned by the `Sender`.
#[derive(Debug)]
pub struct SendError(());
/// Error returned by `Sender::try_send`.
#[derive(Debug)]
pub struct TrySendError<T> {
kind: ErrorKind,
value: T,
}
#[derive(Debug)]
enum ErrorKind {
Closed,
NoCapacity,
}
/// Error returned by `Receiver`.
#[derive(Debug)]
pub struct RecvError(());
/// Create a bounded mpsc channel for communicating between asynchronous tasks,
/// returning the sender/receiver halves.
///
/// All data sent on `Sender` will become available on `Receiver` in the same
/// order as it was sent.
///
/// The `Sender` can be cloned to `send` to the same channel from multiple code
/// locations. Only one `Receiver` is supported.
///
/// If the `Receiver` is disconnected while trying to `send`, the `send` method
/// will return a `SendError`. Similarly, if `Sender` is disconnected while
/// trying to `recv`, the `recv` method will return a `RecvError`.
///
/// # Examples
///
/// ```rust
/// use tokio::sync::mpsc;
///
/// #[tokio::main]
/// async fn main() {
/// let (mut tx, mut rx) = mpsc::channel(100);
///
/// tokio::spawn(async move {
/// for i in 0..10 {
/// if let Err(_) = tx.send(i).await {
/// println!("receiver dropped");
/// return;
/// }
/// }
/// });
///
/// while let Some(i) = rx.recv().await {
/// println!("got = {}", i);
/// }
/// }
/// ```
pub fn channel<T>(buffer: usize) -> (Sender<T>, Receiver<T>) {
assert!(buffer > 0, "mpsc bounded channel requires buffer > 0");
let semaphore = (semaphore::Semaphore::new(buffer), buffer);
let (tx, rx) = chan::channel(semaphore);
let tx = Sender::new(tx);
let rx = Receiver::new(rx);
(tx, rx)
}
/// Channel semaphore is a tuple of the semaphore implementation and a `usize`
/// representing the channel bound.
type Semaphore = (semaphore::Semaphore, usize);
impl<T> Receiver<T> {
pub(crate) fn new(chan: chan::Rx<T, Semaphore>) -> Receiver<T> {
Receiver { chan }
}
/// Receive the next value for this receiver.
///
/// `None` is returned when all `Sender` halves have dropped, indicating
/// that no further values can be sent on the channel.
///
/// # Examples
///
/// ```
/// use tokio::sync::mpsc;
///
/// #[tokio::main]
/// async fn main() {
/// let (mut tx, mut rx) = mpsc::channel(100);
///
/// tokio::spawn(async move {
/// tx.send("hello").await.unwrap();
/// });
///
/// assert_eq!(Some("hello"), rx.recv().await);
/// assert_eq!(None, rx.recv().await);
/// }
/// ```
///
/// Values are buffered:
///
/// ```
/// use tokio::sync::mpsc;
///
/// #[tokio::main]
/// async fn main() {
/// let (mut tx, mut rx) = mpsc::channel(100);
///
/// tx.send("hello").await.unwrap();
/// tx.send("world").await.unwrap();
///
/// assert_eq!(Some("hello"), rx.recv().await);
/// assert_eq!(Some("world"), rx.recv().await);
/// }
/// ```
pub async fn recv(&mut self) -> Option<T> {
use futures_util::future::poll_fn;
poll_fn(|cx| self.poll_recv(cx)).await
}
#[doc(hidden)] // TODO: remove
pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<T>> {
self.chan.recv(cx)
}
/// 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.
pub fn close(&mut self) {
self.chan.close();
}
}
impl<T> futures_core::Stream for Receiver<T> {
type Item = T;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
self.get_mut().poll_recv(cx)
}
}
impl<T> Sender<T> {
pub(crate) fn new(chan: chan::Tx<T, Semaphore>) -> Sender<T> {
Sender { chan }
}
#[doc(hidden)] // TODO: remove
pub fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), SendError>> {
self.chan.poll_ready(cx).map_err(|_| SendError(()))
}
/// Attempts to send a message on this `Sender`, returning the message
/// if there was an error.
pub fn try_send(&mut self, message: T) -> Result<(), TrySendError<T>> {
self.chan.try_send(message)?;
Ok(())
}
/// Send a value, waiting until there is capacity.
///
/// # Examples
///
/// In the following example, each call to `send` will block until the
/// previously sent value was received.
///
/// ```rust
/// use tokio::sync::mpsc;
///
/// #[tokio::main]
/// async fn main() {
/// let (mut tx, mut rx) = mpsc::channel(1);
///
/// tokio::spawn(async move {
/// for i in 0..10 {
/// if let Err(_) = tx.send(i).await {
/// println!("receiver dropped");
/// return;
/// }
/// }
/// });
///
/// while let Some(i) = rx.recv().await {
/// println!("got = {}", i);
/// }
/// }
/// ```
pub async fn send(&mut self, value: T) -> Result<(), SendError> {
use futures_util::future::poll_fn;
poll_fn(|cx| self.poll_ready(cx)).await?;
self.try_send(value).map_err(|_| SendError(()))
}
}
impl<T> futures_sink::Sink<T> for Sender<T> {
type Error = SendError;
fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Sender::poll_ready(self.get_mut(), cx)
}
fn start_send(mut self: Pin<&mut Self>, msg: T) -> Result<(), Self::Error> {
self.as_mut().try_send(msg).map_err(|err| {
assert!(err.is_full(), "call `poll_ready` before sending");
SendError(())
})
}
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
}
// ===== impl SendError =====
impl fmt::Display for SendError {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(fmt, "channel closed")
}
}
impl ::std::error::Error for SendError {}
// ===== impl TrySendError =====
impl<T> TrySendError<T> {
/// Get the inner value.
pub fn into_inner(self) -> T {
self.value
}
/// Did the send fail because the channel has been closed?
pub fn is_closed(&self) -> bool {
if let ErrorKind::Closed = self.kind {
true
} else {
false
}
}
/// Did the send fail because the channel was at capacity?
pub fn is_full(&self) -> bool {
if let ErrorKind::NoCapacity = self.kind {
true
} else {
false
}
}
}
impl<T: fmt::Debug> fmt::Display for TrySendError<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
let descr = match self.kind {
ErrorKind::Closed => "channel closed",
ErrorKind::NoCapacity => "no available capacity",
};
write!(fmt, "{}", descr)
}
}
impl<T: fmt::Debug> ::std::error::Error for TrySendError<T> {}
impl<T> From<(T, chan::TrySendError)> for TrySendError<T> {
fn from((value, err): (T, chan::TrySendError)) -> TrySendError<T> {
TrySendError {
value,
kind: match err {
chan::TrySendError::Closed => ErrorKind::Closed,
chan::TrySendError::NoPermits => ErrorKind::NoCapacity,
},
}
}
}
// ===== impl RecvError =====
impl fmt::Display for RecvError {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(fmt, "channel closed")
}
}
impl ::std::error::Error for RecvError {}
+451
View File
@@ -0,0 +1,451 @@
use crate::sync::loom::{
future::AtomicWaker,
sync::atomic::AtomicUsize,
sync::{Arc, CausalCell},
};
use crate::sync::mpsc::list;
use std::fmt;
use std::process;
use std::sync::atomic::Ordering::{AcqRel, Relaxed};
use std::task::Poll::{Pending, Ready};
use std::task::{Context, Poll};
/// Channel sender
pub(crate) struct Tx<T, S: Semaphore> {
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,
{
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Tx")
.field("inner", &self.inner)
.field("permit", &self.permit)
.finish()
}
}
/// Channel receiver
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,
{
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,
NoPermits,
}
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.
fn drop_permit(&self, permit: &mut Self::Permit);
fn is_idle(&self) -> bool;
fn add_permit(&self);
fn poll_acquire(&self, cx: &mut Context<'_>, permit: &mut Self::Permit)
-> Poll<Result<(), ()>>;
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);
}
struct Chan<T, S> {
/// Handle to the push half of the lock-free list.
tx: list::Tx<T>,
/// Coordinates access to channel's capacity.
semaphore: S,
/// Receiver waker. Notified when a value is pushed into the channel.
rx_waker: AtomicWaker,
/// Tracks the number of outstanding sender handles.
///
/// When this drops to zero, the send half of the channel is closed.
tx_count: AtomicUsize,
/// Only accessed by `Rx` handle.
rx_fields: CausalCell<RxFields<T>>,
}
impl<T, S> fmt::Debug for Chan<T, S>
where
S: fmt::Debug,
{
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Chan")
.field("tx", &self.tx)
.field("semaphore", &self.semaphore)
.field("rx_waker", &self.rx_waker)
.field("tx_count", &self.tx_count)
.field("rx_fields", &"...")
.finish()
}
}
/// Fields only accessed by `Rx` handle.
struct RxFields<T> {
/// Channel receiver. This field is only accessed by the `Receiver` type.
list: list::Rx<T>,
/// `true` if `Rx::close` is called.
rx_closed: bool,
}
impl<T> fmt::Debug for RxFields<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("RxFields")
.field("list", &self.list)
.field("rx_closed", &self.rx_closed)
.finish()
}
}
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,
{
let (tx, rx) = list::channel();
let chan = Arc::new(Chan {
tx,
semaphore,
rx_waker: AtomicWaker::new(),
tx_count: AtomicUsize::new(1),
rx_fields: CausalCell::new(RxFields {
list: rx,
rx_closed: false,
}),
});
(Tx::new(chan.clone()), Rx::new(chan))
}
// ===== impl Tx =====
impl<T, S> Tx<T, S>
where
S: Semaphore,
{
fn new(chan: Arc<Chan<T, S>>) -> Tx<T, S> {
Tx {
inner: chan,
permit: S::new_permit(),
}
}
pub(crate) fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), ()>> {
self.inner.semaphore.poll_acquire(cx, &mut self.permit)
}
/// Send a message and notify the receiver.
pub(crate) fn try_send(&mut self, value: T) -> Result<(), (T, TrySendError)> {
if let Err(e) = self.inner.semaphore.try_acquire(&mut self.permit) {
return Err((value, e));
}
// Push the value
self.inner.tx.push(value);
// Notify the rx task
self.inner.rx_waker.wake();
// Release the permit
self.inner.semaphore.forget(&mut self.permit);
Ok(())
}
}
impl<T, S> Clone for Tx<T, S>
where
S: Semaphore,
{
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.
self.inner.tx_count.fetch_add(1, Relaxed);
Tx {
inner: self.inner.clone(),
permit: S::new_permit(),
}
}
}
impl<T, S> Drop for Tx<T, S>
where
S: Semaphore,
{
fn drop(&mut self) {
self.inner.semaphore.drop_permit(&mut self.permit);
if self.inner.tx_count.fetch_sub(1, AcqRel) != 1 {
return;
}
// Close the list, which sends a `Close` message
self.inner.tx.close();
// Notify the receiver
self.inner.rx_waker.wake();
}
}
// ===== impl Rx =====
impl<T, S> Rx<T, S>
where
S: Semaphore,
{
fn new(chan: Arc<Chan<T, S>>) -> Rx<T, S> {
Rx { inner: chan }
}
pub(crate) fn close(&mut self) {
self.inner.rx_fields.with_mut(|rx_fields_ptr| {
let rx_fields = unsafe { &mut *rx_fields_ptr };
if rx_fields.rx_closed {
return;
}
rx_fields.rx_closed = true;
});
self.inner.semaphore.close();
}
/// Receive the next value
pub(crate) fn recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<T>> {
use super::block::Read::*;
self.inner.rx_fields.with_mut(|rx_fields_ptr| {
let rx_fields = unsafe { &mut *rx_fields_ptr };
macro_rules! try_recv {
() => {
match rx_fields.list.pop(&self.inner.tx) {
Some(Value(value)) => {
self.inner.semaphore.add_permit();
return Ready(Some(value));
}
Some(Closed) => {
// TODO: This check may not be required as it most
// likely can only return `true` at this point. A
// channel is closed when all tx handles are
// dropped. Dropping a tx handle releases memory,
// which ensures that if dropping the tx handle is
// visible, then all messages sent are also visible.
assert!(self.inner.semaphore.is_idle());
return Ready(None);
}
None => {} // fall through
}
};
}
try_recv!();
self.inner.rx_waker.register_by_ref(cx.waker());
// It is possible that a value was pushed between attempting to read
// and registering the task, so we have to check the channel a
// second time here.
try_recv!();
debug!(
"recv; rx_closed = {:?}; is_idle = {:?}",
rx_fields.rx_closed,
self.inner.semaphore.is_idle()
);
if rx_fields.rx_closed && self.inner.semaphore.is_idle() {
Ready(None)
} else {
Pending
}
})
}
}
impl<T, S> Drop for Rx<T, S>
where
S: Semaphore,
{
fn drop(&mut self) {
use super::block::Read::Value;
self.close();
self.inner.rx_fields.with_mut(|rx_fields_ptr| {
let rx_fields = unsafe { &mut *rx_fields_ptr };
while let Some(Value(_)) = rx_fields.list.pop(&self.inner.tx) {
self.inner.semaphore.add_permit();
}
})
}
}
// ===== impl Chan =====
impl<T, S> Drop for Chan<T, S> {
fn drop(&mut self) {
use super::block::Read::Value;
// Safety: the only owner of the rx fields is Chan, and eing
// inside its own Drop means we're the last ones to touch it.
self.rx_fields.with_mut(|rx_fields_ptr| {
let rx_fields = unsafe { &mut *rx_fields_ptr };
while let Some(Value(_)) = rx_fields.list.pop(&self.tx) {}
unsafe { rx_fields.list.free_blocks() };
});
}
}
use crate::sync::semaphore::TryAcquireError;
impl From<TryAcquireError> for TrySendError {
fn from(src: TryAcquireError) -> TrySendError {
if src.is_closed() {
TrySendError::Closed
} else if src.is_no_permits() {
TrySendError::NoPermits
} else {
unreachable!();
}
}
}
// ===== impl Semaphore for (::Semaphore, capacity) =====
use crate::sync::semaphore::Permit;
impl Semaphore for (crate::sync::semaphore::Semaphore, usize) {
type Permit = Permit;
fn new_permit() -> Permit {
Permit::new()
}
fn drop_permit(&self, permit: &mut Permit) {
permit.release(&self.0);
}
fn add_permit(&self) {
self.0.add_permits(1)
}
fn is_idle(&self) -> bool {
self.0.available_permits() == self.1
}
fn poll_acquire(&self, cx: &mut Context<'_>, permit: &mut Permit) -> Poll<Result<(), ()>> {
permit.poll_acquire(cx, &self.0).map_err(|_| ())
}
fn try_acquire(&self, permit: &mut Permit) -> Result<(), TrySendError> {
permit.try_acquire(&self.0)?;
Ok(())
}
fn forget(&self, permit: &mut Self::Permit) {
permit.forget()
}
fn close(&self) {
self.0.close();
}
}
// ===== impl Semaphore for AtomicUsize =====
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 ()) {}
fn add_permit(&self) {
let prev = self.fetch_sub(2, Release);
if prev >> 1 == 0 {
// Something went wrong
process::abort();
}
}
fn is_idle(&self) -> bool {
self.load(Acquire) >> 1 == 0
}
fn poll_acquire(&self, _cx: &mut Context<'_>, permit: &mut ()) -> Poll<Result<(), ()>> {
Ready(self.try_acquire(permit).map_err(|_| ()))
}
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);
}
}
+348
View File
@@ -0,0 +1,348 @@
//! A concurrent, lock-free, FIFO list.
use crate::sync::loom::{
sync::atomic::{AtomicPtr, AtomicUsize},
thread,
};
use crate::sync::mpsc::block::{self, Block};
use std::fmt;
use std::ptr::NonNull;
use std::sync::atomic::Ordering::{AcqRel, Acquire, Relaxed, Release};
/// List queue transmit handle
pub(crate) struct Tx<T> {
/// Tail in the `Block` mpmc list.
block_tail: AtomicPtr<Block<T>>,
/// Position to push the next message. This reference a block and offset
/// into the block.
tail_position: AtomicUsize,
}
/// List queue receive handle
pub(crate) struct Rx<T> {
/// Pointer to the block being processed
head: NonNull<Block<T>>,
/// Next slot index to process
index: usize,
/// Pointer to the next block pending release
free_head: NonNull<Block<T>>,
}
pub(crate) fn channel<T>() -> (Tx<T>, Rx<T>) {
// Create the initial block shared between the tx and rx halves.
let initial_block = Box::new(Block::new(0));
let initial_block_ptr = Box::into_raw(initial_block);
let tx = Tx {
block_tail: AtomicPtr::new(initial_block_ptr),
tail_position: AtomicUsize::new(0),
};
let head = NonNull::new(initial_block_ptr).unwrap();
let rx = Rx {
head,
index: 0,
free_head: head,
};
(tx, rx)
}
impl<T> Tx<T> {
/// Push a value into the list.
pub(crate) fn push(&self, value: T) {
// First, claim a slot for the value. `Acquire` is used here to
// synchronize with the `fetch_add` in `reclaim_blocks`.
let slot_index = self.tail_position.fetch_add(1, Acquire);
// Load the current block and write the value
let block = self.find_block(slot_index);
unsafe {
// Write the value to the block
block.as_ref().write(slot_index, value);
}
}
/// Close the send half of the list
///
/// Similar process as pushing a value, but instead of writing the value &
/// setting the ready flag, the TX_CLOSED flag is set on the block.
pub(crate) fn close(&self) {
// First, claim a slot for the value. This is the last slot that will be
// claimed.
let slot_index = self.tail_position.fetch_add(1, Acquire);
let block = self.find_block(slot_index);
unsafe { block.as_ref().tx_close() }
}
fn find_block(&self, slot_index: usize) -> NonNull<Block<T>> {
// The start index of the block that contains `index`.
let start_index = block::start_index(slot_index);
// The index offset into the block
let offset = block::offset(slot_index);
// Load the current head of the block
let mut block_ptr = self.block_tail.load(Acquire);
let block = unsafe { &*block_ptr };
// Calculate the distance between the tail ptr and the target block
let distance = block.distance(start_index);
// Decide if this call to `find_block` should attempt to update the
// `block_tail` pointer.
//
// Updating `block_tail` is not always performed in order to reduce
// contention.
//
// When set, as the routine walks the linked list, it attempts to update
// `block_tail`. If the update cannot be performed, `try_updating_tail`
// is unset.
let mut try_updating_tail = distance > offset;
// Walk the linked list of blocks until the block with `start_index` is
// found.
loop {
let block = unsafe { &(*block_ptr) };
if block.is_at_index(start_index) {
return unsafe { NonNull::new_unchecked(block_ptr) };
}
let next_block = block
.load_next(Acquire)
// There is no allocated next block, grow the linked list.
.unwrap_or_else(|| block.grow());
// If the block is **not** final, then the tail pointer cannot be
// advanced any more.
try_updating_tail &= block.is_final();
if try_updating_tail {
// Advancing `block_tail` must happen when walking the linked
// list. `block_tail` may not advance passed any blocks that are
// not "final". At the point a block is finalized, it is unknown
// if there are any prior blocks that are unfinalized, which
// makes it impossible to advance `block_tail`.
//
// While walking the linked list, `block_tail` can be advanced
// as long as finalized blocks are traversed.
//
// Release ordering is used to ensure that any subsequent reads
// are able to see the memory pointed to by `block_tail`.
//
// Acquire is not needed as any "actual" value is not accessed.
// At this point, the linked list is walked to acquire blocks.
let actual =
self.block_tail
.compare_and_swap(block_ptr, next_block.as_ptr(), Release);
if actual == block_ptr {
// Synchronize with any senders
let tail_position = self.tail_position.fetch_add(0, Release);
unsafe {
block.tx_release(tail_position);
}
} else {
// A concurrent sender is also working on advancing
// `block_tail` and this thread is falling behind.
//
// Stop trying to advance the tail pointer
try_updating_tail = false;
}
}
block_ptr = next_block.as_ptr();
thread::yield_now();
}
}
pub(crate) unsafe fn reclaim_block(&self, mut block: NonNull<Block<T>>) {
debug!("+ reclaim_block({:p})", block);
// The block has been removed from the linked list and ownership
// is reclaimed.
//
// Before dropping the block, see if it can be reused by
// inserting it back at the end of the linked list.
//
// First, reset the data
block.as_mut().reclaim();
let mut reused = false;
// Attempt to insert the block at the end
//
// Walk at most three times
//
let curr_ptr = self.block_tail.load(Acquire);
// The pointer can never be null
debug_assert!(!curr_ptr.is_null());
let mut curr = NonNull::new_unchecked(curr_ptr);
// TODO: Unify this logic with Block::grow
for _ in 0..3 {
match curr.as_ref().try_push(&mut block, AcqRel) {
Ok(_) => {
reused = true;
break;
}
Err(next) => {
curr = next;
}
}
}
if !reused {
debug!(" + block freed {:p}", block);
let _ = Box::from_raw(block.as_ptr());
}
}
}
impl<T> fmt::Debug for Tx<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Tx")
.field("block_tail", &self.block_tail.load(Relaxed))
.field("tail_position", &self.tail_position.load(Relaxed))
.finish()
}
}
impl<T> Rx<T> {
/// Pop the next value off the queue
pub(crate) fn pop(&mut self, tx: &Tx<T>) -> Option<block::Read<T>> {
// Advance `head`, if needed
if !self.try_advancing_head() {
debug!("+ !self.try_advancing_head() -> false");
return None;
}
self.reclaim_blocks(tx);
unsafe {
let block = self.head.as_ref();
let ret = block.read(self.index);
if let Some(block::Read::Value(..)) = ret {
self.index = self.index.wrapping_add(1);
}
ret
}
}
/// Try advancing the block pointer to the block referenced by `self.index`.
///
/// Returns `true` if successful, `false` if there is no next block to load.
fn try_advancing_head(&mut self) -> bool {
let block_index = block::start_index(self.index);
loop {
let next_block = {
let block = unsafe { self.head.as_ref() };
if block.is_at_index(block_index) {
return true;
}
block.load_next(Acquire)
};
let next_block = match next_block {
Some(next_block) => next_block,
None => {
return false;
}
};
self.head = next_block;
thread::yield_now();
}
}
fn reclaim_blocks(&mut self, tx: &Tx<T>) {
debug!("+ reclaim_blocks()");
while self.free_head != self.head {
unsafe {
// Get a handle to the block that will be freed and update
// `free_head` to point to the next block.
let block = self.free_head;
let observed_tail_position = block.as_ref().observed_tail_position();
let required_index = match observed_tail_position {
Some(i) => i,
None => return,
};
if required_index > self.index {
return;
}
// We may read the next pointer with `Relaxed` ordering as it is
// guaranteed that the `reclaim_blocks` routine trails the `recv`
// routine. Any memory accessed by `reclaim_blocks` has already
// been acquired by `recv`.
let next_block = block.as_ref().load_next(Relaxed);
// Update the free list head
self.free_head = next_block.unwrap();
// Push the emptied block onto the back of the queue, making it
// available to senders.
tx.reclaim_block(block);
}
thread::yield_now();
}
}
/// Effectively `Drop` all the blocks. Should only be called once, when
/// the list is dropping.
pub(super) unsafe fn free_blocks(&mut self) {
debug!("+ free_blocks()");
debug_assert_ne!(self.free_head, NonNull::dangling());
let mut cur = Some(self.free_head);
#[cfg(debug_assertions)]
{
// to trigger the debug assert above so as to catch that we
// don't call `free_blocks` more than once.
self.free_head = NonNull::dangling();
self.head = NonNull::dangling();
}
while let Some(block) = cur {
cur = block.as_ref().load_next(Relaxed);
debug!(" + free: block = {:p}", block);
drop(Box::from_raw(block.as_ptr()));
}
}
}
impl<T> fmt::Debug for Rx<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Rx")
.field("head", &self.head)
.field("index", &self.index)
.field("free_head", &self.free_head)
.finish()
}
}
+67
View File
@@ -0,0 +1,67 @@
//! A multi-producer, single-consumer queue for sending values across
//! asynchronous tasks.
//!
//! Similar to `std`, channel creation provides [`Receiver`] and [`Sender`]
//! handles. [`Receiver`] implements `Stream` and allows a task to read values
//! out of the channel. If there is no message to read, the current task will be
//! notified when a new value is sent. [`Sender`] implements the `Sink` trait
//! and allows sending messages into the channel. If the channel is at capacity,
//! the send is rejected and the task will be notified when additional capacity
//! is available. In other words, the channel provides backpressure.
//!
//! Unbounded channels are also available using the `unbounded_channel`
//! constructor.
//!
//! # Disconnection
//!
//! When all [`Sender`] handles have been dropped, it is no longer
//! possible to send values into the channel. This is considered the termination
//! event of the stream. As such, `Receiver::poll` returns `Ok(Ready(None))`.
//!
//! If the [`Receiver`] handle is dropped, then messages can no longer
//! be read out of the channel. In this case, all further attempts to send will
//! result in an error.
//!
//! # Clean Shutdown
//!
//! When the [`Receiver`] is dropped, it is possible for unprocessed messages to
//! remain in the channel. Instead, it is usually desirable to perform a "clean"
//! shutdown. To do this, the receiver first calls `close`, which will prevent
//! any further messages to be sent into the channel. Then, the receiver
//! consumes the channel to completion, at which point the receiver can be
//! dropped.
//!
//! [`Sender`]: struct.Sender.html
//! [`Receiver`]: struct.Receiver.html
pub(super) mod block;
mod bounded;
pub use self::bounded::{channel, Receiver, Sender};
mod chan;
pub(super) mod list;
mod unbounded;
pub use self::unbounded::{unbounded_channel, UnboundedReceiver, UnboundedSender};
pub mod error {
//! Channel error types
pub use super::bounded::{RecvError, SendError, TrySendError};
pub use super::unbounded::{UnboundedRecvError, UnboundedSendError, UnboundedTrySendError};
}
/// The number of values a block can contain.
///
/// This value must be a power of 2. It also must be smaller than the number of
/// bits in `usize`.
#[cfg(all(target_pointer_width = "64", not(loom)))]
const BLOCK_CAP: usize = 32;
#[cfg(all(not(target_pointer_width = "64"), not(loom)))]
const BLOCK_CAP: usize = 16;
#[cfg(loom)]
const BLOCK_CAP: usize = 2;
+230
View File
@@ -0,0 +1,230 @@
use crate::sync::loom::sync::atomic::AtomicUsize;
use crate::sync::mpsc::chan;
use std::fmt;
use std::task::{Context, Poll};
use std::pin::Pin;
/// Send values to the associated `UnboundedReceiver`.
///
/// Instances are created by the
/// [`unbounded_channel`](fn.unbounded_channel.html) function.
pub struct UnboundedSender<T> {
chan: chan::Tx<T, Semaphore>,
}
impl<T> Clone for UnboundedSender<T> {
fn clone(&self) -> Self {
UnboundedSender {
chan: self.chan.clone(),
}
}
}
impl<T> fmt::Debug for UnboundedSender<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("UnboundedSender")
.field("chan", &self.chan)
.finish()
}
}
/// Receive values from the associated `UnboundedSender`.
///
/// Instances are created by the
/// [`unbounded_channel`](fn.unbounded_channel.html) function.
pub struct UnboundedReceiver<T> {
/// The channel receiver
chan: chan::Rx<T, Semaphore>,
}
impl<T> fmt::Debug for UnboundedReceiver<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("UnboundedReceiver")
.field("chan", &self.chan)
.finish()
}
}
/// Error returned by the `UnboundedSender`.
#[derive(Debug)]
pub struct UnboundedSendError(());
/// Returned by `UnboundedSender::try_send` when the channel has been closed.
#[derive(Debug)]
pub struct UnboundedTrySendError<T>(T);
/// Error returned by `UnboundedReceiver`.
#[derive(Debug)]
pub struct UnboundedRecvError(());
/// Create an unbounded mpsc channel for communicating between asynchronous
/// tasks.
///
/// A `send` on this channel will always succeed as long as the receive half has
/// not been closed. If the receiver falls behind, messages will be arbitrarily
/// buffered.
///
/// **Note** that the amount of available system memory is an implicit bound to
/// the channel. Using an `unbounded` channel has the ability of causing the
/// process to run out of memory. In this case, the process will be aborted.
pub fn unbounded_channel<T>() -> (UnboundedSender<T>, UnboundedReceiver<T>) {
let (tx, rx) = chan::channel(AtomicUsize::new(0));
let tx = UnboundedSender::new(tx);
let rx = UnboundedReceiver::new(rx);
(tx, rx)
}
/// No capacity
type Semaphore = AtomicUsize;
impl<T> UnboundedReceiver<T> {
pub(crate) fn new(chan: chan::Rx<T, Semaphore>) -> UnboundedReceiver<T> {
UnboundedReceiver { chan }
}
#[doc(hidden)] // TODO: remove
pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<T>> {
self.chan.recv(cx)
}
/// Receive the next value for this receiver.
///
/// `None` is returned when all `Sender` halves have dropped, indicating
/// that no further values can be sent on the channel.
///
/// # Examples
///
/// ```
/// use tokio::sync::mpsc;
///
/// #[tokio::main]
/// async fn main() {
/// let (mut tx, mut rx) = mpsc::unbounded_channel();
///
/// tokio::spawn(async move {
/// tx.try_send("hello").unwrap();
/// });
///
/// assert_eq!(Some("hello"), rx.recv().await);
/// assert_eq!(None, rx.recv().await);
/// }
/// ```
///
/// Values are buffered:
///
/// ```
/// use tokio::sync::mpsc;
///
/// #[tokio::main]
/// async fn main() {
/// let (mut tx, mut rx) = mpsc::unbounded_channel();
///
/// tx.try_send("hello").unwrap();
/// tx.try_send("world").unwrap();
///
/// assert_eq!(Some("hello"), rx.recv().await);
/// assert_eq!(Some("world"), rx.recv().await);
/// }
/// ```
pub async fn recv(&mut self) -> Option<T> {
use futures_util::future::poll_fn;
poll_fn(|cx| self.poll_recv(cx)).await
}
/// 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.
pub fn close(&mut self) {
self.chan.close();
}
}
impl<T> futures_core::Stream for UnboundedReceiver<T> {
type Item = T;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
self.chan.recv(cx)
}
}
impl<T> UnboundedSender<T> {
pub(crate) fn new(chan: chan::Tx<T, Semaphore>) -> UnboundedSender<T> {
UnboundedSender { chan }
}
/// Attempts to send a message on this `UnboundedSender` without blocking.
pub fn try_send(&mut self, message: T) -> Result<(), UnboundedTrySendError<T>> {
self.chan.try_send(message)?;
Ok(())
}
}
impl<T> futures_sink::Sink<T> for UnboundedSender<T> {
type Error = UnboundedSendError;
fn poll_ready(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn start_send(mut self: Pin<&mut Self>, msg: T) -> Result<(), Self::Error> {
self.try_send(msg).map_err(|_| UnboundedSendError(()))
}
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
}
// ===== impl UnboundedSendError =====
impl fmt::Display for UnboundedSendError {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(fmt, "channel closed")
}
}
impl ::std::error::Error for UnboundedSendError {}
// ===== impl TrySendError =====
impl<T> UnboundedTrySendError<T> {
/// Get the inner value.
pub fn into_inner(self) -> T {
self.0
}
}
impl<T: fmt::Debug> fmt::Display for UnboundedTrySendError<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(fmt, "channel closed")
}
}
impl<T: fmt::Debug> ::std::error::Error for UnboundedTrySendError<T> {}
impl<T> From<(T, chan::TrySendError)> for UnboundedTrySendError<T> {
fn from((value, err): (T, chan::TrySendError)) -> UnboundedTrySendError<T> {
assert_eq!(chan::TrySendError::Closed, err);
UnboundedTrySendError(value)
}
}
// ===== impl UnboundedRecvError =====
impl fmt::Display for UnboundedRecvError {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(fmt, "channel closed")
}
}
impl ::std::error::Error for UnboundedRecvError {}
+149
View File
@@ -0,0 +1,149 @@
//! An asynchronous `Mutex`-like type.
//!
//! This module provides [`Mutex`], a type that acts similarly to an asynchronous `Mutex`, with one
//! major difference: the [`MutexGuard`] returned by `lock` is not tied to the lifetime of the
//! `Mutex`. This enables you to acquire a lock, and then pass that guard into a future, and then
//! release it at some later point in time.
//!
//! This allows you to do something along the lines of:
//!
//! ```rust,no_run
//! use tokio::sync::Mutex;
//! use std::sync::Arc;
//!
//! #[tokio::main]
//! async fn main() {
//! let data1 = Arc::new(Mutex::new(0));
//! let data2 = Arc::clone(&data1);
//!
//! tokio::spawn(async move {
//! let mut lock = data2.lock().await;
//! *lock += 1;
//! });
//!
//! let mut lock = data1.lock().await;
//! *lock += 1;
//! }
//! ```
//!
//! [`Mutex`]: struct.Mutex.html
//! [`MutexGuard`]: struct.MutexGuard.html
use crate::sync::semaphore;
use futures_util::future::poll_fn;
use std::cell::UnsafeCell;
use std::fmt;
use std::ops::{Deref, DerefMut};
/// An asynchronous mutual exclusion primitive useful for protecting shared data
///
/// Each mutex has a type parameter (`T`) which represents the data that it is protecting. The data
/// can only be accessed through the RAII guards returned from `lock`, which
/// guarantees that the data is only ever accessed when the mutex is locked.
#[derive(Debug)]
pub struct Mutex<T> {
c: UnsafeCell<T>,
s: semaphore::Semaphore,
}
/// A handle to a held `Mutex`.
///
/// As long as you have this guard, you have exclusive access to the underlying `T`. The guard
/// internally keeps a reference-couned pointer to the original `Mutex`, so even if the lock goes
/// away, the guard remains valid.
///
/// The lock is automatically released whenever the guard is dropped, at which point `lock`
/// will succeed yet again.
#[derive(Debug)]
pub struct MutexGuard<'a, T> {
lock: &'a Mutex<T>,
permit: semaphore::Permit,
}
// As long as T: Send, it's fine to send and share Mutex<T> between threads.
// If T was not Send, sending and sharing a Mutex<T> would be bad, since you can access T through
// Mutex<T>.
unsafe impl<T> Send for Mutex<T> where T: Send {}
unsafe impl<T> Sync for Mutex<T> where T: Send {}
unsafe impl<'a, T> Sync for MutexGuard<'a, T> where T: Send + Sync {}
#[test]
#[cfg(not(loom))]
fn bounds() {
fn check<T: Send>() {}
check::<MutexGuard<'_, u32>>();
}
impl<T> Mutex<T> {
/// Creates a new lock in an unlocked state ready for use.
pub fn new(t: T) -> Self {
Self {
c: UnsafeCell::new(t),
s: semaphore::Semaphore::new(1),
}
}
/// A future that resolves on acquiring the lock and returns the `MutexGuard`.
pub async fn lock(&self) -> MutexGuard<'_, T> {
let mut permit = semaphore::Permit::new();
poll_fn(|cx| permit.poll_acquire(cx, &self.s))
.await
.unwrap_or_else(|_| {
// The semaphore was closed. but, we never explicitly close it, and we have a
// handle to it through the Arc, which means that this can never happen.
unreachable!()
});
MutexGuard { lock: self, permit }
}
}
impl<'a, T> Drop for MutexGuard<'a, T> {
fn drop(&mut self) {
if self.permit.is_acquired() {
self.permit.release(&self.lock.s);
} else if ::std::thread::panicking() {
// A guard _should_ always hold its permit, but if the thread is already panicking,
// we don't want to generate a panic-while-panicing, since that's just unhelpful!
} else {
unreachable!("Permit not held when MutexGuard was dropped")
}
}
}
impl<T> From<T> for Mutex<T> {
fn from(s: T) -> Self {
Self::new(s)
}
}
impl<T> Default for Mutex<T>
where
T: Default,
{
fn default() -> Self {
Self::new(T::default())
}
}
impl<'a, T> Deref for MutexGuard<'a, T> {
type Target = T;
fn deref(&self) -> &Self::Target {
assert!(self.permit.is_acquired());
unsafe { &*self.lock.c.get() }
}
}
impl<'a, T> DerefMut for MutexGuard<'a, T> {
fn deref_mut(&mut self) -> &mut Self::Target {
assert!(self.permit.is_acquired());
unsafe { &mut *self.lock.c.get() }
}
}
impl<'a, T: fmt::Display> fmt::Display for MutexGuard<'a, T> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt::Display::fmt(&**self, f)
}
}
+576
View File
@@ -0,0 +1,576 @@
//! A channel for sending a single message between asynchronous tasks.
use crate::sync::loom::sync::{atomic::AtomicUsize, Arc, CausalCell};
use futures_core::ready;
use std::fmt;
use std::future::Future;
use std::mem::MaybeUninit;
use std::pin::Pin;
use std::sync::atomic::Ordering::{self, AcqRel, Acquire};
use std::task::Poll::{Pending, Ready};
use std::task::{Context, Poll, Waker};
/// Sends a value to the associated `Receiver`.
///
/// Instances are created by the [`channel`](fn.channel.html) function.
#[derive(Debug)]
pub struct Sender<T> {
inner: Option<Arc<Inner<T>>>,
}
/// Receive a value from the associated `Sender`.
///
/// Instances are created by the [`channel`](fn.channel.html) function.
#[derive(Debug)]
pub struct Receiver<T> {
inner: Option<Arc<Inner<T>>>,
}
pub mod error {
//! Oneshot error types
use std::fmt;
/// Error returned by the `Future` implementation for `Receiver`.
#[derive(Debug)]
pub struct RecvError(pub(super) ());
/// Error returned by the `try_recv` function on `Receiver`.
#[derive(Debug)]
pub struct TryRecvError(pub(super) ());
// ===== impl RecvError =====
impl fmt::Display for RecvError {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(fmt, "channel closed")
}
}
impl ::std::error::Error for RecvError {}
// ===== impl TryRecvError =====
impl fmt::Display for TryRecvError {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(fmt, "channel closed")
}
}
impl ::std::error::Error for TryRecvError {}
}
use self::error::*;
struct Inner<T> {
/// Manages the state of the inner cell
state: AtomicUsize,
/// The value. This is set by `Sender` and read by `Receiver`. The state of
/// the cell is tracked by `state`.
value: CausalCell<Option<T>>,
/// The task to notify when the receiver drops without consuming the value.
tx_task: CausalCell<MaybeUninit<Waker>>,
/// The task to notify when the value is sent.
rx_task: CausalCell<MaybeUninit<Waker>>,
}
#[derive(Clone, Copy)]
struct State(usize);
/// Create a new one-shot channel for sending single values across asynchronous
/// tasks.
///
/// The function returns separate "send" and "receive" handles. The `Sender`
/// handle is used by the producer to send the value. The `Receiver` handle is
/// used by the consumer to receive the value.
///
/// Each handle can be used on separate tasks.
///
/// # Examples
///
/// ```
/// use tokio::sync::oneshot;
///
/// #[tokio::main]
/// async fn main() {
/// let (tx, rx) = oneshot::channel();
///
/// tokio::spawn(async move {
/// if let Err(_) = tx.send(3) {
/// println!("the receiver dropped");
/// }
/// });
///
/// match rx.await {
/// Ok(v) => println!("got = {:?}", v),
/// Err(_) => println!("the sender dropped"),
/// }
/// }
/// ```
pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
#[allow(deprecated)]
let inner = Arc::new(Inner {
state: AtomicUsize::new(State::new().as_usize()),
value: CausalCell::new(None),
tx_task: CausalCell::new(MaybeUninit::uninit()),
rx_task: CausalCell::new(MaybeUninit::uninit()),
});
let tx = Sender {
inner: Some(inner.clone()),
};
let rx = Receiver { inner: Some(inner) };
(tx, rx)
}
impl<T> Sender<T> {
/// Completes this oneshot with a successful result.
///
/// The function consumes `self` and notifies the `Receiver` handle that a
/// value is ready to be received.
///
/// If the value is successfully enqueued for the remote end to receive,
/// then `Ok(())` is returned. If the receiving end was dropped before this
/// function was called, however, then `Err` is returned with the value
/// provided.
pub fn send(mut self, t: T) -> Result<(), T> {
let inner = self.inner.take().unwrap();
inner.value.with_mut(|ptr| unsafe {
*ptr = Some(t);
});
if !inner.complete() {
return Err(inner
.value
.with_mut(|ptr| unsafe { (*ptr).take() }.unwrap()));
}
Ok(())
}
#[doc(hidden)] // TODO: remove
pub fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll<()> {
let inner = self.inner.as_ref().unwrap();
let mut state = State::load(&inner.state, Acquire);
if state.is_closed() {
return Poll::Ready(());
}
if state.is_tx_task_set() {
let will_notify = unsafe { inner.with_tx_task(|w| w.will_wake(cx.waker())) };
if !will_notify {
state = State::unset_tx_task(&inner.state);
if state.is_closed() {
// Set the flag again so that the waker is released in drop
State::set_tx_task(&inner.state);
return Ready(());
} else {
unsafe { inner.drop_tx_task() };
}
}
}
if !state.is_tx_task_set() {
// Attempt to set the task
unsafe {
inner.set_tx_task(cx);
}
// Update the state
state = State::set_tx_task(&inner.state);
if state.is_closed() {
return Ready(());
}
}
Pending
}
/// Wait for the associated [`Receiver`] handle to drop.
///
/// # Return
///
/// Returns a `Future` which must be awaited on.
///
/// [`Receiver`]: struct.Receiver.html
///
/// # Examples
///
/// ```
/// use tokio::sync::oneshot;
///
/// #[tokio::main]
/// async fn main() {
/// let (mut tx, rx) = oneshot::channel::<()>();
///
/// tokio::spawn(async move {
/// drop(rx);
/// });
///
/// tx.closed().await;
/// println!("the receiver dropped");
/// }
/// ```
pub async fn closed(&mut self) {
use futures_util::future::poll_fn;
poll_fn(|cx| self.poll_closed(cx)).await
}
/// Check if the associated [`Receiver`] handle has been dropped.
///
/// Unlike [`poll_closed`], this function does not register a task for
/// wakeup upon close.
///
/// [`Receiver`]: struct.Receiver.html
/// [`poll_closed`]: struct.Sender.html#method.poll_closed
pub fn is_closed(&self) -> bool {
let inner = self.inner.as_ref().unwrap();
let state = State::load(&inner.state, Acquire);
state.is_closed()
}
}
impl<T> Drop for Sender<T> {
fn drop(&mut self) {
if let Some(inner) = self.inner.as_ref() {
inner.complete();
}
}
}
impl<T> Receiver<T> {
/// Prevent the associated [`Sender`] handle from sending a value.
///
/// Any `send` operation which happens after calling `close` is guaranteed
/// to fail. After calling `close`, `Receiver::poll`] should be called to
/// receive a value if one was sent **before** the call to `close`
/// completed.
///
/// [`Sender`]: struct.Sender.html
pub fn close(&mut self) {
let inner = self.inner.as_ref().unwrap();
inner.close();
}
/// Attempts to receive a value outside of the context of a task.
///
/// Does not register a task if no value has been sent.
///
/// A return value of `None` must be considered immediately stale (out of
/// date) unless [`close`] has been called first.
///
/// Returns an error if the sender was dropped.
///
/// [`close`]: #method.close
pub fn try_recv(&mut self) -> Result<T, TryRecvError> {
let result = if let Some(inner) = self.inner.as_ref() {
let state = State::load(&inner.state, Acquire);
if state.is_complete() {
match unsafe { inner.consume_value() } {
Some(value) => Ok(value),
None => Err(TryRecvError(())),
}
} else if state.is_closed() {
Err(TryRecvError(()))
} else {
// Not ready, this does not clear `inner`
return Err(TryRecvError(()));
}
} else {
panic!("called after complete");
};
self.inner = None;
result
}
}
impl<T> Drop for Receiver<T> {
fn drop(&mut self) {
if let Some(inner) = self.inner.as_ref() {
inner.close();
}
}
}
impl<T> Future for Receiver<T> {
type Output = Result<T, RecvError>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
// If `inner` is `None`, then `poll()` has already completed.
let ret = if let Some(inner) = self.as_ref().get_ref().inner.as_ref() {
ready!(inner.poll_recv(cx))?
} else {
panic!("called after complete");
};
self.inner = None;
Ready(Ok(ret))
}
}
impl<T> Inner<T> {
fn complete(&self) -> bool {
let prev = State::set_complete(&self.state);
if prev.is_closed() {
return false;
}
if prev.is_rx_task_set() {
// TODO: Consume waker?
unsafe {
self.with_rx_task(Waker::wake_by_ref);
}
}
true
}
fn poll_recv(&self, cx: &mut Context<'_>) -> Poll<Result<T, RecvError>> {
// Load the state
let mut state = State::load(&self.state, Acquire);
if state.is_complete() {
match unsafe { self.consume_value() } {
Some(value) => Ready(Ok(value)),
None => Ready(Err(RecvError(()))),
}
} else if state.is_closed() {
Ready(Err(RecvError(())))
} else {
if state.is_rx_task_set() {
let will_notify = unsafe { self.with_rx_task(|w| w.will_wake(cx.waker())) };
// Check if the task is still the same
if !will_notify {
// Unset the task
state = State::unset_rx_task(&self.state);
if state.is_complete() {
// Set the flag again so that the waker is released in drop
State::set_rx_task(&self.state);
return match unsafe { self.consume_value() } {
Some(value) => Ready(Ok(value)),
None => Ready(Err(RecvError(()))),
};
} else {
unsafe { self.drop_rx_task() };
}
}
}
if !state.is_rx_task_set() {
// Attempt to set the task
unsafe {
self.set_rx_task(cx);
}
// Update the state
state = State::set_rx_task(&self.state);
if state.is_complete() {
match unsafe { self.consume_value() } {
Some(value) => Ready(Ok(value)),
None => Ready(Err(RecvError(()))),
}
} else {
Pending
}
} else {
Pending
}
}
}
/// Called by `Receiver` to indicate that the value will never be received.
fn close(&self) {
let prev = State::set_closed(&self.state);
if prev.is_tx_task_set() && !prev.is_complete() {
unsafe {
self.with_tx_task(Waker::wake_by_ref);
}
}
}
/// Consume the value. This function does not check `state`.
unsafe fn consume_value(&self) -> Option<T> {
self.value.with_mut(|ptr| (*ptr).take())
}
unsafe fn with_rx_task<F, R>(&self, f: F) -> R
where
F: FnOnce(&Waker) -> R,
{
self.rx_task.with(|ptr| {
let waker: *const Waker = (&*ptr).as_ptr();
f(&*waker)
})
}
unsafe fn with_tx_task<F, R>(&self, f: F) -> R
where
F: FnOnce(&Waker) -> R,
{
self.tx_task.with(|ptr| {
let waker: *const Waker = (&*ptr).as_ptr();
f(&*waker)
})
}
unsafe fn drop_rx_task(&self) {
self.rx_task.with_mut(|ptr| {
let ptr: *mut Waker = (&mut *ptr).as_mut_ptr();
ptr.drop_in_place();
});
}
unsafe fn drop_tx_task(&self) {
self.tx_task.with_mut(|ptr| {
let ptr: *mut Waker = (&mut *ptr).as_mut_ptr();
ptr.drop_in_place();
});
}
unsafe fn set_rx_task(&self, cx: &mut Context<'_>) {
self.rx_task.with_mut(|ptr| {
let ptr: *mut Waker = (&mut *ptr).as_mut_ptr();
ptr.write(cx.waker().clone());
});
}
unsafe fn set_tx_task(&self, cx: &mut Context<'_>) {
self.tx_task.with_mut(|ptr| {
let ptr: *mut Waker = (&mut *ptr).as_mut_ptr();
ptr.write(cx.waker().clone());
});
}
}
unsafe impl<T: Send> Send for Inner<T> {}
unsafe impl<T: Send> Sync for Inner<T> {}
impl<T> Drop for Inner<T> {
fn drop(&mut self) {
let state = State(*self.state.get_mut());
if state.is_rx_task_set() {
unsafe {
self.drop_rx_task();
}
}
if state.is_tx_task_set() {
unsafe {
self.drop_tx_task();
}
}
}
}
impl<T: fmt::Debug> fmt::Debug for Inner<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
use std::sync::atomic::Ordering::Relaxed;
fmt.debug_struct("Inner")
.field("state", &State::load(&self.state, Relaxed))
.finish()
}
}
const RX_TASK_SET: usize = 0b00001;
const VALUE_SENT: usize = 0b00010;
const CLOSED: usize = 0b00100;
const TX_TASK_SET: usize = 0b01000;
impl State {
fn new() -> State {
State(0)
}
fn is_complete(self) -> bool {
self.0 & VALUE_SENT == VALUE_SENT
}
fn set_complete(cell: &AtomicUsize) -> State {
// TODO: This could be `Release`, followed by an `Acquire` fence *if*
// the `RX_TASK_SET` flag is set. However, `loom` does not support
// fences yet.
let val = cell.fetch_or(VALUE_SENT, AcqRel);
State(val)
}
fn is_rx_task_set(self) -> bool {
self.0 & RX_TASK_SET == RX_TASK_SET
}
fn set_rx_task(cell: &AtomicUsize) -> State {
let val = cell.fetch_or(RX_TASK_SET, AcqRel);
State(val | RX_TASK_SET)
}
fn unset_rx_task(cell: &AtomicUsize) -> State {
let val = cell.fetch_and(!RX_TASK_SET, AcqRel);
State(val & !RX_TASK_SET)
}
fn is_closed(self) -> bool {
self.0 & CLOSED == CLOSED
}
fn set_closed(cell: &AtomicUsize) -> State {
// Acquire because we want all later writes (attempting to poll) to be
// ordered after this.
let val = cell.fetch_or(CLOSED, Acquire);
State(val)
}
fn set_tx_task(cell: &AtomicUsize) -> State {
let val = cell.fetch_or(TX_TASK_SET, AcqRel);
State(val | TX_TASK_SET)
}
fn unset_tx_task(cell: &AtomicUsize) -> State {
let val = cell.fetch_and(!TX_TASK_SET, AcqRel);
State(val & !TX_TASK_SET)
}
fn is_tx_task_set(self) -> bool {
self.0 & TX_TASK_SET == TX_TASK_SET
}
fn as_usize(self) -> usize {
self.0
}
fn load(cell: &AtomicUsize, order: Ordering) -> State {
let val = cell.load(order);
State(val)
}
}
impl fmt::Debug for State {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("State")
.field("is_complete", &self.is_complete())
.field("is_closed", &self.is_closed())
.field("is_rx_task_set", &self.is_rx_task_set())
.field("is_tx_task_set", &self.is_tx_task_set())
.finish()
}
}
File diff suppressed because it is too large Load Diff
+323
View File
@@ -0,0 +1,323 @@
use crate::sync::loom::sync::atomic::{self, AtomicUsize};
use crate::sync::loom::sync::CausalCell;
use std::fmt;
use std::sync::atomic::Ordering::{AcqRel, Acquire, Release};
use std::task::Waker;
/// A synchronization primitive for task waking.
///
/// `AtomicWaker` will coordinate concurrent wakes with the consumer
/// potentially "waking" the underlying task. This is useful in scenarios
/// where a computation completes in another thread and wants to wake the
/// consumer, but the consumer is in the process of being migrated to a new
/// logical task.
///
/// Consumers should call `register` before checking the result of a computation
/// and producers should call `wake` after producing the computation (this
/// differs from the usual `thread::park` pattern). It is also permitted for
/// `wake` to be called **before** `register`. This results in a no-op.
///
/// A single `AtomicWaker` may be reused for any number of calls to `register` or
/// `wake`.
pub struct AtomicWaker {
state: AtomicUsize,
waker: CausalCell<Option<Waker>>,
}
// `AtomicWaker` is a multi-consumer, single-producer transfer cell. The cell
// stores a `Waker` value produced by calls to `register` and many threads can
// race to take the waker by calling `wake.
//
// If a new `Waker` instance is produced by calling `register` before an existing
// one is consumed, then the existing one is overwritten.
//
// While `AtomicWaker` is single-producer, the implementation ensures memory
// safety. In the event of concurrent calls to `register`, there will be a
// single winner whose waker will get stored in the cell. The losers will not
// have their tasks woken. As such, callers should ensure to add synchronization
// to calls to `register`.
//
// The implementation uses a single `AtomicUsize` value to coordinate access to
// the `Waker` cell. There are two bits that are operated on independently. These
// are represented by `REGISTERING` and `WAKING`.
//
// The `REGISTERING` bit is set when a producer enters the critical section. The
// `WAKING` bit is set when a consumer enters the critical section. Neither
// bit being set is represented by `WAITING`.
//
// A thread obtains an exclusive lock on the waker cell by transitioning the
// state from `WAITING` to `REGISTERING` or `WAKING`, depending on the
// operation the thread wishes to perform. When this transition is made, it is
// guaranteed that no other thread will access the waker cell.
//
// # Registering
//
// On a call to `register`, an attempt to transition the state from WAITING to
// REGISTERING is made. On success, the caller obtains a lock on the waker cell.
//
// If the lock is obtained, then the thread sets the waker cell to the waker
// provided as an argument. Then it attempts to transition the state back from
// `REGISTERING` -> `WAITING`.
//
// If this transition is successful, then the registering process is complete
// and the next call to `wake` will observe the waker.
//
// If the transition fails, then there was a concurrent call to `wake` that
// was unable to access the waker cell (due to the registering thread holding the
// lock). To handle this, the registering thread removes the waker it just set
// from the cell and calls `wake` on it. This call to wake represents the
// attempt to wake by the other thread (that set the `WAKING` bit). The
// state is then transitioned from `REGISTERING | WAKING` back to `WAITING`.
// This transition must succeed because, at this point, the state cannot be
// transitioned by another thread.
//
// # Waking
//
// On a call to `wake`, an attempt to transition the state from `WAITING` to
// `WAKING` is made. On success, the caller obtains a lock on the waker cell.
//
// If the lock is obtained, then the thread takes ownership of the current value
// in the waker cell, and calls `wake` on it. The state is then transitioned
// back to `WAITING`. This transition must succeed as, at this point, the state
// cannot be transitioned by another thread.
//
// If the thread is unable to obtain the lock, the `WAKING` bit is still.
// This is because it has either been set by the current thread but the previous
// value included the `REGISTERING` bit **or** a concurrent thread is in the
// `WAKING` critical section. Either way, no action must be taken.
//
// If the current thread is the only concurrent call to `wake` and another
// thread is in the `register` critical section, when the other thread **exits**
// the `register` critical section, it will observe the `WAKING` bit and
// handle the waker itself.
//
// If another thread is in the `waker` critical section, then it will handle
// waking the caller task.
//
// # A potential race (is safely handled).
//
// Imagine the following situation:
//
// * Thread A obtains the `wake` lock and wakes a task.
//
// * Before thread A releases the `wake` lock, the woken task is scheduled.
//
// * Thread B attempts to wake the task. In theory this should result in the
// task being woken, but it cannot because thread A still holds the wake
// lock.
//
// This case is handled by requiring users of `AtomicWaker` to call `register`
// **before** attempting to observe the application state change that resulted
// in the task being woken. The wakers also change the application state
// before calling wake.
//
// Because of this, the task will do one of two things.
//
// 1) Observe the application state change that Thread B is waking on. In
// this case, it is OK for Thread B's wake to be lost.
//
// 2) Call register before attempting to observe the application state. Since
// Thread A still holds the `wake` lock, the call to `register` will result
// in the task waking itself and get scheduled again.
/// Idle state
const WAITING: usize = 0;
/// A new waker value is being registered with the `AtomicWaker` cell.
const REGISTERING: usize = 0b01;
/// The task currently registered with the `AtomicWaker` cell is being woken.
const WAKING: usize = 0b10;
impl AtomicWaker {
/// Create an `AtomicWaker`
pub fn new() -> AtomicWaker {
AtomicWaker {
state: AtomicUsize::new(WAITING),
waker: CausalCell::new(None),
}
}
/// Registers the current waker to be notified on calls to `wake`.
///
/// This is the same as calling `register_task` with `task::current()`.
pub fn register(&self, waker: Waker) {
self.do_register(waker);
}
/// Registers the provided waker to be notified on calls to `wake`.
///
/// The new waker will take place of any previous wakers that were registered
/// by previous calls to `register`. Any calls to `wake` that happen after
/// a call to `register` (as defined by the memory ordering rules), will
/// wake the `register` caller's task.
///
/// It is safe to call `register` with multiple other threads concurrently
/// calling `wake`. This will result in the `register` caller's current
/// task being woken once.
///
/// This function is safe to call concurrently, but this is generally a bad
/// idea. Concurrent calls to `register` will attempt to register different
/// tasks to be woken. One of the callers will win and have its task set,
/// but there is no guarantee as to which caller will succeed.
pub fn register_by_ref(&self, waker: &Waker) {
self.do_register(waker);
}
fn do_register<W>(&self, waker: W)
where
W: WakerRef,
{
debug!(" + register_task");
match self.state.compare_and_swap(WAITING, REGISTERING, Acquire) {
WAITING => {
debug!(" + WAITING");
unsafe {
// Locked acquired, update the waker cell
self.waker.with_mut(|t| *t = Some(waker.into_waker()));
// Release the lock. If the state transitioned to include
// the `WAKING` bit, this means that a wake has been
// called concurrently, so we have to remove the waker and
// wake it.`
//
// Start by assuming that the state is `REGISTERING` as this
// is what we jut set it to.
let res = self
.state
.compare_exchange(REGISTERING, WAITING, AcqRel, Acquire);
match res {
Ok(_) => {}
Err(actual) => {
// This branch can only be reached if a
// concurrent thread called `wake`. In this
// case, `actual` **must** be `REGISTERING |
// `WAKING`.
debug_assert_eq!(actual, REGISTERING | WAKING);
// Take the waker to wake once the atomic operation has
// completed.
let waker = self.waker.with_mut(|t| (*t).take()).unwrap();
// Just swap, because no one could change state
// while state == `Registering | `Waking`
self.state.swap(WAITING, AcqRel);
// The atomic swap was complete, now
// wake the waker and return.
waker.wake();
}
}
}
}
WAKING => {
debug!(" + WAKING");
// Currently in the process of waking the task, i.e.,
// `wake` is currently being called on the old waker.
// So, we call wake on the new waker.
waker.wake();
// This is equivalent to a spin lock, so use a spin hint.
atomic::spin_loop_hint();
}
state => {
// In this case, a concurrent thread is holding the
// "registering" lock. This probably indicates a bug in the
// caller's code as racing to call `register` doesn't make much
// sense.
//
// We just want to maintain memory safety. It is ok to drop the
// call to `register`.
debug_assert!(state == REGISTERING || state == REGISTERING | WAKING);
}
}
}
/// Wakes the task that last called `register`.
///
/// If `register` has not been called yet, then this does nothing.
pub fn wake(&self) {
debug!(" + wake");
if let Some(waker) = self.take_waker() {
waker.wake();
}
}
/// Attempts to take the `Waker` value out of the `AtomicWaker` with the
/// intention that the caller will wake the task later.
pub fn take_waker(&self) -> Option<Waker> {
debug!(" + take_waker");
// AcqRel ordering is used in order to acquire the value of the `waker`
// cell as well as to establish a `release` ordering with whatever
// memory the `AtomicWaker` is associated with.
match self.state.fetch_or(WAKING, AcqRel) {
WAITING => {
debug!(" + WAITING");
// The waking lock has been acquired.
let waker = unsafe { self.waker.with_mut(|t| (*t).take()) };
// Release the lock
self.state.fetch_and(!WAKING, Release);
debug!(" + Done taking");
waker
}
state => {
debug!(" + state = {:?}", state);
// There is a concurrent thread currently updating the
// associated waker.
//
// Nothing more to do as the `WAKING` bit has been set. It
// doesn't matter if there are concurrent registering threads or
// not.
//
debug_assert!(
state == REGISTERING || state == REGISTERING | WAKING || state == WAKING
);
None
}
}
}
}
impl Default for AtomicWaker {
fn default() -> Self {
AtomicWaker::new()
}
}
impl fmt::Debug for AtomicWaker {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(fmt, "AtomicWaker")
}
}
unsafe impl Send for AtomicWaker {}
unsafe impl Sync for AtomicWaker {}
trait WakerRef {
fn wake(self);
fn into_waker(self) -> Waker;
}
impl WakerRef for Waker {
fn wake(self) {
self.wake()
}
fn into_waker(self) -> Waker {
self
}
}
impl WakerRef for &Waker {
fn wake(self) {
self.wake_by_ref()
}
fn into_waker(self) -> Waker {
self.clone()
}
}
+4
View File
@@ -0,0 +1,4 @@
//! Thread-safe task notification primitives.
mod atomic_waker;
pub use self::atomic_waker::AtomicWaker;
+45
View File
@@ -0,0 +1,45 @@
use crate::sync::task::AtomicWaker;
use futures_util::future::poll_fn;
use loom::future::block_on;
use loom::sync::atomic::AtomicUsize;
use loom::thread;
use std::sync::atomic::Ordering::Relaxed;
use std::sync::Arc;
use std::task::Poll::{Pending, Ready};
struct Chan {
num: AtomicUsize,
task: AtomicWaker,
}
#[test]
fn basic_notification() {
const NUM_NOTIFY: usize = 2;
loom::model(|| {
let chan = Arc::new(Chan {
num: AtomicUsize::new(0),
task: AtomicWaker::new(),
});
for _ in 0..NUM_NOTIFY {
let chan = chan.clone();
thread::spawn(move || {
chan.num.fetch_add(1, Relaxed);
chan.task.wake();
});
}
block_on(poll_fn(move |cx| {
chan.task.register_by_ref(cx.waker());
if NUM_NOTIFY == chan.num.load(Relaxed) {
return Ready(());
}
Pending
}));
});
}
+52
View File
@@ -0,0 +1,52 @@
use crate::sync::mpsc::list;
use loom::thread;
use std::sync::Arc;
#[test]
fn smoke() {
use crate::sync::mpsc::block::Read::*;
const NUM_TX: usize = 2;
const NUM_MSG: usize = 2;
loom::model(|| {
let (tx, mut rx) = list::channel();
let tx = Arc::new(tx);
for th in 0..NUM_TX {
let tx = tx.clone();
thread::spawn(move || {
for i in 0..NUM_MSG {
tx.push((th, i));
}
debug!(" + tx thread done");
});
}
let mut next = vec![0; NUM_TX];
loop {
debug!(" + rx.pop()");
match rx.pop(&tx) {
Some(Value((th, v))) => {
debug!(" + pop() -> Some(Value({}))", v);
assert_eq!(v, next[th]);
next[th] += 1;
if next.iter().all(|&i| i == NUM_MSG) {
break;
}
}
Some(Closed) => {
panic!();
}
None => {
debug!(" + pop() -> None");
thread::yield_now();
}
}
}
});
}
+23
View File
@@ -0,0 +1,23 @@
use crate::sync::mpsc;
use futures_util::future::poll_fn;
use loom::future::block_on;
use loom::thread;
#[test]
fn closing_tx() {
loom::model(|| {
let (mut 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)));
assert!(v.is_some());
let v = block_on(poll_fn(|cx| rx.poll_recv(cx)));
assert!(v.is_none());
});
}
+109
View File
@@ -0,0 +1,109 @@
use crate::sync::oneshot;
use futures_util::future::poll_fn;
use loom::future::block_on;
use loom::thread;
use std::task::Poll::{Pending, Ready};
#[test]
fn smoke() {
loom::model(|| {
let (tx, rx) = oneshot::channel();
thread::spawn(move || {
tx.send(1).unwrap();
});
let value = block_on(rx).unwrap();
assert_eq!(1, value);
});
}
#[test]
fn changing_rx_task() {
loom::model(|| {
let (tx, mut rx) = oneshot::channel();
thread::spawn(move || {
tx.send(1).unwrap();
});
let rx = thread::spawn(move || {
let ready = block_on(poll_fn(|cx| match Pin::new(&mut rx).poll(cx) {
Ready(Ok(value)) => {
assert_eq!(1, value);
Ready(true)
}
Ready(Err(_)) => unimplemented!(),
Pending => Ready(false),
}));
if ready {
None
} else {
Some(rx)
}
})
.join()
.unwrap();
if let Some(rx) = rx {
// Previous task parked, use a new task...
let value = block_on(rx).unwrap();
assert_eq!(1, value);
}
});
}
// TODO: Move this into `oneshot` proper.
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
struct OnClose<'a> {
tx: &'a mut oneshot::Sender<i32>,
}
impl<'a> OnClose<'a> {
fn new(tx: &'a mut oneshot::Sender<i32>) -> Self {
OnClose { tx }
}
}
impl Future for OnClose<'_> {
type Output = bool;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<bool> {
let res = self.get_mut().tx.poll_closed(cx);
Ready(res.is_ready())
}
}
#[test]
fn changing_tx_task() {
loom::model(|| {
let (mut tx, rx) = oneshot::channel::<i32>();
thread::spawn(move || {
drop(rx);
});
let tx = thread::spawn(move || {
let t1 = block_on(OnClose::new(&mut tx));
if t1 {
None
} else {
Some(tx)
}
})
.join()
.unwrap();
if let Some(mut tx) = tx {
// Previous task parked, use a new task...
block_on(OnClose::new(&mut tx));
}
});
}
+151
View File
@@ -0,0 +1,151 @@
use crate::sync::semaphore::*;
use futures_core::ready;
use futures_util::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, &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(&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, &semaphore))).unwrap();
permit.release(&semaphore);
});
}
let mut permit = Permit::new();
block_on(poll_fn(|cx| permit.poll_acquire(cx, &semaphore))).unwrap();
permit.release(&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, &semaphore).map_err(|_| ())
}))?;
permit.release(&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, &semaphore).map_err(|_| ())
}))?;
permit.release(&semaphore);
semaphore.close();
Ok::<(), ()>(())
});
}
});
}
+7
View File
@@ -0,0 +1,7 @@
#![cfg(loom)]
mod loom_atomic_waker;
mod loom_list;
mod loom_mpsc;
mod loom_oneshot;
mod loom_semaphore;
+454
View File
@@ -0,0 +1,454 @@
//! A single-producer, multi-consumer channel that only retains the *last* sent
//! value.
//!
//! This channel is useful for watching for changes to a value from multiple
//! points in the code base, for example, changes to configuration values.
//!
//! # Usage
//!
//! [`channel`] returns a [`Sender`] / [`Receiver`] pair. These are
//! the producer and sender halves of the channel. The channel is
//! created with an initial value. [`Receiver::get_ref`] will always
//! be ready upon creation and will yield either this initial value or
//! the latest value that has been sent by `Sender`.
//!
//! Calls to [`Receiver::get_ref`] will always yield the latest value.
//!
//! # Examples
//!
//! ```
//! use tokio::sync::watch;
//!
//! # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
//! let (tx, mut rx) = watch::channel("hello");
//!
//! tokio::spawn(async move {
//! while let Some(value) = rx.recv().await {
//! println!("received = {:?}", value);
//! }
//! });
//!
//! tx.broadcast("world")?;
//! # Ok(())
//! # }
//! ```
//!
//! # Closing
//!
//! [`Sender::closed`] allows the producer to detect when all [`Receiver`]
//! handles have been dropped. This indicates that there is no further interest
//! in the values being produced and work can be stopped.
//!
//! # Thread safety
//!
//! Both [`Sender`] and [`Receiver`] are thread safe. They can be moved to other
//! threads and can be used in a concurrent environment. Clones of [`Receiver`]
//! handles may be moved to separate threads and also used concurrently.
//!
//! [`Sender`]: struct.Sender.html
//! [`Receiver`]: struct.Receiver.html
//! [`channel`]: fn.channel.html
//! [`Sender::closed`]: struct.Sender.html#method.closed
//! [`Receiver::get_ref`]: struct.Receiver.html#method.get_ref
use crate::sync::task::AtomicWaker;
use core::task::Poll::{Pending, Ready};
use core::task::{Context, Poll};
use fnv::FnvHashMap;
use futures_util::future::poll_fn;
use std::ops;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering::SeqCst;
use std::sync::{Arc, Mutex, RwLock, RwLockReadGuard, Weak};
use futures_core::ready;
use futures_util::pin_mut;
use std::pin::Pin;
/// Receives values from the associated [`Sender`](struct.Sender.html).
///
/// Instances are created by the [`channel`](fn.channel.html) function.
#[derive(Debug)]
pub struct Receiver<T> {
/// Pointer to the shared state
shared: Arc<Shared<T>>,
/// Pointer to the watcher's internal state
inner: Arc<WatchInner>,
/// Watcher ID.
id: u64,
/// Last observed version
ver: usize,
}
/// Sends values to the associated [`Receiver`](struct.Receiver.html).
///
/// Instances are created by the [`channel`](fn.channel.html) function.
#[derive(Debug)]
pub struct Sender<T> {
shared: Weak<Shared<T>>,
}
/// Returns a reference to the inner value
///
/// Outstanding borrows hold a read lock on the inner value. This means that
/// long lived borrows could cause the produce half to block. It is recommended
/// to keep the borrow as short lived as possible.
#[derive(Debug)]
pub struct Ref<'a, T> {
inner: RwLockReadGuard<'a, T>,
}
pub mod error {
//! Watch error types
use std::fmt;
/// Error produced when sending a value fails.
#[derive(Debug)]
pub struct SendError<T> {
pub(crate) inner: T,
}
// ===== impl SendError =====
impl<T: fmt::Debug> fmt::Display for SendError<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(fmt, "channel closed")
}
}
impl<T: fmt::Debug> ::std::error::Error for SendError<T> {}
}
#[derive(Debug)]
struct Shared<T> {
/// The most recent value
value: RwLock<T>,
/// The current version
///
/// The lowest bit represents a "closed" state. The rest of the bits
/// represent the current version.
version: AtomicUsize,
/// All watchers
watchers: Mutex<Watchers>,
/// Task to notify when all watchers drop
cancel: AtomicWaker,
}
#[derive(Debug)]
struct Watchers {
next_id: u64,
watchers: FnvHashMap<u64, Arc<WatchInner>>,
}
#[derive(Debug)]
struct WatchInner {
waker: AtomicWaker,
}
const CLOSED: usize = 1;
/// Create a new watch channel, returning the "send" and "receive" handles.
///
/// All values sent by [`Sender`] will become visible to the [`Receiver`] handles.
/// Only the last value sent is made available to the [`Receiver`] half. All
/// intermediate values are dropped.
///
/// # Examples
///
/// ```
/// use tokio::sync::watch;
///
/// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
/// let (tx, mut rx) = watch::channel("hello");
///
/// tokio::spawn(async move {
/// while let Some(value) = rx.recv().await {
/// println!("received = {:?}", value);
/// }
/// });
///
/// tx.broadcast("world")?;
/// # Ok(())
/// # }
/// ```
///
/// [`Sender`]: struct.Sender.html
/// [`Receiver`]: struct.Receiver.html
pub fn channel<T>(init: T) -> (Sender<T>, Receiver<T>) {
const INIT_ID: u64 = 0;
let inner = Arc::new(WatchInner::new());
// Insert the watcher
let mut watchers = FnvHashMap::with_capacity_and_hasher(0, Default::default());
watchers.insert(INIT_ID, inner.clone());
let shared = Arc::new(Shared {
value: RwLock::new(init),
version: AtomicUsize::new(2),
watchers: Mutex::new(Watchers {
next_id: INIT_ID + 1,
watchers,
}),
cancel: AtomicWaker::new(),
});
let tx = Sender {
shared: Arc::downgrade(&shared),
};
let rx = Receiver {
shared,
inner,
id: INIT_ID,
ver: 0,
};
(tx, rx)
}
impl<T> Receiver<T> {
/// Returns a reference to the most recently sent value
///
/// Outstanding borrows hold a read lock. This means that long lived borrows
/// could cause the send half to block. It is recommended to keep the borrow
/// as short lived as possible.
///
/// # Examples
///
/// ```
/// use tokio::sync::watch;
///
/// let (_, rx) = watch::channel("hello");
/// assert_eq!(*rx.get_ref(), "hello");
/// ```
pub fn get_ref(&self) -> Ref<'_, T> {
let inner = self.shared.value.read().unwrap();
Ref { inner }
}
/// Attempts to receive the latest value sent via the channel.
///
/// If a new, unobserved, value has been sent, a reference to it is
/// returned. If no new value has been sent, then `Pending` is returned and
/// the current task is notified once a new value is sent.
///
/// Only the **most recent** value is returned. If the receiver is falling
/// behind the sender, intermediate values are dropped.
pub async fn recv_ref(&mut self) -> Option<Ref<'_, T>> {
let shared = &self.shared;
let inner = &self.inner;
let version = self.ver;
match poll_fn(|cx| poll_lock(cx, shared, inner, version)).await {
Some((lock, version)) => {
self.ver = version;
Some(lock)
}
None => None,
}
}
}
fn poll_lock<'a, T>(
cx: &mut Context<'_>,
shared: &'a Arc<Shared<T>>,
inner: &Arc<WatchInner>,
ver: usize,
) -> Poll<Option<(Ref<'a, T>, usize)>> {
// Make sure the task is up to date
inner.waker.register_by_ref(cx.waker());
let state = shared.version.load(SeqCst);
let version = state & !CLOSED;
if version != ver {
let inner = shared.value.read().unwrap();
return Ready(Some((Ref { inner }, version)));
}
if CLOSED == state & CLOSED {
// The `Store` handle has been dropped.
return Ready(None);
}
Pending
}
impl<T: Clone> Receiver<T> {
/// Attempts to clone the latest value sent via the channel.
///
/// This is equivalent to calling `clone()` on the value returned by
/// `recv_ref()`.
#[allow(clippy::map_clone)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3274
pub async fn recv(&mut self) -> Option<T> {
self.recv_ref().await.map(|v_ref| v_ref.clone())
}
}
impl<T: Clone> futures_core::Stream for Receiver<T> {
type Item = T;
#[allow(clippy::map_clone)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3274
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
use std::future::Future;
let fut = self.get_mut().recv();
pin_mut!(fut);
let item = ready!(fut.poll(cx));
Ready(item.map(|v_ref| v_ref.clone()))
}
}
impl<T> Clone for Receiver<T> {
fn clone(&self) -> Self {
let inner = Arc::new(WatchInner::new());
let shared = self.shared.clone();
let id = {
let mut watchers = shared.watchers.lock().unwrap();
let id = watchers.next_id;
watchers.next_id += 1;
watchers.watchers.insert(id, inner.clone());
id
};
let ver = self.ver;
Receiver {
shared,
inner,
id,
ver,
}
}
}
impl<T> Drop for Receiver<T> {
fn drop(&mut self) {
let mut watchers = self.shared.watchers.lock().unwrap();
watchers.watchers.remove(&self.id);
}
}
impl WatchInner {
fn new() -> Self {
WatchInner {
waker: AtomicWaker::new(),
}
}
}
impl<T> Sender<T> {
/// Broadcast a new value via the channel, notifying all receivers.
pub fn broadcast(&self, value: T) -> Result<(), error::SendError<T>> {
let shared = match self.shared.upgrade() {
Some(shared) => shared,
// All `Watch` handles have been canceled
None => return Err(error::SendError { inner: value }),
};
// Replace the value
{
let mut lock = shared.value.write().unwrap();
*lock = value;
}
// Update the version. 2 is used so that the CLOSED bit is not set.
shared.version.fetch_add(2, SeqCst);
// Notify all watchers
notify_all(&*shared);
// Return the old value
Ok(())
}
/// Completes when all receivers have dropped.
///
/// This allows the producer to get notified when interest in the produced
/// values is canceled and immediately stop doing work.
pub async fn closed(&mut self) {
poll_fn(|cx| self.poll_close(cx)).await
}
fn poll_close(&mut self, cx: &mut Context<'_>) -> Poll<()> {
match self.shared.upgrade() {
Some(shared) => {
shared.cancel.register_by_ref(cx.waker());
Pending
}
None => Ready(()),
}
}
}
impl<T> futures_sink::Sink<T> for Sender<T> {
type Error = error::SendError<T>;
fn poll_ready(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Ready(Ok(()))
}
fn start_send(self: Pin<&mut Self>, item: T) -> Result<(), Self::Error> {
self.as_ref().get_ref().broadcast(item)?;
Ok(())
}
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Ready(Ok(()))
}
fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Ready(Ok(()))
}
}
/// Notify all watchers of a change
fn notify_all<T>(shared: &Shared<T>) {
let watchers = shared.watchers.lock().unwrap();
for watcher in watchers.watchers.values() {
// Notify the task
watcher.waker.wake();
}
}
impl<T> Drop for Sender<T> {
fn drop(&mut self) {
if let Some(shared) = self.shared.upgrade() {
shared.version.fetch_or(CLOSED, SeqCst);
notify_all(&*shared);
}
}
}
// ===== impl Ref =====
impl<T> ops::Deref for Ref<'_, T> {
type Target = T;
fn deref(&self) -> &T {
self.inner.deref()
}
}
// ===== impl Shared =====
impl<T> Drop for Shared<T> {
fn drop(&mut self) {
self.cancel.wake();
}
}
+1 -2
View File
@@ -1,9 +1,8 @@
use crate::sync::AtomicWaker;
use crate::timer::atomic::AtomicU64;
use crate::timer::timer::{HandlePriv, Inner};
use crate::timer::Error;
use tokio_sync::AtomicWaker;
use crossbeam_utils::CachePadded;
use std::cell::UnsafeCell;
use std::ptr;
+1 -1
View File
@@ -1,4 +1,4 @@
use tokio_sync::oneshot;
use tokio::sync::oneshot;
use std::cell::RefCell;
use std::collections::VecDeque;
+38
View File
@@ -0,0 +1,38 @@
#![warn(rust_2018_idioms)]
use tokio::sync::AtomicWaker;
use tokio_test::task::MockTask;
use std::task::Waker;
trait AssertSend: Send {}
trait AssertSync: Send {}
impl AssertSend for AtomicWaker {}
impl AssertSync for AtomicWaker {}
impl AssertSend for Waker {}
impl AssertSync for Waker {}
#[test]
fn basic_usage() {
let waker = AtomicWaker::new();
let mut task = MockTask::new();
task.enter(|cx| waker.register_by_ref(cx.waker()));
waker.wake();
assert!(task.is_woken());
}
#[test]
fn wake_without_register() {
let waker = AtomicWaker::new();
waker.wake();
// Registering should not result in a notification
let mut task = MockTask::new();
task.enter(|cx| waker.register_by_ref(cx.waker()));
assert!(!task.is_woken());
}
+94
View File
@@ -0,0 +1,94 @@
#![warn(rust_2018_idioms)]
use tokio::sync::Barrier;
use tokio_test::task::spawn;
use tokio_test::{assert_pending, assert_ready};
struct IsSend<T: Send>(T);
#[test]
fn barrier_future_is_send() {
let b = Barrier::new(0);
IsSend(b.wait());
}
#[test]
fn zero_does_not_block() {
let b = Barrier::new(0);
{
let mut w = spawn(b.wait());
let wr = assert_ready!(w.poll());
assert!(wr.is_leader());
}
{
let mut w = spawn(b.wait());
let wr = assert_ready!(w.poll());
assert!(wr.is_leader());
}
}
#[test]
fn single() {
let b = Barrier::new(1);
{
let mut w = spawn(b.wait());
let wr = assert_ready!(w.poll());
assert!(wr.is_leader());
}
{
let mut w = spawn(b.wait());
let wr = assert_ready!(w.poll());
assert!(wr.is_leader());
}
{
let mut w = spawn(b.wait());
let wr = assert_ready!(w.poll());
assert!(wr.is_leader());
}
}
#[test]
fn tango() {
let b = Barrier::new(2);
let mut w1 = spawn(b.wait());
assert_pending!(w1.poll());
let mut w2 = spawn(b.wait());
let wr2 = assert_ready!(w2.poll());
let wr1 = assert_ready!(w1.poll());
assert!(wr1.is_leader() || wr2.is_leader());
assert!(!(wr1.is_leader() && wr2.is_leader()));
}
#[test]
fn lots() {
let b = Barrier::new(100);
for _ in 0..10 {
let mut wait = Vec::new();
for _ in 0..99 {
let mut w = spawn(b.wait());
assert_pending!(w.poll());
wait.push(w);
}
for w in &mut wait {
assert_pending!(w.poll());
}
// pass the barrier
let mut w = spawn(b.wait());
let mut found_leader = assert_ready!(w.poll()).is_leader();
for mut w in wait {
let wr = assert_ready!(w.poll());
if wr.is_leader() {
assert!(!found_leader);
found_leader = true;
}
}
assert!(found_leader);
}
}
+29
View File
@@ -0,0 +1,29 @@
#![warn(rust_2018_idioms)]
fn is_error<T: ::std::error::Error + Send + Sync>() {}
#[test]
fn mpsc_error_bound() {
use tokio::sync::mpsc::error;
is_error::<error::SendError>();
is_error::<error::TrySendError<()>>();
is_error::<error::UnboundedRecvError>();
is_error::<error::UnboundedSendError>();
is_error::<error::UnboundedTrySendError<()>>();
}
#[test]
fn oneshot_error_bound() {
use tokio::sync::oneshot::error;
is_error::<error::RecvError>();
is_error::<error::TryRecvError>();
}
#[test]
fn watch_error_bound() {
use tokio::sync::watch::error;
is_error::<error::SendError<()>>();
}
+451
View File
@@ -0,0 +1,451 @@
#![warn(rust_2018_idioms)]
use tokio::sync::mpsc;
use tokio_test::task::MockTask;
use tokio_test::{
assert_err, assert_ok, assert_pending, assert_ready, assert_ready_err, assert_ready_ok,
};
use std::sync::Arc;
trait AssertSend: Send {}
impl AssertSend for mpsc::Sender<i32> {}
impl AssertSend for mpsc::Receiver<i32> {}
#[test]
fn send_recv_with_buffer() {
let mut t1 = MockTask::new();
let mut t2 = MockTask::new();
let (mut tx, mut rx) = mpsc::channel::<i32>(16);
// Using poll_ready / try_send
assert_ready_ok!(t1.enter(|cx| tx.poll_ready(cx)));
tx.try_send(1).unwrap();
// Without poll_ready
tx.try_send(2).unwrap();
drop(tx);
let val = assert_ready!(t2.enter(|cx| rx.poll_recv(cx)));
assert_eq!(val, Some(1));
let val = assert_ready!(t2.enter(|cx| rx.poll_recv(cx)));
assert_eq!(val, Some(2));
let val = assert_ready!(t2.enter(|cx| rx.poll_recv(cx)));
assert!(val.is_none());
}
#[tokio::test]
async fn async_send_recv_with_buffer() {
let (mut tx, mut rx) = mpsc::channel(16);
tokio::spawn(async move {
assert_ok!(tx.send(1).await);
assert_ok!(tx.send(2).await);
});
assert_eq!(Some(1), rx.recv().await);
assert_eq!(Some(2), rx.recv().await);
assert_eq!(None, rx.recv().await);
}
#[test]
fn send_sink_recv_with_buffer() {
use futures_core::Stream;
use futures_sink::Sink;
use futures_util::pin_mut;
let mut t1 = MockTask::new();
let (tx, rx) = mpsc::channel::<i32>(16);
t1.enter(|cx| {
pin_mut!(tx);
assert_ready_ok!(tx.as_mut().poll_ready(cx));
assert_ok!(tx.as_mut().start_send(1));
assert_ready_ok!(tx.as_mut().poll_ready(cx));
assert_ok!(tx.as_mut().start_send(2));
assert_ready_ok!(tx.as_mut().poll_flush(cx));
assert_ready_ok!(tx.as_mut().poll_close(cx));
});
t1.enter(|cx| {
pin_mut!(rx);
let val = assert_ready!(rx.as_mut().poll_next(cx));
assert_eq!(val, Some(1));
let val = assert_ready!(rx.as_mut().poll_next(cx));
assert_eq!(val, Some(2));
let val = assert_ready!(rx.as_mut().poll_next(cx));
assert!(val.is_none());
});
}
#[test]
fn start_send_past_cap() {
let mut t1 = MockTask::new();
let mut t2 = MockTask::new();
let mut t3 = MockTask::new();
let (mut tx1, mut rx) = mpsc::channel(1);
let mut tx2 = tx1.clone();
assert_ok!(tx1.try_send(()));
t1.enter(|cx| {
assert_pending!(tx1.poll_ready(cx));
});
t2.enter(|cx| {
assert_pending!(tx2.poll_ready(cx));
});
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());
}
#[test]
#[should_panic]
fn buffer_gteq_one() {
mpsc::channel::<i32>(0);
}
#[test]
fn send_recv_unbounded() {
let mut t1 = MockTask::new();
let (mut tx, mut rx) = mpsc::unbounded_channel::<i32>();
// Using `try_send`
assert_ok!(tx.try_send(1));
assert_ok!(tx.try_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));
drop(tx);
let val = assert_ready!(t1.enter(|cx| rx.poll_recv(cx)));
assert!(val.is_none());
}
#[tokio::test]
async fn async_send_recv_unbounded() {
let (mut tx, mut rx) = mpsc::unbounded_channel();
tokio::spawn(async move {
assert_ok!(tx.try_send(1));
assert_ok!(tx.try_send(2));
});
assert_eq!(Some(1), rx.recv().await);
assert_eq!(Some(2), rx.recv().await);
assert_eq!(None, rx.recv().await);
}
#[test]
fn sink_send_recv_unbounded() {
use futures_core::Stream;
use futures_sink::Sink;
use futures_util::pin_mut;
let mut t1 = MockTask::new();
let (tx, rx) = mpsc::unbounded_channel::<i32>();
t1.enter(|cx| {
pin_mut!(tx);
assert_ready_ok!(tx.as_mut().poll_ready(cx));
assert_ok!(tx.as_mut().start_send(1));
assert_ready_ok!(tx.as_mut().poll_ready(cx));
assert_ok!(tx.as_mut().start_send(2));
assert_ready_ok!(tx.as_mut().poll_flush(cx));
assert_ready_ok!(tx.as_mut().poll_close(cx));
});
t1.enter(|cx| {
pin_mut!(rx);
let val = assert_ready!(rx.as_mut().poll_next(cx));
assert_eq!(val, Some(1));
let val = assert_ready!(rx.as_mut().poll_next(cx));
assert_eq!(val, Some(2));
let val = assert_ready!(rx.as_mut().poll_next(cx));
assert!(val.is_none());
});
}
#[test]
fn no_t_bounds_buffer() {
struct NoImpls;
let mut t1 = MockTask::new();
let (tx, mut rx) = mpsc::channel(100);
// sender should be Debug even though T isn't Debug
println!("{:?}", tx);
// same with Receiver
println!("{:?}", rx);
// 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());
}
#[test]
fn no_t_bounds_unbounded() {
struct NoImpls;
let mut t1 = MockTask::new();
let (tx, mut rx) = mpsc::unbounded_channel();
// sender should be Debug even though T isn't Debug
println!("{:?}", tx);
// same with Receiver
println!("{:?}", rx);
// 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());
}
#[test]
fn send_recv_buffer_limited() {
let mut t1 = MockTask::new();
let mut t2 = MockTask::new();
let (mut tx, mut rx) = mpsc::channel::<i32>(1);
// Run on a task context
t1.enter(|cx| {
assert_ready_ok!(tx.poll_ready(cx));
// Send first message
assert_ok!(tx.try_send(1));
// Not ready
assert_pending!(tx.poll_ready(cx));
// Send second message
assert_err!(tx.try_send(1337));
});
t2.enter(|cx| {
// Take the value
let val = assert_ready!(rx.poll_recv(cx));
assert_eq!(Some(1), val);
});
assert!(t1.is_woken());
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));
});
}
#[test]
fn recv_close_gets_none_idle() {
let mut t1 = MockTask::new();
let (mut 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));
});
}
#[test]
fn recv_close_gets_none_reserved() {
let mut t1 = MockTask::new();
let mut t2 = MockTask::new();
let mut t3 = MockTask::new();
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));
});
rx.close();
assert!(t2.is_woken());
t2.enter(|cx| {
assert_ready_err!(tx2.poll_ready(cx));
});
t3.enter(|cx| assert_pending!(rx.poll_recv(cx)));
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));
assert_eq!(v, Some(123));
let v = assert_ready!(rx.poll_recv(cx));
assert!(v.is_none());
});
}
#[test]
fn tx_close_gets_none() {
let mut t1 = MockTask::new();
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());
});
}
#[test]
fn try_send_fail() {
let mut t1 = MockTask::new();
let (mut tx, mut rx) = mpsc::channel(1);
tx.try_send("hello").unwrap();
// This should fail
let err = assert_err!(tx.try_send("fail"));
assert!(err.is_full());
let val = assert_ready!(t1.enter(|cx| rx.poll_recv(cx)));
assert_eq!(val, 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());
}
#[test]
fn drop_tx_with_permit_releases_permit() {
let mut t1 = MockTask::new();
let mut t2 = MockTask::new();
// 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();
assert_ready_ok!(t1.enter(|cx| tx1.poll_ready(cx)));
t2.enter(|cx| {
assert_pending!(tx2.poll_ready(cx));
});
drop(tx1);
assert!(t2.is_woken());
assert_ready_ok!(t2.enter(|cx| tx2.poll_ready(cx)));
}
#[test]
fn dropping_rx_closes_channel() {
let mut t1 = MockTask::new();
let (mut 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_eq!(1, Arc::strong_count(&msg));
}
#[test]
fn dropping_rx_closes_channel_for_try() {
let (mut tx, rx) = mpsc::channel(100);
let msg = Arc::new(());
tx.try_send(msg.clone()).unwrap();
drop(rx);
{
let err = assert_err!(tx.try_send(msg.clone()));
assert!(err.is_closed());
}
assert_eq!(1, Arc::strong_count(&msg));
}
#[test]
fn unconsumed_messages_are_dropped() {
let msg = Arc::new(());
let (mut tx, rx) = mpsc::channel(100);
tx.try_send(msg.clone()).unwrap();
assert_eq!(2, Arc::strong_count(&msg));
drop((tx, rx));
assert_eq!(1, Arc::strong_count(&msg));
}
+80
View File
@@ -0,0 +1,80 @@
#![warn(rust_2018_idioms)]
use tokio::sync::Mutex;
use tokio_test::task::spawn;
use tokio_test::{assert_pending, assert_ready};
use std::sync::Arc;
#[test]
fn straight_execution() {
let l = Mutex::new(100);
{
let mut t = spawn(l.lock());
let mut g = assert_ready!(t.poll());
assert_eq!(&*g, &100);
*g = 99;
}
{
let mut t = spawn(l.lock());
let mut g = assert_ready!(t.poll());
assert_eq!(&*g, &99);
*g = 98;
}
{
let mut t = spawn(l.lock());
let g = assert_ready!(t.poll());
assert_eq!(&*g, &98);
}
}
#[test]
fn readiness() {
let l1 = Arc::new(Mutex::new(100));
let l2 = Arc::clone(&l1);
let mut t1 = spawn(l1.lock());
let mut t2 = spawn(l2.lock());
let g = assert_ready!(t1.poll());
// We can't now acquire the lease since it's already held in g
assert_pending!(t2.poll());
// But once g unlocks, we can acquire it
drop(g);
assert!(t2.is_woken());
assert_ready!(t2.poll());
}
/*
#[test]
#[ignore]
fn lock() {
let mut lock = Mutex::new(false);
let mut lock2 = lock.clone();
std::thread::spawn(move || {
let l = lock2.lock();
pin_mut!(l);
let mut task = MockTask::new();
let mut g = assert_ready!(task.poll(&mut l));
std::thread::sleep(std::time::Duration::from_millis(500));
*g = true;
drop(g);
});
std::thread::sleep(std::time::Duration::from_millis(50));
let mut task = MockTask::new();
let l = lock.lock();
pin_mut!(l);
assert_pending!(task.poll(&mut l));
std::thread::sleep(std::time::Duration::from_millis(500));
assert!(task.is_woken());
let result = assert_ready!(task.poll(&mut l));
assert!(*result);
}
*/
+228
View File
@@ -0,0 +1,228 @@
#![warn(rust_2018_idioms)]
use tokio::sync::oneshot;
use tokio_test::task::MockTask;
use tokio_test::*;
trait AssertSend: Send {}
impl AssertSend for oneshot::Sender<i32> {}
impl AssertSend for oneshot::Receiver<i32> {}
#[test]
fn send_recv() {
let (tx, mut rx) = oneshot::channel();
let mut task = MockTask::new();
assert_pending!(task.poll(&mut rx));
assert_ok!(tx.send(1));
assert!(task.is_woken());
let val = assert_ready_ok!(task.poll(&mut rx));
assert_eq!(val, 1);
}
#[tokio::test]
async fn async_send_recv() {
let (tx, rx) = oneshot::channel();
assert_ok!(tx.send(1));
assert_eq!(1, assert_ok!(rx.await));
}
#[test]
fn close_tx() {
let (tx, mut rx) = oneshot::channel::<i32>();
let mut task = MockTask::new();
assert_pending!(task.poll(&mut rx));
drop(tx);
assert!(task.is_woken());
assert_ready_err!(task.poll(&mut rx));
}
#[test]
fn close_rx() {
// First, without checking poll_closed()
//
let (tx, _) = oneshot::channel();
assert_err!(tx.send(1));
// Second, via poll_closed();
let (mut tx, rx) = oneshot::channel();
let mut task = MockTask::new();
assert_pending!(task.enter(|cx| tx.poll_closed(cx)));
drop(rx);
assert!(task.is_woken());
assert!(tx.is_closed());
assert_ready!(task.enter(|cx| tx.poll_closed(cx)));
assert_err!(tx.send(1));
}
#[tokio::test]
async fn async_rx_closed() {
let (mut tx, rx) = oneshot::channel::<()>();
tokio::spawn(async move {
drop(rx);
});
tx.closed().await;
}
#[test]
fn explicit_close_poll() {
// First, with message sent
let (tx, mut rx) = oneshot::channel();
let mut task = MockTask::new();
assert_ok!(tx.send(1));
rx.close();
let value = assert_ready_ok!(task.poll(&mut rx));
assert_eq!(value, 1);
// Second, without the message sent
let (mut tx, mut rx) = oneshot::channel::<i32>();
assert_pending!(task.enter(|cx| tx.poll_closed(cx)));
rx.close();
assert!(task.is_woken());
assert!(tx.is_closed());
assert_ready!(task.enter(|cx| tx.poll_closed(cx)));
assert_err!(tx.send(1));
assert_ready_err!(task.poll(&mut rx));
// Again, but without sending the value this time
let (mut tx, mut rx) = oneshot::channel::<i32>();
let mut task = MockTask::new();
assert_pending!(task.enter(|cx| tx.poll_closed(cx)));
rx.close();
assert!(task.is_woken());
assert!(tx.is_closed());
assert_ready!(task.enter(|cx| tx.poll_closed(cx)));
assert_ready_err!(task.poll(&mut rx));
}
#[test]
fn explicit_close_try_recv() {
// First, with message sent
let (tx, mut rx) = oneshot::channel();
assert_ok!(tx.send(1));
rx.close();
let val = assert_ok!(rx.try_recv());
assert_eq!(1, val);
// Second, without the message sent
let (mut tx, mut rx) = oneshot::channel::<i32>();
let mut task = MockTask::new();
assert_pending!(task.enter(|cx| tx.poll_closed(cx)));
rx.close();
assert!(task.is_woken());
assert!(tx.is_closed());
assert_ready!(task.enter(|cx| tx.poll_closed(cx)));
assert_err!(rx.try_recv());
}
#[test]
#[should_panic]
fn close_try_recv_poll() {
let (_tx, mut rx) = oneshot::channel::<i32>();
let mut task = MockTask::new();
rx.close();
assert_err!(rx.try_recv());
let _ = task.poll(&mut rx);
}
#[test]
fn drops_tasks() {
let (mut tx, mut rx) = oneshot::channel::<i32>();
let mut tx_task = MockTask::new();
let mut rx_task = MockTask::new();
assert_pending!(tx_task.enter(|cx| tx.poll_closed(cx)));
assert_pending!(rx_task.poll(&mut rx));
drop(tx);
drop(rx);
assert_eq!(1, tx_task.waker_ref_count());
assert_eq!(1, rx_task.waker_ref_count());
}
#[test]
fn receiver_changes_task() {
let (tx, mut rx) = oneshot::channel();
let mut task1 = MockTask::new();
let mut task2 = MockTask::new();
assert_pending!(task1.poll(&mut rx));
assert_eq!(2, task1.waker_ref_count());
assert_eq!(1, task2.waker_ref_count());
assert_pending!(task2.poll(&mut rx));
assert_eq!(1, task1.waker_ref_count());
assert_eq!(2, task2.waker_ref_count());
assert_ok!(tx.send(1));
assert!(!task1.is_woken());
assert!(task2.is_woken());
assert_ready_ok!(task2.poll(&mut rx));
}
#[test]
fn sender_changes_task() {
let (mut tx, rx) = oneshot::channel::<i32>();
let mut task1 = MockTask::new();
let mut task2 = MockTask::new();
assert_pending!(task1.enter(|cx| tx.poll_closed(cx)));
assert_eq!(2, task1.waker_ref_count());
assert_eq!(1, task2.waker_ref_count());
assert_pending!(task2.enter(|cx| tx.poll_closed(cx)));
assert_eq!(1, task1.waker_ref_count());
assert_eq!(2, task2.waker_ref_count());
drop(rx);
assert!(!task1.is_woken());
assert!(task2.is_woken());
assert_ready!(task2.enter(|cx| tx.poll_closed(cx)));
}
+153
View File
@@ -0,0 +1,153 @@
#![warn(rust_2018_idioms)]
use tokio::sync::semaphore::{Permit, Semaphore};
use tokio_test::task::MockTask;
use tokio_test::{assert_pending, assert_ready_err, assert_ready_ok};
#[test]
fn available_permits() {
let mut t1 = MockTask::new();
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_ready_ok!(t1.enter(|cx| permit.poll_acquire(cx, &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!(t1.enter(|cx| permit.poll_acquire(cx, &s)));
assert_eq!(s.available_permits(), 99);
assert!(permit.is_acquired());
}
#[test]
fn unavailable_permits() {
let mut t1 = MockTask::new();
let mut t2 = MockTask::new();
let s = Semaphore::new(1);
let mut permit_1 = Permit::new();
let mut permit_2 = Permit::new();
// Acquire the first permit
assert_ready_ok!(t1.enter(|cx| permit_1.poll_acquire(cx, &s)));
assert_eq!(s.available_permits(), 0);
t2.enter(|cx| {
// Try to acquire the second permit
assert_pending!(permit_2.poll_acquire(cx, &s));
});
permit_1.release(&s);
assert_eq!(s.available_permits(), 0);
assert!(t2.is_woken());
assert_ready_ok!(t2.enter(|cx| permit_2.poll_acquire(cx, &s)));
permit_2.release(&s);
assert_eq!(s.available_permits(), 1);
}
#[test]
fn zero_permits() {
let mut t1 = MockTask::new();
let s = Semaphore::new(0);
assert_eq!(s.available_permits(), 0);
let mut permit = Permit::new();
// Try to acquire the permit
t1.enter(|cx| {
assert_pending!(permit.poll_acquire(cx, &s));
});
s.add_permits(1);
assert!(t1.is_woken());
assert_ready_ok!(t1.enter(|cx| permit.poll_acquire(cx, &s)));
}
#[test]
#[should_panic]
fn validates_max_permits() {
use std::usize;
Semaphore::new((usize::MAX >> 2) + 1);
}
#[test]
fn close_semaphore_prevents_acquire() {
let mut t1 = MockTask::new();
let s = Semaphore::new(1);
s.close();
assert_eq!(1, s.available_permits());
let mut permit = Permit::new();
assert_ready_err!(t1.enter(|cx| permit.poll_acquire(cx, &s)));
assert_eq!(1, s.available_permits());
}
#[test]
fn close_semaphore_notifies_permit1() {
let mut t1 = MockTask::new();
let s = Semaphore::new(0);
let mut permit = Permit::new();
assert_pending!(t1.enter(|cx| permit.poll_acquire(cx, &s)));
s.close();
assert!(t1.is_woken());
assert_ready_err!(t1.enter(|cx| permit.poll_acquire(cx, &s)));
}
#[test]
fn close_semaphore_notifies_permit2() {
let mut t1 = MockTask::new();
let mut t2 = MockTask::new();
let mut t3 = MockTask::new();
let mut t4 = MockTask::new();
let s = Semaphore::new(2);
let mut permit1 = Permit::new();
let mut permit2 = Permit::new();
let mut permit3 = Permit::new();
let mut permit4 = Permit::new();
// Acquire a couple of permits
assert_ready_ok!(t1.enter(|cx| permit1.poll_acquire(cx, &s)));
assert_ready_ok!(t2.enter(|cx| permit2.poll_acquire(cx, &s)));
assert_pending!(t3.enter(|cx| permit3.poll_acquire(cx, &s)));
assert_pending!(t4.enter(|cx| permit4.poll_acquire(cx, &s)));
s.close();
assert!(t3.is_woken());
assert!(t4.is_woken());
assert_ready_err!(t3.enter(|cx| permit3.poll_acquire(cx, &s)));
assert_ready_err!(t4.enter(|cx| permit4.poll_acquire(cx, &s)));
assert_eq!(0, s.available_permits());
permit1.release(&s);
assert_eq!(1, s.available_permits());
assert_ready_err!(t1.enter(|cx| permit1.poll_acquire(cx, &s)));
permit2.release(&s);
assert_eq!(2, s.available_permits());
}
+264
View File
@@ -0,0 +1,264 @@
#![warn(rust_2018_idioms)]
use tokio::sync::watch;
use tokio_test::task::spawn;
use tokio_test::{assert_pending, assert_ready};
#[test]
fn single_rx_recv_ref() {
let (tx, mut rx) = watch::channel("one");
{
let mut t = spawn(rx.recv_ref());
let v = assert_ready!(t.poll()).unwrap();
assert_eq!(*v, "one");
}
{
let mut t = spawn(rx.recv_ref());
assert_pending!(t.poll());
tx.broadcast("two").unwrap();
assert!(t.is_woken());
let v = assert_ready!(t.poll()).unwrap();
assert_eq!(*v, "two");
}
{
let mut t = spawn(rx.recv_ref());
assert_pending!(t.poll());
drop(tx);
let res = assert_ready!(t.poll());
assert!(res.is_none());
}
}
#[test]
fn single_rx_recv() {
let (tx, mut rx) = watch::channel("one");
{
let mut t = spawn(rx.recv());
let v = assert_ready!(t.poll()).unwrap();
assert_eq!(v, "one");
}
{
let mut t = spawn(rx.recv());
assert_pending!(t.poll());
tx.broadcast("two").unwrap();
assert!(t.is_woken());
let v = assert_ready!(t.poll()).unwrap();
assert_eq!(v, "two");
}
{
let mut t = spawn(rx.recv());
assert_pending!(t.poll());
drop(tx);
let res = assert_ready!(t.poll());
assert!(res.is_none());
}
}
#[test]
fn stream_impl() {
use tokio::prelude::*;
let (tx, mut rx) = watch::channel("one");
{
let mut t = spawn(rx.next());
let v = assert_ready!(t.poll()).unwrap();
assert_eq!(v, "one");
}
{
let mut t = spawn(rx.next());
assert_pending!(t.poll());
tx.broadcast("two").unwrap();
assert!(t.is_woken());
let v = assert_ready!(t.poll()).unwrap();
assert_eq!(v, "two");
}
{
let mut t = spawn(rx.next());
assert_pending!(t.poll());
drop(tx);
let res = assert_ready!(t.poll());
assert!(res.is_none());
}
}
#[test]
fn multi_rx() {
let (tx, mut rx1) = watch::channel("one");
let mut rx2 = rx1.clone();
{
let mut t1 = spawn(rx1.recv_ref());
let mut t2 = spawn(rx2.recv_ref());
let res = assert_ready!(t1.poll());
assert_eq!(*res.unwrap(), "one");
let res = assert_ready!(t2.poll());
assert_eq!(*res.unwrap(), "one");
}
let mut t2 = spawn(rx2.recv_ref());
{
let mut t1 = spawn(rx1.recv_ref());
assert_pending!(t1.poll());
assert_pending!(t2.poll());
tx.broadcast("two").unwrap();
assert!(t1.is_woken());
assert!(t2.is_woken());
let res = assert_ready!(t1.poll());
assert_eq!(*res.unwrap(), "two");
}
{
let mut t1 = spawn(rx1.recv_ref());
assert_pending!(t1.poll());
tx.broadcast("three").unwrap();
assert!(t1.is_woken());
assert!(t2.is_woken());
let res = assert_ready!(t1.poll());
assert_eq!(*res.unwrap(), "three");
let res = assert_ready!(t2.poll());
assert_eq!(*res.unwrap(), "three");
}
drop(t2);
{
let mut t1 = spawn(rx1.recv_ref());
let mut t2 = spawn(rx2.recv_ref());
assert_pending!(t1.poll());
assert_pending!(t2.poll());
tx.broadcast("four").unwrap();
let res = assert_ready!(t1.poll());
assert_eq!(*res.unwrap(), "four");
drop(t1);
let mut t1 = spawn(rx1.recv_ref());
assert_pending!(t1.poll());
drop(tx);
assert!(t1.is_woken());
let res = assert_ready!(t1.poll());
assert!(res.is_none());
let res = assert_ready!(t2.poll());
assert_eq!(*res.unwrap(), "four");
drop(t2);
let mut t2 = spawn(rx2.recv_ref());
let res = assert_ready!(t2.poll());
assert!(res.is_none());
}
}
#[test]
fn rx_observes_final_value() {
// Initial value
let (tx, mut rx) = watch::channel("one");
drop(tx);
{
let mut t1 = spawn(rx.recv_ref());
let res = assert_ready!(t1.poll());
assert_eq!(*res.unwrap(), "one");
}
{
let mut t1 = spawn(rx.recv_ref());
let res = assert_ready!(t1.poll());
assert!(res.is_none());
}
// Sending a value
let (tx, mut rx) = watch::channel("one");
tx.broadcast("two").unwrap();
{
let mut t1 = spawn(rx.recv_ref());
let res = assert_ready!(t1.poll());
assert_eq!(*res.unwrap(), "two");
}
{
let mut t1 = spawn(rx.recv_ref());
assert_pending!(t1.poll());
tx.broadcast("three").unwrap();
drop(tx);
assert!(t1.is_woken());
let res = assert_ready!(t1.poll());
assert_eq!(*res.unwrap(), "three");
}
{
let mut t1 = spawn(rx.recv_ref());
let res = assert_ready!(t1.poll());
assert!(res.is_none());
}
}
#[test]
fn poll_close() {
let (mut tx, rx) = watch::channel("one");
{
let mut t = spawn(tx.closed());
assert_pending!(t.poll());
drop(rx);
assert!(t.is_woken());
assert_ready!(t.poll());
}
assert!(tx.broadcast("two").is_err());
}
+1 -1
View File
@@ -274,7 +274,7 @@ fn panic_in_task() {
#[test]
fn multi_threadpool() {
use tokio_sync::oneshot;
use tokio::sync::oneshot;
let pool1 = new_pool();
let pool2 = new_pool();
+2 -5
View File
@@ -129,7 +129,6 @@ fn deadline_future_elapses() {
});
}
#[cfg(feature = "async-traits")]
macro_rules! poll {
($task:ident, $stream:ident) => {{
use futures_core::Stream;
@@ -138,9 +137,8 @@ macro_rules! poll {
}
#[test]
#[cfg(feature = "async-traits")]
fn stream_and_timeout_in_future() {
use tokio_sync::mpsc;
use tokio::sync::mpsc;
let mut t = MockTask::new();
@@ -168,9 +166,8 @@ fn stream_and_timeout_in_future() {
}
#[test]
#[cfg(feature = "async-traits")]
fn idle_stream_timesout_periodically() {
use tokio_sync::mpsc;
use tokio::sync::mpsc;
let mut t = MockTask::new();