sync: polish and update API doc examples (#1398)

- Remove `poll_*` fns from some of the sync types.
- Move `AtomicWaker` and `Lock` to the root of the `sync` crate.

tokio-reactor/src/lib.rs

@@ -58,7 +58,7 @@ use std::task::Waker;
 use std::time::{Duration, Instant};
 use std::{fmt, usize};
 use tokio_executor::park::{Park, Unpark};
-use tokio_sync::task::AtomicWaker;
+use tokio_sync::AtomicWaker;
 
 /// The core reactor, or event loop.
 ///

tokio-sync/Cargo.toml

@@ -25,10 +25,10 @@ publish = false
 async-traits = ["futures-sink-preview"]
 
 [dependencies]
-async-util = { git = "https://github.com/tokio-rs/async" }
 fnv = "1.0.6"
-futures-core-preview = { version = "0.3.0-alpha.17" }
+futures-core-preview = { version = "= 0.3.0-alpha.17" }
-futures-sink-preview = { version = "0.3.0-alpha.17", optional = true }
+futures-sink-preview = { version = "= 0.3.0-alpha.17", optional = true }
+futures-util-preview = { version = "= 0.3.0-alpha.17" }
 
 [dev-dependencies]
 env_logger = { version = "0.5", default-features = false }

tokio-sync/src/lib.rs

@@ -27,10 +27,13 @@ macro_rules! if_fuzz {
     }}
 }
 
-pub mod lock;
+mod lock;
 mod loom;
 pub mod mpsc;
 pub mod oneshot;
 pub mod semaphore;
-pub mod task;
+mod task;
 pub mod watch;
+
+pub use lock::{Lock, LockGuard};
+pub use task::AtomicWaker;

tokio-sync/src/lock.rs

@@ -1,39 +1,29 @@
 //! An asynchronous `Mutex`-like type.
 //!
 //! This module provides [`Lock`], a type that acts similarly to an asynchronous `Mutex`, with one
-//! major difference: the [`LockGuard`] returned by `poll_lock` is not tied to the lifetime of the
+//! major difference: the [`LockGuard`] 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 futures::{try_ready, future, Poll, Async, Future, Stream};
+//! #![feature(async_await)]
-//! use tokio::sync::lock::{Lock, LockGuard};
 //!
-//! struct MyType<S> {
+//! use tokio::sync::Lock;
-//!     lock: Lock<S>,
-//! }
 //!
-//! impl<S> Future for MyType<S>
+//! #[tokio::main]
-//!   where S: Stream<Item = u32> + Send + 'static
+//! async fn main() {
-//! {
+//!     let mut data1 = Lock::new(0);
-//!     type Item = ();
+//!     let mut data2 = data1.clone();
-//!     type Error = ();
 //!
-//!     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+//!     tokio::spawn(async move {
-//!         match self.lock.poll_lock() {
+//!         let mut lock = data2.lock().await;
-//!             Async::Ready(mut guard) => {
+//!         *lock += 1;
-//!                 tokio::spawn(future::poll_fn(move || {
+//!     });
-//!                     let item = try_ready!(guard.poll().map_err(|_| ()));
+//!
-//!                     println!("item = {:?}", item);
+//!     let mut lock = data1.lock().await;
-//!                     Ok(().into())
+//!     *lock += 1;
-//!                 }));
-//!                 Ok(().into())
-//!             },
-//!             Async::NotReady => Ok(Async::NotReady)
-//!         }
-//!     }
 //! }
 //! ```
 //!
@@ -43,11 +33,10 @@
 use crate::semaphore;
 
 use futures_core::ready;
+use futures_util::future::poll_fn;
 use std::cell::UnsafeCell;
 use std::fmt;
-use std::future::Future;
 use std::ops::{Deref, DerefMut};
-use std::pin::Pin;
 use std::sync::Arc;
 use std::task::Poll::Ready;
 use std::task::{Context, Poll};
@@ -55,8 +44,8 @@ use std::task::{Context, Poll};
 /// 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 `poll_lock`, which guarantees that
+/// can only be accessed through the RAII guards returned from `lock`, which
-/// the data is only ever accessed when the mutex is locked.
+/// guarantees that the data is only ever accessed when the mutex is locked.
 #[derive(Debug)]
 pub struct Lock<T> {
     inner: Arc<State<T>>,
@@ -69,17 +58,11 @@ pub struct Lock<T> {
 /// internally keeps a reference-couned pointer to the original `Lock`, so even if the lock goes
 /// away, the guard remains valid.
 ///
-/// The lock is automatically released whenever the guard is dropped, at which point `poll_lock`
+/// The lock is automatically released whenever the guard is dropped, at which point `lock`
 /// will succeed yet again.
 #[derive(Debug)]
 pub struct LockGuard<T>(Lock<T>);
 
-/// A future that resolves to a `LockGuard`.
-#[derive(Debug)]
-pub struct LockFuture<'a, T> {
-    lock: &'a mut Lock<T>,
-}
-
 // As long as T: Send, it's fine to send and share Lock<T> between threads.
 // If T was not Send, sending and sharing a Lock<T> would be bad, since you can access T through
 // Lock<T>.
@@ -111,10 +94,7 @@ impl<T> Lock<T> {
         }
     }
 
-    /// Try to acquire the lock.
+    fn poll_lock(&mut self, cx: &mut Context<'_>) -> Poll<LockGuard<T>> {
-    ///
-    /// If the lock is already held, the current task is notified when it is released.
-    pub fn poll_lock(&mut self, cx: &mut Context<'_>) -> Poll<LockGuard<T>> {
         ready!(self.permit.poll_acquire(cx, &self.inner.s)).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.
@@ -131,8 +111,9 @@ impl<T> Lock<T> {
     }
 
     /// A future that resolves on acquiring the lock and returns the `LockGuard`.
-    pub fn lock(&mut self) -> LockFuture<'_, T> {
+    #[allow(clippy::needless_lifetimes)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3988
-        LockFuture { lock: self }
+    pub async fn lock(&mut self) -> LockGuard<T> {
+        poll_fn(|cx| self.poll_lock(cx)).await
     }
 }
 
@@ -193,12 +174,3 @@ impl<T: fmt::Display> fmt::Display for LockGuard<T> {
         fmt::Display::fmt(&**self, f)
     }
 }
-
-impl<T> Future for LockFuture<'_, T> {
-    type Output = LockGuard<T>;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let me = &mut *self;
-        Pin::new(&mut *me.lock).poll_lock(cx)
-    }
-}

tokio-sync/src/mpsc/bounded.rs

@@ -82,34 +82,27 @@ pub struct RecvError(());
 /// # Examples
 ///
 /// ```rust
-/// use tokio::sync::mpsc::channel;
+/// #![feature(async_await)]
-/// use tokio::prelude::*;
-/// use futures::future::lazy;
 ///
-/// # fn some_computation() -> impl Future<Item = (), Error = ()> + Send {
+/// use tokio::sync::mpsc;
-/// # futures::future::ok::<(), ()>(())
-/// # }
 ///
-/// tokio::run(lazy(|| {
+/// #[tokio::main]
-///     let (tx, rx) = channel(100);
+/// async fn main() {
+///     let (mut tx, mut rx) = mpsc::channel(100);
 ///
-///     tokio::spawn({
+///     tokio::spawn(async move {
-///         some_computation()
+///         for i in 0..10 {
-///             .and_then(|value| {
+///             if let Err(_) = tx.send(i).await {
-///                 tx.send(value)
+///                 println!("receiver dropped");
-///                     .map_err(|_| ())
+///                 return;
-///             })
+///             }
-///             .map(|_| ())
+///         }
-///             .map_err(|_| ())
 ///     });
 ///
-///     rx.for_each(|value| {
+///     while let Some(i) = rx.recv().await {
-///         println!("got value = {:?}", value);
+///         println!("got = {}", i);
-///         Ok(())
+///     }
-///     })
+/// }
-///     .map(|_| ())
-///     .map_err(|_| ())
-/// }));
 /// ```
 pub fn channel<T>(buffer: usize) -> (Sender<T>, Receiver<T>) {
     assert!(buffer > 0, "mpsc bounded channel requires buffer > 0");
@@ -134,7 +127,7 @@ impl<T> Receiver<T> {
     /// TODO: Dox
     #[allow(clippy::needless_lifetimes)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3988
     pub async fn recv(&mut self) -> Option<T> {
-        use async_util::future::poll_fn;
+        use futures_util::future::poll_fn;
 
         poll_fn(|cx| self.poll_recv(cx)).await
     }
@@ -202,12 +195,35 @@ impl<T> Sender<T> {
     ///
     /// # Examples
     ///
-    /// ```
+    /// In the following example, each call to `send` will block until the
-    /// unimplemented!();
+    /// previously sent value was received.
+    ///
+    /// ```rust
+    /// #![feature(async_await)]
+    ///
+    /// 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);
+    ///     }
+    /// }
     /// ```
     #[allow(clippy::needless_lifetimes)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3988
     pub async fn send(&mut self, value: T) -> Result<(), SendError> {
-        use async_util::future::poll_fn;
+        use futures_util::future::poll_fn;
 
         poll_fn(|cx| self.poll_ready(cx)).await?;
 

tokio-sync/src/mpsc/unbounded.rs

@@ -95,7 +95,7 @@ impl<T> UnboundedReceiver<T> {
     /// TODO: Dox
     #[allow(clippy::needless_lifetimes)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3988
     pub async fn recv(&mut self) -> Option<T> {
-        use async_util::future::poll_fn;
+        use futures_util::future::poll_fn;
 
         poll_fn(|cx| self.poll_recv(cx)).await
     }

tokio-sync/src/oneshot.rs

@@ -94,23 +94,25 @@ struct State(usize);
 /// # Examples
 ///
 /// ```
+/// #![feature(async_await)]
+///
 /// use tokio::sync::oneshot;
-/// use futures::Future;
-/// use std::thread;
 ///
-/// let (sender, receiver) = oneshot::channel::<i32>();
+/// #[tokio::main]
+/// async fn main() {
+///     let (tx, rx) = oneshot::channel();
 ///
-/// # let t =
+///     tokio::spawn(async move {
-/// thread::spawn(|| {
+///         if let Err(_) = tx.send(3) {
-///     let future = receiver.map(|i| {
+///             println!("the receiver dropped");
-///         println!("got: {:?}", i);
+///         }
 ///     });
-///     // ...
-/// # return future;
-/// });
 ///
-/// sender.send(3).unwrap();
+///     match rx.await {
-/// # t.join().unwrap().wait().unwrap();
+///         Ok(v) => println!("got = {:?}", v),
+///         Err(_) => println!("the sender dropped"),
+///     }
+/// }
 /// ```
 pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
     #[allow(deprecated)]
@@ -218,11 +220,25 @@ impl<T> Sender<T> {
     /// # Examples
     ///
     /// ```
-    /// unimplemented!();
+    /// #![feature(async_await)]
+    ///
+    /// 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");
+    /// }
     /// ```
     #[allow(clippy::needless_lifetimes)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3988
     pub async fn closed(&mut self) {
-        use async_util::future::poll_fn;
+        use futures_util::future::poll_fn;
 
         poll_fn(|cx| self.poll_closed(cx)).await
     }

tokio-sync/src/watch.rs

@@ -18,20 +18,22 @@
 //! # Examples
 //!
 //! ```
-//! use tokio::prelude::*;
+//! #![feature(async_await)]
+//!
 //! use tokio::sync::watch;
 //!
-//! # tokio::run(futures::future::lazy(|| {
+//! # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
-//! let (mut tx, rx) = watch::channel("hello");
+//!     let (mut tx, mut rx) = watch::channel("hello");
 //!
-//! tokio::spawn(rx.for_each(|value| {
+//!     tokio::spawn(async move {
-//!     println!("received = {:?}", value);
+//!         while let Some(value) = rx.recv().await {
-//!     Ok(())
+//!             println!("received = {:?}", value);
-//! }).map_err(|_| ()));
+//!         }
+//!     });
 //!
-//! tx.broadcast("world").unwrap();
+//!     tx.broadcast("world")?;
 //! # Ok(())
-//! # }));
+//! # }
 //! ```
 //!
 //! # Closing
@@ -59,6 +61,8 @@ use core::task::Poll::{Pending, Ready};
 use core::task::{Context, Poll};
 use fnv::FnvHashMap;
 use futures_core::ready;
+use futures_util::future::poll_fn;
+use futures_util::pin_mut;
 use std::ops;
 use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering::SeqCst;
@@ -165,20 +169,22 @@ const CLOSED: usize = 1;
 /// # Examples
 ///
 /// ```
-/// use tokio::prelude::*;
+/// #![feature(async_await)]
+///
 /// use tokio::sync::watch;
 ///
-/// # tokio::run(futures::future::lazy(|| {
+/// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
-/// let (mut tx, rx) = watch::channel("hello");
+///     let (mut tx, mut rx) = watch::channel("hello");
 ///
-/// tokio::spawn(rx.for_each(|value| {
+///     tokio::spawn(async move {
-///     println!("received = {:?}", value);
+///         while let Some(value) = rx.recv().await {
-///     Ok(())
+///             println!("received = {:?}", value);
-/// }).map_err(|_| ()));
+///         }
+///     });
 ///
-/// tx.broadcast("world").unwrap();
+///     tx.broadcast("world")?;
 /// # Ok(())
-/// # }));
+/// # }
 /// ```
 pub fn channel<T>(init: T) -> (Sender<T>, Receiver<T>) {
     const INIT_ID: u64 = 0;
@@ -241,39 +247,56 @@ impl<T> Receiver<T> {
     ///
     /// Only the **most recent** value is returned. If the receiver is falling
     /// behind the sender, intermediate values are dropped.
-    pub fn poll_ref(&mut self, cx: &mut Context<'_>) -> Poll<Option<Ref<'_, T>>> {
+    #[allow(clippy::needless_lifetimes)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3988
-        // Make sure the task is up to date
+    pub async fn recv_ref<'a>(&'a mut self) -> Option<Ref<'a, T>> {
-        self.inner.waker.register_by_ref(cx.waker());
+        let shared = &self.shared;
+        let inner = &self.inner;
+        let version = self.ver;
 
-        let state = self.shared.version.load(SeqCst);
+        match poll_fn(|cx| poll_lock(cx, shared, inner, version)).await {
-        let version = state & !CLOSED;
+            Some((lock, version)) => {
-
+                self.ver = version;
-        if version != self.ver {
+                Some(lock)
-            // Track the latest version
+            }
-            self.ver = version;
+            None => None,
-
-            let inner = self.shared.value.read().unwrap();
-
-            return Ready(Some(Ref { inner }));
         }
-
-        if CLOSED == state & CLOSED {
-            // The `Store` handle has been dropped.
-            return Ready(None);
-        }
-
-        Pending
     }
 }
 
+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 `poll_ref`.
+    /// This is equivalent to calling `clone()` on the value returned by
-    #[allow(clippy::map_clone)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3274
+    /// `recv()`.
-    pub fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<T>> {
+    #[allow(clippy::needless_lifetimes, clippy::map_clone)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3988
-        let item = ready!(self.poll_ref(cx));
+    pub async fn recv(&mut self) -> Option<T> {
-        Ready(item.map(|v_ref| v_ref.clone()))
+        self.recv_ref().await.map(|v_ref| v_ref.clone())
     }
 }
 
@@ -282,8 +305,13 @@ 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(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
-        let item = ready!(self.poll_ref(cx));
+        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()))
     }
 }
@@ -354,11 +382,16 @@ impl<T> Sender<T> {
         Ok(())
     }
 
-    /// Returns `Ready` when all receivers have dropped.
+    /// 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 fn poll_close(&mut self, cx: &mut Context<'_>) -> Poll<()> {
+    #[allow(clippy::needless_lifetimes)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3988
+    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());

tokio-sync/tests/atomic_waker.rs

@@ -1,7 +1,7 @@
 #![deny(warnings, rust_2018_idioms)]
 
 use std::task::Waker;
-use tokio_sync::task::AtomicWaker;
+use tokio_sync::AtomicWaker;
 use tokio_test::task::MockTask;
 
 trait AssertSend: Send {}

tokio-sync/tests/fuzz_atomic_waker.rs

@@ -8,7 +8,7 @@ extern crate loom;
 mod atomic_waker;
 use crate::atomic_waker::AtomicWaker;
 
-use async_util::future::poll_fn;
+use futures_util::future::poll_fn;
 use loom::futures::block_on;
 use loom::sync::atomic::AtomicUsize;
 use loom::thread;

tokio-sync/tests/fuzz_mpsc.rs

@@ -18,8 +18,7 @@ mod mpsc;
 #[allow(warnings)]
 mod semaphore;
 
-// use futures::{future::poll_fn, Stream};
+use futures_util::future::poll_fn;
-use async_util::future::poll_fn;
 use loom::futures::block_on;
 use loom::thread;
 

tokio-sync/tests/fuzz_semaphore.rs

@@ -9,8 +9,8 @@ mod semaphore;
 
 use crate::semaphore::*;
 
-use async_util::future::poll_fn;
 use futures_core::ready;
+use futures_util::future::poll_fn;
 use loom::futures::block_on;
 use loom::thread;
 use std::future::Future;

tokio-sync/tests/lock.rs

@@ -1,55 +1,57 @@
 #![deny(warnings, rust_2018_idioms)]
 
-use pin_utils::pin_mut;
+use tokio_sync::Lock;
-use tokio_sync::lock::Lock;
+use tokio_test::task::spawn;
-use tokio_test::task::MockTask;
 use tokio_test::{assert_pending, assert_ready};
 
 #[test]
 fn straight_execution() {
-    let mut task = MockTask::new();
     let mut l = Lock::new(100);
 
-    // We can immediately acquire the lock and take the value
+    {
-    task.enter(|cx| {
+        let mut t = spawn(l.lock());
-        let mut g = assert_ready!(l.poll_lock(cx));
+        let mut g = assert_ready!(t.poll());
         assert_eq!(&*g, &100);
         *g = 99;
-        drop(g);
+    }
-
+    {
-        let mut g = assert_ready!(l.poll_lock(cx));
+        let mut t = spawn(l.lock());
+        let mut g = assert_ready!(t.poll());
         assert_eq!(&*g, &99);
         *g = 98;
-        drop(g);
+    }
-
+    {
-        let mut g = assert_ready!(l.poll_lock(cx));
+        let mut t = spawn(l.lock());
+        let mut g = assert_ready!(t.poll());
         assert_eq!(&*g, &98);
 
         // We can continue to access the guard even if the lock is dropped
+        drop(t);
         drop(l);
         *g = 97;
         assert_eq!(&*g, &97);
-    });
+    }
 }
 
 #[test]
 fn readiness() {
-    let mut t1 = MockTask::new();
+    let mut l1 = Lock::new(100);
-    let mut t2 = MockTask::new();
+    let mut l2 = l1.clone();
+    let mut t1 = spawn(l1.lock());
+    let mut t2 = spawn(l2.lock());
 
-    let mut l = Lock::new(100);
+    let g = assert_ready!(t1.poll());
-
-    let g = assert_ready!(t1.enter(|cx| l.poll_lock(cx)));
 
     // We can't now acquire the lease since it's already held in g
-    assert_pending!(t2.enter(|cx| l.poll_lock(cx)));
+    assert_pending!(t2.poll());
 
     // But once g unlocks, we can acquire it
     drop(g);
     assert!(t2.is_woken());
-    assert_ready!(t2.enter(|cx| l.poll_lock(cx)));
+    assert_ready!(t2.poll());
 }
 
+/*
 #[test]
 #[ignore]
 fn lock() {
@@ -79,3 +81,4 @@ fn lock() {
     let result = assert_ready!(task.poll(&mut l));
     assert!(*result);
 }
+*/

tokio-sync/tests/watch.rs

@@ -1,7 +1,7 @@
 #![deny(warnings, rust_2018_idioms)]
 
 use tokio_sync::watch;
-use tokio_test::task::MockTask;
+use tokio_test::task::spawn;
 use tokio_test::{assert_pending, assert_ready};
 
 /*
@@ -27,110 +27,111 @@ macro_rules! assert_not_ready {
 */
 
 #[test]
-fn single_rx_poll_ref() {
+fn single_rx_recv_ref() {
     let (tx, mut rx) = watch::channel("one");
-    let mut task = MockTask::new();
 
-    task.enter(|cx| {
+    {
-        {
+        let mut t = spawn(rx.recv_ref());
-            let v = assert_ready!(rx.poll_ref(cx)).unwrap();
+        let v = assert_ready!(t.poll()).unwrap();
-            assert_eq!(*v, "one");
+        assert_eq!(*v, "one");
-        }
+    }
-        assert_pending!(rx.poll_ref(cx));
-    });
 
-    tx.broadcast("two").unwrap();
+    {
+        let mut t = spawn(rx.recv_ref());
 
-    assert!(task.is_woken());
+        assert_pending!(t.poll());
 
-    task.enter(|cx| {
+        tx.broadcast("two").unwrap();
-        {
-            let v = assert_ready!(rx.poll_ref(cx)).unwrap();
-            assert_eq!(*v, "two");
-        }
-        assert_pending!(rx.poll_ref(cx));
-    });
 
-    drop(tx);
+        assert!(t.is_woken());
 
-    assert!(task.is_woken());
+        let v = assert_ready!(t.poll()).unwrap();
+        assert_eq!(*v, "two");
+    }
 
-    task.enter(|cx| {
+    {
-        let res = assert_ready!(rx.poll_ref(cx));
+        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_poll_next() {
+fn single_rx_recv() {
     let (tx, mut rx) = watch::channel("one");
-    let mut task = MockTask::new();
 
-    task.enter(|cx| {
+    {
-        let v = assert_ready!(rx.poll_next(cx)).unwrap();
+        let mut t = spawn(rx.recv());
+        let v = assert_ready!(t.poll()).unwrap();
         assert_eq!(v, "one");
-        assert_pending!(rx.poll_ref(cx));
+    }
-    });
 
-    tx.broadcast("two").unwrap();
+    {
+        let mut t = spawn(rx.recv());
 
-    assert!(task.is_woken());
+        assert_pending!(t.poll());
 
-    task.enter(|cx| {
+        tx.broadcast("two").unwrap();
-        let v = assert_ready!(rx.poll_next(cx)).unwrap();
+
+        assert!(t.is_woken());
+
+        let v = assert_ready!(t.poll()).unwrap();
         assert_eq!(v, "two");
-        assert_pending!(rx.poll_ref(cx));
+    }
-    });
 
-    drop(tx);
+    {
+        let mut t = spawn(rx.recv());
 
-    assert!(task.is_woken());
+        assert_pending!(t.poll());
 
-    task.enter(|cx| {
+        drop(tx);
-        let res = assert_ready!(rx.poll_next(cx));
+
+        let res = assert_ready!(t.poll());
         assert!(res.is_none());
-    });
+    }
 }
 
 #[test]
 #[cfg(feature = "async-traits")]
 fn stream_impl() {
-    use futures_core::Stream;
+    use tokio::prelude::*;
-    use pin_utils::pin_mut;
 
-    let (tx, rx) = watch::channel("one");
+    let (tx, mut rx) = watch::channel("one");
-    let mut task = MockTask::new();
 
-    pin_mut!(rx);
+    {
+        let mut t = spawn(rx.next());
+        let v = assert_ready!(t.poll()).unwrap();
+        assert_eq!(v, "one");
+    }
 
-    task.enter(|cx| {
+    {
-        {
+        let mut t = spawn(rx.next());
-            let v = assert_ready!(Stream::poll_next(rx.as_mut(), cx)).unwrap();
-            assert_eq!(v, "one");
-        }
-        assert_pending!(rx.poll_ref(cx));
-    });
 
-    tx.broadcast("two").unwrap();
+        assert_pending!(t.poll());
 
-    assert!(task.is_woken());
+        tx.broadcast("two").unwrap();
 
-    task.enter(|cx| {
+        assert!(t.is_woken());
-        {
-            let v = assert_ready!(Stream::poll_next(rx.as_mut(), cx)).unwrap();
-            assert_eq!(v, "two");
-        }
-        assert_pending!(rx.poll_ref(cx));
-    });
 
-    drop(tx);
+        let v = assert_ready!(t.poll()).unwrap();
+        assert_eq!(v, "two");
+    }
 
-    assert!(task.is_woken());
+    {
+        let mut t = spawn(rx.next());
 
-    task.enter(|cx| {
+        assert_pending!(t.poll());
-        let res = assert_ready!(Stream::poll_next(rx, cx));
+
+        drop(tx);
+
+        let res = assert_ready!(t.poll());
         assert!(res.is_none());
-    });
+    }
 }
 
 #[test]
@@ -138,67 +139,83 @@ fn multi_rx() {
     let (tx, mut rx1) = watch::channel("one");
     let mut rx2 = rx1.clone();
 
-    let mut task1 = MockTask::new();
+    {
-    let mut task2 = MockTask::new();
+        let mut t1 = spawn(rx1.recv_ref());
+        let mut t2 = spawn(rx2.recv_ref());
 
-    task1.enter(|cx| {
+        let res = assert_ready!(t1.poll());
-        let res = assert_ready!(rx1.poll_ref(cx));
         assert_eq!(*res.unwrap(), "one");
-    });
 
-    task2.enter(|cx| {
+        let res = assert_ready!(t2.poll());
-        let res = assert_ready!(rx2.poll_ref(cx));
         assert_eq!(*res.unwrap(), "one");
-    });
+    }
 
-    tx.broadcast("two").unwrap();
+    let mut t2 = spawn(rx2.recv_ref());
 
-    assert!(task1.is_woken());
+    {
-    assert!(task2.is_woken());
+        let mut t1 = spawn(rx1.recv_ref());
 
-    task1.enter(|cx| {
+        assert_pending!(t1.poll());
-        let res = assert_ready!(rx1.poll_ref(cx));
+        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");
-    });
+    }
 
-    tx.broadcast("three").unwrap();
+    {
+        let mut t1 = spawn(rx1.recv_ref());
 
-    assert!(task1.is_woken());
+        assert_pending!(t1.poll());
-    assert!(task2.is_woken());
 
-    task1.enter(|cx| {
+        tx.broadcast("three").unwrap();
-        let res = assert_ready!(rx1.poll_ref(cx));
+
+        assert!(t1.is_woken());
+        assert!(t2.is_woken());
+
+        let res = assert_ready!(t1.poll());
         assert_eq!(*res.unwrap(), "three");
-    });
 
-    task2.enter(|cx| {
+        let res = assert_ready!(t2.poll());
-        let res = assert_ready!(rx2.poll_ref(cx));
         assert_eq!(*res.unwrap(), "three");
-    });
+    }
 
-    tx.broadcast("four").unwrap();
+    drop(t2);
 
-    task1.enter(|cx| {
+    {
-        let res = assert_ready!(rx1.poll_ref(cx));
+        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);
 
-    drop(tx);
+        let mut t1 = spawn(rx1.recv_ref());
+        assert_pending!(t1.poll());
 
-    task1.enter(|cx| {
+        drop(tx);
-        let res = assert_ready!(rx1.poll_ref(cx));
+
+        assert!(t1.is_woken());
+        let res = assert_ready!(t1.poll());
         assert!(res.is_none());
-    });
 
-    task2.enter(|cx| {
+        let res = assert_ready!(t2.poll());
-        let res = assert_ready!(rx2.poll_ref(cx));
         assert_eq!(*res.unwrap(), "four");
-    });
 
-    task2.enter(|cx| {
+        drop(t2);
-        let res = assert_ready!(rx2.poll_ref(cx));
+        let mut t2 = spawn(rx2.recv_ref());
+        let res = assert_ready!(t2.poll());
         assert!(res.is_none());
-    });
+    }
 }
 
 #[test]
@@ -206,67 +223,65 @@ fn rx_observes_final_value() {
     // Initial value
 
     let (tx, mut rx) = watch::channel("one");
-    let mut task = MockTask::new();
-
     drop(tx);
 
-    task.enter(|cx| {
+    {
-        let res = assert_ready!(rx.poll_ref(cx));
+        let mut t1 = spawn(rx.recv_ref());
-        assert!(res.is_some());
+        let res = assert_ready!(t1.poll());
         assert_eq!(*res.unwrap(), "one");
-    });
+    }
 
-    task.enter(|cx| {
+    {
-        let res = assert_ready!(rx.poll_ref(cx));
+        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");
-    let mut task = MockTask::new();
 
     tx.broadcast("two").unwrap();
 
-    task.enter(|cx| {
+    {
-        {
+        let mut t1 = spawn(rx.recv_ref());
-            let res = assert_ready!(rx.poll_ref(cx));
+        let res = assert_ready!(t1.poll());
-            assert!(res.is_some());
+        assert_eq!(*res.unwrap(), "two");
-            assert_eq!(*res.unwrap(), "two");
+    }
-        }
 
-        assert_pending!(rx.poll_ref(cx));
+    {
-    });
+        let mut t1 = spawn(rx.recv_ref());
+        assert_pending!(t1.poll());
 
-    tx.broadcast("three").unwrap();
+        tx.broadcast("three").unwrap();
-    drop(tx);
+        drop(tx);
 
-    assert!(task.is_woken());
+        assert!(t1.is_woken());
 
-    task.enter(|cx| {
+        let res = assert_ready!(t1.poll());
-        let res = assert_ready!(rx.poll_ref(cx));
-        assert!(res.is_some());
         assert_eq!(*res.unwrap(), "three");
-    });
+    }
 
-    task.enter(|cx| {
+    {
-        let res = assert_ready!(rx.poll_ref(cx));
+        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 task = MockTask::new();
 
-    assert_pending!(task.enter(|cx| tx.poll_close(cx)));
+    {
+        let mut t = spawn(tx.closed());
+        assert_pending!(t.poll());
 
-    drop(rx);
+        drop(rx);
 
-    assert!(task.is_woken());
+        assert!(t.is_woken());
-
+        assert_ready!(t.poll());
-    assert_ready!(task.enter(|cx| tx.poll_close(cx)));
+    }
 
     assert!(tx.broadcast("two").is_err());
 }

tokio-test/src/task.rs

@@ -22,6 +22,7 @@ use tokio_executor::enter;
 use pin_convert::AsPinMut;
 use std::future::Future;
 use std::mem;
+use std::pin::Pin;
 use std::sync::{Arc, Condvar, Mutex};
 use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
 
@@ -33,6 +34,21 @@ pub struct MockTask {
     waker: Arc<ThreadWaker>,
 }
 
+/// Future spawned on a mock task
+#[derive(Debug)]
+pub struct Spawn<T> {
+    task: MockTask,
+    future: Pin<Box<T>>,
+}
+
+/// TOOD: dox
+pub fn spawn<T>(task: T) -> Spawn<T> {
+    Spawn {
+        task: MockTask::new(),
+        future: Box::pin(task),
+    }
+}
+
 #[derive(Debug)]
 struct ThreadWaker {
     state: Mutex<usize>,
@@ -43,6 +59,27 @@ const IDLE: usize = 0;
 const WAKE: usize = 1;
 const SLEEP: usize = 2;
 
+impl<T: Future> Spawn<T> {
+    /// Poll a future
+    pub fn poll(&mut self) -> Poll<T::Output> {
+        let fut = self.future.as_mut();
+        self.task.enter(|cx| fut.poll(cx))
+    }
+
+    /// Returns `true` if the inner future has received a wake notification
+    /// since the last call to `enter`.
+    pub fn is_woken(&self) -> bool {
+        self.task.is_woken()
+    }
+
+    /// Returns the number of references to the task waker
+    ///
+    /// The task itself holds a reference. The return value will never be zero.
+    pub fn waker_ref_count(&self) -> usize {
+        self.task.waker_ref_count()
+    }
+}
+
 impl MockTask {
     /// Create a new mock task
     pub fn new() -> Self {

tokio-threadpool/src/shutdown.rs

@@ -1,7 +1,7 @@
 use crate::task::Task;
 use crate::worker;
 
-use tokio_sync::task::AtomicWaker;
+use tokio_sync::AtomicWaker;
 
 use crossbeam_deque::Injector;
 use std::future::Future;

tokio-timer/src/timer/entry.rs

@@ -10,7 +10,7 @@ use std::sync::{Arc, Weak};
 use std::task::{self, Poll};
 use std::time::{Duration, Instant};
 use std::u64;
-use tokio_sync::task::AtomicWaker;
+use tokio_sync::AtomicWaker;
 
 /// Internal state shared between a `Delay` instance and the timer.
 ///

tokio/examples/chat.rs

@@ -33,7 +33,7 @@ use tokio::{
     self,
     codec::{Framed, LinesCodec, LinesCodecError},
     net::{TcpListener, TcpStream},
-    sync::{lock::Lock, mpsc},
+    sync::{mpsc, Lock},
 };
 
 #[tokio::main]

tokio/src/sync.rs

@@ -13,4 +13,5 @@
 //! - [watch](watch/index.html), a single-producer, multi-consumer channel that
 //!   only stores the **most recently** sent value.
 
-pub use tokio_sync::{lock, mpsc, oneshot, watch};
+pub use tokio_sync::{mpsc, oneshot, watch};
+pub use tokio_sync::{Lock, LockGuard};