From b479ce78d3e05f689d2a7c501942d2de67bed722 Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Mon, 27 Aug 2018 12:24:51 -0700 Subject: [PATCH] add experimental async/await support. (#582) This patch adds experimental async/await support to Tokio. It does this by adding feature flags to existing libs only where necessary in order to add nightly specific code (mostly `Unpin` implementations). It then provides a new crate: `tokio-async-await` which is a shim layer on top of `tokio`. The `tokio-async-await` crate is expected to look exactly like `tokio` does, but with async / await support. This strategy reduces the amount of cfg guarding in the main libraries. This patch also adds `tokio-channel`, which is copied from futures-rs 0.1 and adds the necessary `Unpin` implementations. In general, futures 0.1 is mostly unmaintained, so it will make sense for Tokio to take over maintainership of key components regardless of async / await support. --- Cargo.toml | 1 + tokio-async-await/Cargo.toml | 45 + tokio-async-await/LICENSE | 52 + tokio-async-await/README.md | 57 + tokio-async-await/examples/chat.rs | 133 +++ tokio-async-await/examples/echo_client.rs | 53 + tokio-async-await/examples/echo_server.rs | 45 + tokio-async-await/examples/hyper.rs | 25 + tokio-async-await/src/async_await/await.rs | 14 + .../src/async_await/compat/backward.rs | 112 ++ .../src/async_await/compat/forward.rs | 69 ++ .../src/async_await/compat/mod.rs | 4 + tokio-async-await/src/async_await/io/flush.rs | 32 + tokio-async-await/src/async_await/io/mod.rs | 192 ++++ tokio-async-await/src/async_await/io/read.rs | 38 + .../src/async_await/io/read_exact.rs | 56 + tokio-async-await/src/async_await/io/write.rs | 38 + .../src/async_await/io/write_all.rs | 57 + tokio-async-await/src/async_await/mod.rs | 8 + tokio-async-await/src/async_await/sink/mod.rs | 26 + .../src/async_await/sink/send.rs | 59 ++ .../src/async_await/stream/mod.rs | 40 + .../src/async_await/stream/next.rs | 31 + tokio-async-await/src/lib.rs | 102 ++ tokio-channel/CHANGELOG.md | 0 tokio-channel/Cargo.toml | 25 + tokio-channel/LICENSE | 51 + tokio-channel/README.md | 0 tokio-channel/src/async_await.rs | 10 + tokio-channel/src/lib.rs | 22 + tokio-channel/src/lock.rs | 105 ++ tokio-channel/src/mpsc/mod.rs | 989 ++++++++++++++++++ tokio-channel/src/mpsc/queue.rs | 151 +++ tokio-channel/src/oneshot.rs | 426 ++++++++ tokio-channel/tests/mpsc-close.rs | 22 + tokio-channel/tests/mpsc.rs | 481 +++++++++ tokio-channel/tests/oneshot.rs | 124 +++ tokio-channel/tests/support/mod.rs | 16 + tokio-reactor/CHANGELOG.md | 4 + tokio-reactor/Cargo.toml | 9 +- tokio-reactor/src/async_await.rs | 5 + tokio-reactor/src/lib.rs | 10 +- 42 files changed, 3736 insertions(+), 3 deletions(-) create mode 100644 tokio-async-await/Cargo.toml create mode 100644 tokio-async-await/LICENSE create mode 100644 tokio-async-await/README.md create mode 100644 tokio-async-await/examples/chat.rs create mode 100644 tokio-async-await/examples/echo_client.rs create mode 100644 tokio-async-await/examples/echo_server.rs create mode 100644 tokio-async-await/examples/hyper.rs create mode 100644 tokio-async-await/src/async_await/await.rs create mode 100644 tokio-async-await/src/async_await/compat/backward.rs create mode 100644 tokio-async-await/src/async_await/compat/forward.rs create mode 100644 tokio-async-await/src/async_await/compat/mod.rs create mode 100644 tokio-async-await/src/async_await/io/flush.rs create mode 100644 tokio-async-await/src/async_await/io/mod.rs create mode 100644 tokio-async-await/src/async_await/io/read.rs create mode 100644 tokio-async-await/src/async_await/io/read_exact.rs create mode 100644 tokio-async-await/src/async_await/io/write.rs create mode 100644 tokio-async-await/src/async_await/io/write_all.rs create mode 100644 tokio-async-await/src/async_await/mod.rs create mode 100644 tokio-async-await/src/async_await/sink/mod.rs create mode 100644 tokio-async-await/src/async_await/sink/send.rs create mode 100644 tokio-async-await/src/async_await/stream/mod.rs create mode 100644 tokio-async-await/src/async_await/stream/next.rs create mode 100644 tokio-async-await/src/lib.rs create mode 100644 tokio-channel/CHANGELOG.md create mode 100644 tokio-channel/Cargo.toml create mode 100644 tokio-channel/LICENSE create mode 100644 tokio-channel/README.md create mode 100644 tokio-channel/src/async_await.rs create mode 100644 tokio-channel/src/lib.rs create mode 100644 tokio-channel/src/lock.rs create mode 100644 tokio-channel/src/mpsc/mod.rs create mode 100644 tokio-channel/src/mpsc/queue.rs create mode 100644 tokio-channel/src/oneshot.rs create mode 100644 tokio-channel/tests/mpsc-close.rs create mode 100644 tokio-channel/tests/mpsc.rs create mode 100644 tokio-channel/tests/oneshot.rs create mode 100644 tokio-channel/tests/support/mod.rs create mode 100644 tokio-reactor/src/async_await.rs diff --git a/Cargo.toml b/Cargo.toml index 41501063a..82d3902df 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,6 +24,7 @@ keywords = ["io", "async", "non-blocking", "futures"] members = [ "./", + "tokio-channel", "tokio-codec", "tokio-current-thread", "tokio-executor", diff --git a/tokio-async-await/Cargo.toml b/tokio-async-await/Cargo.toml new file mode 100644 index 000000000..18ba4e3f8 --- /dev/null +++ b/tokio-async-await/Cargo.toml @@ -0,0 +1,45 @@ +cargo-features = ["edition", "rename-dependency"] + +[package] +name = "tokio-async-await" +edition = "2018" + +# When releasing to crates.io: +# - Update html_root_url. +version = "0.1.0" +authors = ["Carl Lerche "] +license = "MIT" +repository = "https://github.com/tokio-rs/tokio" +homepage = "https://tokio.rs" +documentation = "https://docs.rs/tokio-async-await/0.1.0" +description = """ +Experimental async/await support for Tokio +""" +categories = ["asynchronous"] + +[workspace] + +[lib] +name = "tokio" + +[dependencies] +futures = "0.1.23" +tokio_main = { package = "tokio", version = "0.1.7", path = ".." } +tokio-io = { version = "0.1.7", path = "../tokio-io" } +tokio-channel = { version = "0.1.0", path = "../tokio-channel", features = ["async-await-preview"] } +tokio-reactor = { version = "0.1.5", path = "../tokio-reactor", features = ["async-await-preview"] } +futures-core-preview = { version = "0.3.0-alpha.2" } +futures-util-preview = { version = "0.3.0-alpha.2" } + +[dev-dependencies] +bytes = "0.4.9" +tokio-codec = { version = "0.1.0", path = "../tokio-codec" } +hyper = "0.12.8" + +[patch.crates-io] +tokio = { path = "../" } +tokio-executor = { path = "../tokio-executor" } +tokio-io = { path = "../tokio-io" } +tokio-reactor = { path = "../tokio-reactor" } +tokio-tcp = { path = "../tokio-tcp" } +tokio-timer = { path = "../tokio-timer" } diff --git a/tokio-async-await/LICENSE b/tokio-async-await/LICENSE new file mode 100644 index 000000000..e4657ab2b --- /dev/null +++ b/tokio-async-await/LICENSE @@ -0,0 +1,52 @@ +Copyright (c) 2018 Tokio Contributors + +Permission is hereby granted, free of charge, to any +person obtaining a copy of this software and associated +documentation files (the "Software"), to deal in the +Software without restriction, including without +limitation the rights to use, copy, modify, merge, +publish, distribute, sublicense, and/or sell copies of +the Software, and to permit persons to whom the Software +is furnished to do so, subject to the following +conditions: + +The above copyright notice and this permission notice +shall be included in all copies or substantial portions +of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF +ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED +TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT +SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR +IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +DEALINGS IN THE SOFTWARE. + +Copyright (c) 2016 futures-rs authors + +Permission is hereby granted, free of charge, to any +person obtaining a copy of this software and associated +documentation files (the "Software"), to deal in the +Software without restriction, including without +limitation the rights to use, copy, modify, merge, +publish, distribute, sublicense, and/or sell copies of +the Software, and to permit persons to whom the Software +is furnished to do so, subject to the following +conditions: + +The above copyright notice and this permission notice +shall be included in all copies or substantial portions +of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF +ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED +TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT +SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR +IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +DEALINGS IN THE SOFTWARE. + diff --git a/tokio-async-await/README.md b/tokio-async-await/README.md new file mode 100644 index 000000000..24f7ac01e --- /dev/null +++ b/tokio-async-await/README.md @@ -0,0 +1,57 @@ +# Tokio async/await preview + +This crate provides a preview of Tokio with async / await support. It is a shim +layer on top of `tokio`. + +**This crate requires Rust nightly and does not provide API stability +guarantees. You are living on the here.** + +## Usage + +To use this crate, you need need to start with a Rust 2018 edition crate. + +Add this to your `Cargo.toml`: + +```toml +# At the very top of the file +cargo-features = ["edition"] + +# In the `[packages]` section +edition = "2018" + +# In the `[dependencies]` section +tokio-async-await = "0.1.0" +``` + +Then, get started. In your application, add: + +```rust +// The nightly features that are commonly needed with async / await +#![feature(await_macro, async_await, futures_api)] + +// This pulls in the `tokio-async-await` crate. While Rust 2018 doesn't require +// `extern crate`, we need to pull in the macros. +#[macro_use] +extern crate tokio; + +fn main() { + // And we are async... + tokio::run_async(async { + println!("Hello"); + }); +} +``` + +Because nightly is required, run the app with `cargo +nightly run` + +Check the [examples](examples) directory for more. + +## License + +This project is licensed under the [MIT license](LICENSE). + +### Contribution + +Unless you explicitly state otherwise, any contribution intentionally submitted +for inclusion in Tokio by you, shall be licensed as MIT, without any additional +terms or conditions. diff --git a/tokio-async-await/examples/chat.rs b/tokio-async-await/examples/chat.rs new file mode 100644 index 000000000..672e245f6 --- /dev/null +++ b/tokio-async-await/examples/chat.rs @@ -0,0 +1,133 @@ +#![feature(await_macro, async_await, futures_api)] + +#[macro_use] +extern crate tokio; + +use tokio::codec::{LinesCodec, Decoder}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::prelude::*; +use tokio::sync::mpsc; + +use std::collections::HashMap; +use std::io; +use std::net::SocketAddr; +use std::sync::{Arc, Mutex}; + +/// Shorthand for the transmit half of the message channel. +type Tx = mpsc::UnboundedSender; + +struct Shared { + peers: HashMap, +} + +impl Shared { + /// Create a new, empty, instance of `Shared`. + fn new() -> Self { + Shared { + peers: HashMap::new(), + } + } +} + +async fn process(stream: TcpStream, state: Arc>) -> io::Result<()> { + let addr = stream.peer_addr().unwrap(); + let mut lines = LinesCodec::new().framed(stream); + + // Extract the peer's name + let name = match await!(lines.next()) { + Some(name) => name?, + None => { + // Disconnected early + return Ok(()); + } + }; + + println!("`{}` is joining the chat", name); + + let (tx, mut rx) = mpsc::unbounded(); + + // Register the socket + state.lock().unwrap() + .peers.insert(addr, tx); + + // Split the `lines` handle into send and recv handles. This allows spawning + // separate tasks. + let (mut lines_tx, mut lines_rx) = lines.split(); + + // Spawn a task that receives all lines broadcasted to us from other peers + // and writes it to the client. + tokio::spawn_async(async move { + while let Some(line) = await!(rx.next()) { + let line = line.unwrap(); + await!(lines_tx.send_async(line)); + } + }); + + // Use the current task to read lines from the socket and broadcast them to + // other peers. + while let Some(message) = await!(lines_rx.next()) { + // TODO: Error handling + let message = message.unwrap(); + + let mut line = name.clone(); + line.push_str(": "); + line.push_str(&message); + line.push_str("\r\n"); + + let state = state.lock().unwrap(); + + for (peer_addr, tx) in &state.peers { + if *peer_addr != addr { + // TODO: Error handling + tx.unbounded_send(line.clone()).unwrap(); + } + } + } + + // Remove the client from the shared state. Doing so will also result in the + // tx task to terminate. + state.lock().unwrap() + .peers.remove(&addr) + .expect("bug"); + + Ok(()) +} + +fn main() { + // Create the shared state. This is how all the peers communicate. + // + // The server task will hold a handle to this. For every new client, the + // `state` handle is cloned and passed into the task that processes the + // client connection. + let state = Arc::new(Mutex::new(Shared::new())); + + let addr = "127.0.0.1:6142".parse().unwrap(); + + // Bind a TCP listener to the socket address. + // + // Note that this is the Tokio TcpListener, which is fully async. + let listener = TcpListener::bind(&addr).unwrap(); + + println!("server running on localhost:6142"); + + // Start the Tokio runtime. + tokio::run_async(async move { + let mut incoming = listener.incoming(); + + while let Some(stream) = await!(incoming.next()) { + let stream = match stream { + Ok(stream) => stream, + Err(_) => continue, + }; + + let state = state.clone(); + + tokio::spawn_async(async move { + if let Err(_) = await!(process(stream, state)) { + eprintln!("failed to process connection"); + } + }); + } + }); +} + diff --git a/tokio-async-await/examples/echo_client.rs b/tokio-async-await/examples/echo_client.rs new file mode 100644 index 000000000..614fb50ac --- /dev/null +++ b/tokio-async-await/examples/echo_client.rs @@ -0,0 +1,53 @@ +#![feature(await_macro, async_await, futures_api)] + +#[macro_use] +extern crate tokio; + +use tokio::net::TcpStream; +use tokio::prelude::*; + +use std::io; +use std::net::SocketAddr; + +const MESSAGES: &[&str] = &[ + "hello", + "world", + "one two three", +]; + +async fn run_client(addr: &SocketAddr) -> io::Result<()> { + let mut stream = await!(TcpStream::connect(addr))?; + + // Buffer to read into + let mut buf = [0; 128]; + + for msg in MESSAGES { + println!(" > write = {:?}", msg); + + // Write the message to the server + await!(stream.write_all_async(msg.as_bytes()))?; + + // Read the message back from the server + await!(stream.read_exact_async(&mut buf[..msg.len()]))?; + + assert_eq!(&buf[..msg.len()], msg.as_bytes()); + } + + Ok(()) +} + +fn main() { + use std::env; + + let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string()); + let addr = addr.parse::().unwrap(); + + // Connect to the echo serveer + + tokio::run_async(async move { + match await!(run_client(&addr)) { + Ok(_) => println!("done."), + Err(e) => eprintln!("echo client failed; error = {:?}", e), + } + }); +} diff --git a/tokio-async-await/examples/echo_server.rs b/tokio-async-await/examples/echo_server.rs new file mode 100644 index 000000000..8155c9d91 --- /dev/null +++ b/tokio-async-await/examples/echo_server.rs @@ -0,0 +1,45 @@ +#![feature(await_macro, async_await)] + +#[macro_use] +extern crate tokio; + +use tokio::net::{TcpListener, TcpStream}; +use tokio::prelude::*; + +use std::net::SocketAddr; + +fn handle(mut stream: TcpStream) { + tokio::spawn_async(async move { + let mut buf = [0; 1024]; + + loop { + match await!(stream.read_async(&mut buf)).unwrap() { + 0 => break, // Socket closed + n => { + // Send the data back + await!(stream.write_all_async(&buf[0..n])).unwrap(); + } + } + } + }); +} + +fn main() { + use std::env; + + let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string()); + let addr = addr.parse::().unwrap(); + + // Bind the TCP listener + let listener = TcpListener::bind(&addr).unwrap(); + println!("Listening on: {}", addr); + + tokio::run_async(async { + let mut incoming = listener.incoming(); + + while let Some(stream) = await!(incoming.next()) { + let stream = stream.unwrap(); + handle(stream); + } + }); +} diff --git a/tokio-async-await/examples/hyper.rs b/tokio-async-await/examples/hyper.rs new file mode 100644 index 000000000..c647cb941 --- /dev/null +++ b/tokio-async-await/examples/hyper.rs @@ -0,0 +1,25 @@ +#![feature(await_macro, async_await, futures_api)] + +#[macro_use] +extern crate tokio; +extern crate hyper; + +use tokio::prelude::*; +use hyper::Client; + +use std::time::Duration; + +pub fn main() { + tokio::run_async(async { + let client = Client::new(); + + let uri = "http://httpbin.org/ip".parse().unwrap(); + + let response = await!({ + client.get(uri) + .timeout(Duration::from_secs(10)) + }).unwrap(); + + println!("Response: {}", response.status()); + }); +} diff --git a/tokio-async-await/src/async_await/await.rs b/tokio-async-await/src/async_await/await.rs new file mode 100644 index 000000000..a2d4c09d0 --- /dev/null +++ b/tokio-async-await/src/async_await/await.rs @@ -0,0 +1,14 @@ +/// Wait for a future to complete. +#[macro_export] +macro_rules! await { + ($e:expr) => {{ + use $crate::std_await; + use $crate::async_await::compat::forward::IntoAwaitable as IntoAwaitableForward; + use $crate::async_await::compat::backward::IntoAwaitable as IntoAwaitableBackward; + + #[allow(unused_mut)] + let mut e = $e; + let e = e.into_awaitable(); + std_await!(e) + }} +} diff --git a/tokio-async-await/src/async_await/compat/backward.rs b/tokio-async-await/src/async_await/compat/backward.rs new file mode 100644 index 000000000..00ffe3962 --- /dev/null +++ b/tokio-async-await/src/async_await/compat/backward.rs @@ -0,0 +1,112 @@ +use futures::{ + Future as Future01, + Poll as Poll01, +}; +use futures_core::{Future as Future03}; + +use std::boxed::PinBox; +use std::future::FutureObj; +use std::ptr::NonNull; +use std::task::{ + Context, + Spawn, + UnsafeWake, + LocalWaker, + Poll as Poll03, + Waker, + SpawnObjError, +}; + +/// Convert an 0.3 `Future` to an 0.1 `Future`. +#[derive(Debug)] +pub struct Compat(PinBox); + +impl Compat { + pub fn new(data: T) -> Compat { + Compat(PinBox::new(data)) + } +} + +/// Convert a valuee into one that can be used with `await!`. +pub trait IntoAwaitable { + type Awaitable; + + fn into_awaitable(self) -> Self::Awaitable; +} + +impl IntoAwaitable for T +where T: Future03, +{ + type Awaitable = Self; + + fn into_awaitable(self) -> Self { + self + } +} + +impl Future01 for Compat +where T: Future03>, +{ + type Item = Item; + type Error = Error; + + fn poll(&mut self) -> Poll01 { + use futures::Async::*; + + let local_waker = noop_local_waker(); + let mut executor = NoopExecutor; + + let mut cx = Context::new(&local_waker, &mut executor); + + let res = self.0.as_pin_mut().poll(&mut cx); + + match res { + Poll03::Ready(Ok(val)) => Ok(Ready(val)), + Poll03::Ready(Err(err)) => Err(err), + Poll03::Pending => Ok(NotReady), + } + } +} + +// ===== NoopWaker ===== + +struct NoopWaker; + +fn noop_local_waker() -> LocalWaker { + let w: NonNull = NonNull::dangling(); + unsafe { LocalWaker::new(w) } +} + +fn noop_waker() -> Waker { + let w: NonNull = NonNull::dangling(); + unsafe { Waker::new(w) } +} + +unsafe impl UnsafeWake for NoopWaker { + unsafe fn clone_raw(&self) -> Waker { + noop_waker() + } + + unsafe fn drop_raw(&self) { + } + + unsafe fn wake(&self) { + panic!("NoopWake cannot wake"); + } +} + +// ===== NoopExecutor ===== + +struct NoopExecutor; + +impl Spawn for NoopExecutor { + fn spawn_obj(&mut self, future: FutureObj<'static, ()>) -> Result<(), SpawnObjError> { + use std::task::SpawnErrorKind; + + // NoopExecutor cannot execute + Err(SpawnObjError { + kind: SpawnErrorKind::shutdown(), + future, + }) + } +} diff --git a/tokio-async-await/src/async_await/compat/forward.rs b/tokio-async-await/src/async_await/compat/forward.rs new file mode 100644 index 000000000..2c0968288 --- /dev/null +++ b/tokio-async-await/src/async_await/compat/forward.rs @@ -0,0 +1,69 @@ + +use futures::{Future, Async}; +use futures_core::future::Future as Future03; +use futures_core::task::Poll as Poll03; + +use std::marker::Unpin; +use std::mem::PinMut; +use std::task::Context; + +/// Converts an 0.1 `Future` into an 0.3 `Future`. +#[derive(Debug)] +pub struct Compat(T); + +pub(crate) fn convert_poll(poll: Result, E>) -> Poll03> { + use futures::Async::{Ready, NotReady}; + + match poll { + Ok(Ready(val)) => Poll03::Ready(Ok(val)), + Ok(NotReady) => Poll03::Pending, + Err(err) => Poll03::Ready(Err(err)), + } +} + +pub(crate) fn convert_poll_stream( + poll: Result>, E>) -> Poll03>> +{ + use futures::Async::{Ready, NotReady}; + + match poll { + Ok(Ready(Some(val))) => Poll03::Ready(Some(Ok(val))), + Ok(Ready(None)) => Poll03::Ready(None), + Ok(NotReady) => Poll03::Pending, + Err(err) => Poll03::Ready(Some(Err(err))), + } +} + +/// Convert a value into one that can be used with `await!`. +pub trait IntoAwaitable { + type Awaitable; + + /// Convert `self` into a value that can be used with `await!`. + fn into_awaitable(self) -> Self::Awaitable; +} + +impl IntoAwaitable for T { + type Awaitable = Compat; + + fn into_awaitable(self) -> Self::Awaitable { + Compat(self) + } +} + +impl Future03 for Compat +where T: Future + Unpin +{ + type Output = Result; + + fn poll(self: PinMut, _cx: &mut Context) -> Poll03 { + use futures::Async::{Ready, NotReady}; + + // TODO: wire in cx + + match PinMut::get_mut(self).0.poll() { + Ok(Ready(val)) => Poll03::Ready(Ok(val)), + Ok(NotReady) => Poll03::Pending, + Err(e) => Poll03::Ready(Err(e)), + } + } +} diff --git a/tokio-async-await/src/async_await/compat/mod.rs b/tokio-async-await/src/async_await/compat/mod.rs new file mode 100644 index 000000000..6672f07a1 --- /dev/null +++ b/tokio-async-await/src/async_await/compat/mod.rs @@ -0,0 +1,4 @@ +#![doc(hidden)] + +pub mod forward; +pub mod backward; diff --git a/tokio-async-await/src/async_await/io/flush.rs b/tokio-async-await/src/async_await/io/flush.rs new file mode 100644 index 000000000..d2279eed0 --- /dev/null +++ b/tokio-async-await/src/async_await/io/flush.rs @@ -0,0 +1,32 @@ +use tokio_io::AsyncWrite; + +use futures_core::future::Future; +use futures_core::task::{self, Poll}; + +use std::io; +use std::marker::Unpin; +use std::mem::PinMut; + +/// A future used to fully flush an I/O object. +#[derive(Debug)] +pub struct Flush<'a, T: ?Sized + 'a> { + writer: &'a mut T, +} + +// PinMut is never projected to fields +impl<'a, T: ?Sized> Unpin for Flush<'a, T> {} + +impl<'a, T: AsyncWrite + ?Sized> Flush<'a, T> { + pub(super) fn new(writer: &'a mut T) -> Flush<'a, T> { + Flush { writer } + } +} + +impl<'a, T: AsyncWrite + ?Sized> Future for Flush<'a, T> { + type Output = io::Result<()>; + + fn poll(mut self: PinMut, _cx: &mut task::Context) -> Poll { + use crate::async_await::compat::forward::convert_poll; + convert_poll(self.writer.poll_flush()) + } +} diff --git a/tokio-async-await/src/async_await/io/mod.rs b/tokio-async-await/src/async_await/io/mod.rs new file mode 100644 index 000000000..44438af16 --- /dev/null +++ b/tokio-async-await/src/async_await/io/mod.rs @@ -0,0 +1,192 @@ +//! Use I/O with `async` / `await`. + +mod flush; +mod read; +mod read_exact; +mod write; +mod write_all; + +pub use self::flush::Flush; +pub use self::read::Read; +pub use self::read_exact::ReadExact; +pub use self::write::Write; +pub use self::write_all::WriteAll; + +use tokio_io::{AsyncRead, AsyncWrite}; + +/// An extension trait which adds utility methods to `AsyncRead` types. +pub trait AsyncReadExt: AsyncRead { + /// Tries to read some bytes directly into the given `buf` in asynchronous + /// manner, returning a future. + /// + /// The returned future will resolve to the number of bytes read once the read + /// operation is completed. + /// + /// # Examples + /// + /// ``` + /// #![feature(async_await, await_macro, futures_api)] + /// tokio::run_async(async { + /// // The extension trait can also be imported with + /// // `use tokio::prelude::*`. + /// use tokio::prelude::AsyncReadExt; + /// use std::io::Cursor; + /// + /// let mut reader = Cursor::new([1, 2, 3, 4]); + /// let mut output = [0u8; 5]; + /// + /// let bytes = await!(reader.read_async(&mut output[..])).unwrap(); + /// + /// // This is only guaranteed to be 4 because `&[u8]` is a synchronous + /// // reader. In a real system you could get anywhere from 1 to + /// // `output.len()` bytes in a single read. + /// assert_eq!(bytes, 4); + /// assert_eq!(output, [1, 2, 3, 4, 0]); + /// }); + /// ``` + fn read_async<'a>(&'a mut self, buf: &'a mut [u8]) -> Read<'a, Self> { + Read::new(self, buf) + } + + /// Creates a future which will read exactly enough bytes to fill `buf`, + /// returning an error if end of file (EOF) is hit sooner. + /// + /// The returned future will resolve once the read operation is completed. + /// + /// In the case of an error the buffer and the object will be discarded, with + /// the error yielded. + /// + /// # Examples + /// + /// ``` + /// #![feature(async_await, await_macro, futures_api)] + /// tokio::run_async(async { + /// // The extension trait can also be imported with + /// // `use tokio::prelude::*`. + /// use tokio::prelude::AsyncReadExt; + /// use std::io::Cursor; + /// + /// let mut reader = Cursor::new([1, 2, 3, 4]); + /// let mut output = [0u8; 4]; + /// + /// await!(reader.read_exact_async(&mut output)).unwrap(); + /// + /// assert_eq!(output, [1, 2, 3, 4]); + /// }); + /// ``` + /// + /// ## EOF is hit before `buf` is filled + /// + /// ``` + /// #![feature(async_await, await_macro, futures_api)] + /// tokio::run_async(async { + /// // The extension trait can also be imported with + /// // `use tokio::prelude::*`. + /// use tokio::prelude::AsyncReadExt; + /// use std::io::{self, Cursor}; + /// + /// let mut reader = Cursor::new([1, 2, 3, 4]); + /// let mut output = [0u8; 5]; + /// + /// let result = await!(reader.read_exact_async(&mut output)); + /// + /// assert_eq!(result.unwrap_err().kind(), io::ErrorKind::UnexpectedEof); + /// }); + /// ``` + fn read_exact_async<'a>(&'a mut self, buf: &'a mut [u8]) -> ReadExact<'a, Self> { + ReadExact::new(self, buf) + } +} + +/// An extension trait which adds utility methods to `AsyncWrite` types. +pub trait AsyncWriteExt: AsyncWrite { + /// Write data into this object. + /// + /// Creates a future that will write the entire contents of the buffer `buf` into + /// this `AsyncWrite`. + /// + /// The returned future will not complete until all the data has been written. + /// + /// # Examples + /// + /// ``` + /// #![feature(async_await, await_macro, futures_api)] + /// tokio::run_async(async { + /// // The extension trait can also be imported with + /// // `use tokio::prelude::*`. + /// use tokio::prelude::AsyncWriteExt; + /// use std::io::Cursor; + /// + /// let mut buf = [0u8; 5]; + /// let mut writer = Cursor::new(&mut buf[..]); + /// + /// let n = await!(writer.write_async(&[1, 2, 3, 4])).unwrap(); + /// + /// assert_eq!(writer.into_inner()[..n], [1, 2, 3, 4, 0][..n]); + /// }); + /// ``` + fn write_async<'a>(&'a mut self, buf: &'a [u8]) -> Write<'a, Self> { + Write::new(self, buf) + } + + /// Write an entire buffer into this object. + /// + /// Creates a future that will write the entire contents of the buffer `buf` into + /// this `AsyncWrite`. + /// + /// The returned future will not complete until all the data has been written. + /// + /// # Examples + /// + /// ``` + /// #![feature(async_await, await_macro, futures_api)] + /// tokio::run_async(async { + /// // The extension trait can also be imported with + /// // `use tokio::prelude::*`. + /// use tokio::prelude::AsyncWriteExt; + /// use std::io::Cursor; + /// + /// let mut buf = [0u8; 5]; + /// let mut writer = Cursor::new(&mut buf[..]); + /// + /// await!(writer.write_all_async(&[1, 2, 3, 4])).unwrap(); + /// + /// assert_eq!(writer.into_inner(), [1, 2, 3, 4, 0]); + /// }); + /// ``` + fn write_all_async<'a>(&'a mut self, buf: &'a [u8]) -> WriteAll<'a, Self> { + WriteAll::new(self, buf) + } + + /// Creates a future which will entirely flush this `AsyncWrite`. + /// + /// # Examples + /// + /// ``` + /// #![feature(async_await, await_macro, futures_api)] + /// tokio::run_async(async { + /// // The extension trait can also be imported with + /// // `use tokio::prelude::*`. + /// use tokio::prelude::AsyncWriteExt; + /// use std::io::{BufWriter, Cursor}; + /// + /// let mut output = [0u8; 5]; + /// + /// { + /// let mut writer = Cursor::new(&mut output[..]); + /// let mut buffered = BufWriter::new(writer); + /// await!(buffered.write_all_async(&[1, 2])).unwrap(); + /// await!(buffered.write_all_async(&[3, 4])).unwrap(); + /// await!(buffered.flush_async()).unwrap(); + /// } + /// + /// assert_eq!(output, [1, 2, 3, 4, 0]); + /// }); + /// ``` + fn flush_async<'a>(&mut self) -> Flush { + Flush::new(self) + } +} + +impl AsyncReadExt for T {} +impl AsyncWriteExt for T {} diff --git a/tokio-async-await/src/async_await/io/read.rs b/tokio-async-await/src/async_await/io/read.rs new file mode 100644 index 000000000..6780ca244 --- /dev/null +++ b/tokio-async-await/src/async_await/io/read.rs @@ -0,0 +1,38 @@ +use tokio_io::AsyncRead; + +use futures_core::future::Future; +use futures_core::task::{self, Poll}; + +use std::io; +use std::marker::Unpin; +use std::mem::PinMut; + +/// A future which can be used to read bytes. +#[derive(Debug)] +pub struct Read<'a, T: ?Sized + 'a> { + reader: &'a mut T, + buf: &'a mut [u8], +} + +// Pinning is never projected to fields +impl<'a, T: ?Sized> Unpin for Read<'a, T> {} + +impl<'a, T: AsyncRead + ?Sized> Read<'a, T> { + pub(super) fn new(reader: &'a mut T, buf: &'a mut [u8]) -> Read<'a, T> { + Read { + reader, + buf, + } + } +} + +impl<'a, T: AsyncRead + ?Sized> Future for Read<'a, T> { + type Output = io::Result; + + fn poll(mut self: PinMut, _cx: &mut task::Context) -> Poll { + use crate::async_await::compat::forward::convert_poll; + + let this = &mut *self; + convert_poll(this.reader.poll_read(this.buf)) + } +} diff --git a/tokio-async-await/src/async_await/io/read_exact.rs b/tokio-async-await/src/async_await/io/read_exact.rs new file mode 100644 index 000000000..df319c89a --- /dev/null +++ b/tokio-async-await/src/async_await/io/read_exact.rs @@ -0,0 +1,56 @@ +use tokio_io::AsyncRead; + +use futures_core::future::Future; +use futures_core::task::{self, Poll}; +use futures_util::try_ready; + +use std::io; +use std::marker::Unpin; +use std::mem::{self, PinMut}; + +/// A future which can be used to read exactly enough bytes to fill a buffer. +#[derive(Debug)] +pub struct ReadExact<'a, T: ?Sized + 'a> { + reader: &'a mut T, + buf: &'a mut [u8], +} + +// Pinning is never projected to fields +impl<'a, T: ?Sized> Unpin for ReadExact<'a, T> {} + +impl<'a, T: AsyncRead + ?Sized> ReadExact<'a, T> { + pub(super) fn new(reader: &'a mut T, buf: &'a mut [u8]) -> ReadExact<'a, T> { + ReadExact { + reader, + buf, + } + } +} + +fn eof() -> io::Error { + io::Error::new(io::ErrorKind::UnexpectedEof, "early eof") +} + +impl<'a, T: AsyncRead + ?Sized> Future for ReadExact<'a, T> { + type Output = io::Result<()>; + + fn poll(mut self: PinMut, _cx: &mut task::Context) -> Poll { + use crate::async_await::compat::forward::convert_poll; + + let this = &mut *self; + + while !this.buf.is_empty() { + let n = try_ready!(convert_poll(this.reader.poll_read(this.buf))); + + { + let (_, rest) = mem::replace(&mut this.buf, &mut []).split_at_mut(n); + this.buf = rest; + } + if n == 0 { + return Poll::Ready(Err(eof())) + } + } + + Poll::Ready(Ok(())) + } +} diff --git a/tokio-async-await/src/async_await/io/write.rs b/tokio-async-await/src/async_await/io/write.rs new file mode 100644 index 000000000..bf54e5c7d --- /dev/null +++ b/tokio-async-await/src/async_await/io/write.rs @@ -0,0 +1,38 @@ +use tokio_io::AsyncWrite; + +use futures_core::future::Future; +use futures_core::task::{self, Poll}; + +use std::io; +use std::marker::Unpin; +use std::mem::PinMut; + +/// A future used to write data. +#[derive(Debug)] +pub struct Write<'a, T: 'a + ?Sized> { + writer: &'a mut T, + buf: &'a [u8], +} + +// Pinning is never projected to fields +impl<'a, T: ?Sized> Unpin for Write<'a, T> {} + +impl<'a, T: AsyncWrite + ?Sized> Write<'a, T> { + pub(super) fn new(writer: &'a mut T, buf: &'a [u8]) -> Write<'a, T> { + Write { + writer, + buf, + } + } +} + +impl<'a, T: AsyncWrite + ?Sized> Future for Write<'a, T> { + type Output = io::Result; + + fn poll(mut self: PinMut, _cx: &mut task::Context) -> Poll> { + use crate::async_await::compat::forward::convert_poll; + + let this = &mut *self; + convert_poll(this.writer.poll_write(this.buf)) + } +} diff --git a/tokio-async-await/src/async_await/io/write_all.rs b/tokio-async-await/src/async_await/io/write_all.rs new file mode 100644 index 000000000..757017253 --- /dev/null +++ b/tokio-async-await/src/async_await/io/write_all.rs @@ -0,0 +1,57 @@ +use tokio_io::AsyncWrite; + +use futures_core::future::Future; +use futures_core::task::{self, Poll}; +use futures_util::try_ready; + +use std::io; +use std::marker::Unpin; +use std::mem::{self, PinMut}; + +/// A future used to write the entire contents of a buffer. +#[derive(Debug)] +pub struct WriteAll<'a, T: ?Sized + 'a> { + writer: &'a mut T, + buf: &'a [u8], +} + +// Pinning is never projected to fields +impl<'a, T: ?Sized> Unpin for WriteAll<'a, T> {} + +impl<'a, T: AsyncWrite + ?Sized> WriteAll<'a, T> { + pub(super) fn new(writer: &'a mut T, buf: &'a [u8]) -> WriteAll<'a, T> { + WriteAll { + writer, + buf, + } + } +} + +fn zero_write() -> io::Error { + io::Error::new(io::ErrorKind::WriteZero, "zero-length write") +} + +impl<'a, T: AsyncWrite + ?Sized> Future for WriteAll<'a, T> { + type Output = io::Result<()>; + + fn poll(mut self: PinMut, _cx: &mut task::Context) -> Poll> { + use crate::async_await::compat::forward::convert_poll; + + let this = &mut *self; + + while !this.buf.is_empty() { + let n = try_ready!(convert_poll(this.writer.poll_write(this.buf))); + + { + let (_, rest) = mem::replace(&mut this.buf, &[]).split_at(n); + this.buf = rest; + } + + if n == 0 { + return Poll::Ready(Err(zero_write())) + } + } + + Poll::Ready(Ok(())) + } +} diff --git a/tokio-async-await/src/async_await/mod.rs b/tokio-async-await/src/async_await/mod.rs new file mode 100644 index 000000000..8731fa3dd --- /dev/null +++ b/tokio-async-await/src/async_await/mod.rs @@ -0,0 +1,8 @@ +//! Utilities for working with `async` / `await`. + +#[macro_use] +mod await; +pub mod compat; +pub mod io; +pub mod sink; +pub mod stream; diff --git a/tokio-async-await/src/async_await/sink/mod.rs b/tokio-async-await/src/async_await/sink/mod.rs new file mode 100644 index 000000000..69c3f079e --- /dev/null +++ b/tokio-async-await/src/async_await/sink/mod.rs @@ -0,0 +1,26 @@ +//! Use sinks with `async` / `await`. + +mod send; + +pub use self::send::Send; + +use futures::Sink; + +use std::marker::Unpin; + +/// An extension trait which adds utility methods to `Sink` types. +pub trait SinkExt: Sink { + /// Send an item into the sink. + /// + /// Note that, **because of the flushing requirement, it is usually better + /// to batch together items to send via `send_all`, rather than flushing + /// between each item.** + fn send_async(&mut self, item: Self::SinkItem) -> Send + where + Self: Sized + Unpin, + { + Send::new(self, item) + } +} + +impl SinkExt for T {} diff --git a/tokio-async-await/src/async_await/sink/send.rs b/tokio-async-await/src/async_await/sink/send.rs new file mode 100644 index 000000000..3c17a7f4e --- /dev/null +++ b/tokio-async-await/src/async_await/sink/send.rs @@ -0,0 +1,59 @@ +use futures::Sink; + +use futures_core::future::Future; +use futures_core::task::{self, Poll}; + +use std::marker::Unpin; +use std::mem::PinMut; + +/// Future for the `SinkExt::send_async` combinator, which sends a value to a +/// sink and then waits until the sink has fully flushed. +#[derive(Debug)] +pub struct Send<'a, T: Sink + 'a + ?Sized> { + sink: &'a mut T, + item: Option, +} + +impl Unpin for Send<'_, T> {} + +impl<'a, T: Sink + Unpin + ?Sized> Send<'a, T> { + pub(super) fn new(sink: &'a mut T, item: T::SinkItem) -> Self { + Send { + sink, + item: Some(item), + } + } +} + +impl Future for Send<'_, T> { + type Output = Result<(), T::SinkError>; + + fn poll(mut self: PinMut, _cx: &mut task::Context) -> Poll { + use crate::async_await::compat::forward::convert_poll; + use futures::AsyncSink::{Ready, NotReady}; + use futures_util::try_ready; + + // use crate::compat::forward::convert_poll; + + let this = &mut *self; + + if let Some(item) = this.item.take() { + match this.sink.start_send(item) { + Ok(Ready) => {} + Ok(NotReady(val)) => { + self.item = Some(val); + return Poll::Pending; + } + Err(err) => { + return Poll::Ready(Err(err)); + } + } + } + + // we're done sending the item, but want to block on flushing the + // sink + try_ready!(convert_poll(this.sink.poll_complete())); + + Poll::Ready(Ok(())) + } +} diff --git a/tokio-async-await/src/async_await/stream/mod.rs b/tokio-async-await/src/async_await/stream/mod.rs new file mode 100644 index 000000000..b376f49c2 --- /dev/null +++ b/tokio-async-await/src/async_await/stream/mod.rs @@ -0,0 +1,40 @@ +//! Use streams with `async` / `await`. + +mod next; + +pub use self::next::Next; + +use futures::Stream; + +use std::marker::Unpin; + +/// An extension trait which adds utility methods to `Stream` types. +pub trait StreamExt: Stream { + /// Creates a future that resolves to the next item in the stream. + /// + /// # Examples + /// + /// ``` + /// #![feature(await_macro, async_await)] + /// tokio::run_async(async { + /// // The extension trait can also be imported with + /// // `use tokio::prelude::*`. + /// use tokio::prelude::{stream, StreamExt}; + /// + /// let mut stream = stream::iter_ok::<_, ()>(1..3); + /// + /// assert_eq!(await!(stream.next()), Some(Ok(1))); + /// assert_eq!(await!(stream.next()), Some(Ok(2))); + /// assert_eq!(await!(stream.next()), Some(Ok(3))); + /// assert_eq!(await!(stream.next()), None); + /// }); + /// ``` + fn next(&mut self) -> Next + where + Self: Sized + Unpin, + { + Next::new(self) + } +} + +impl StreamExt for T {} diff --git a/tokio-async-await/src/async_await/stream/next.rs b/tokio-async-await/src/async_await/stream/next.rs new file mode 100644 index 000000000..3d2f6bd76 --- /dev/null +++ b/tokio-async-await/src/async_await/stream/next.rs @@ -0,0 +1,31 @@ +use futures::Stream; +use futures_core::future::Future; +use futures_core::task::{self, Poll}; + +use std::marker::Unpin; +use std::mem::PinMut; + +/// A future of the next element of a stream. +#[derive(Debug)] +pub struct Next<'a, T: 'a> { + stream: &'a mut T, +} + +impl<'a, T: Stream + Unpin> Unpin for Next<'a, T> {} + +impl<'a, T: Stream + Unpin> Next<'a, T> { + pub(super) fn new(stream: &'a mut T) -> Next<'a, T> { + Next { stream } + } +} + +impl<'a, T: Stream + Unpin> Future for Next<'a, T> { + type Output = Option>; + + fn poll(self: PinMut, _cx: &mut task::Context) -> Poll { + use crate::async_await::compat::forward::convert_poll_stream; + + convert_poll_stream( + PinMut::get_mut(self).stream.poll()) + } +} diff --git a/tokio-async-await/src/lib.rs b/tokio-async-await/src/lib.rs new file mode 100644 index 000000000..a4bc60a86 --- /dev/null +++ b/tokio-async-await/src/lib.rs @@ -0,0 +1,102 @@ +#![feature(futures_api, await_macro, pin, arbitrary_self_types)] + +#![doc(html_root_url = "https://docs.rs/tokio-async-await/0.1.0")] +#![deny(missing_docs, missing_debug_implementations)] +#![cfg_attr(test, deny(warnings))] + +//! A preview of Tokio w/ `async` / `await` support. + +extern crate futures; +extern crate futures_core; +extern crate futures_util; + +// Re-export all of Tokio +pub use tokio_main::{ + // Modules + clock, + codec, + executor, + fs, + io, + net, + reactor, + runtime, + timer, + util, + + // Functions + run, + spawn, +}; + +pub mod sync { + //! Asynchronous aware synchronization + + pub use tokio_channel::{ + mpsc, + oneshot, + }; +} + +pub mod async_await; + +pub mod prelude { + //! A "prelude" for users of the `tokio` crate. + //! + //! This prelude is similar to the standard library's prelude in that you'll + //! almost always want to import its entire contents, but unlike the standard + //! library's prelude you'll have to do so manually: + //! + //! ``` + //! use tokio::prelude::*; + //! ``` + //! + //! The prelude may grow over time as additional items see ubiquitous use. + + pub use tokio_main::prelude::*; + + #[doc(inline)] + pub use crate::async_await::{ + io::{ + AsyncReadExt, + AsyncWriteExt, + }, + sink::{ + SinkExt, + }, + stream::{ + StreamExt, + }, + }; +} + +use futures_core::{ + Future as Future03, +}; + +// Rename the `await` macro in `std` +#[doc(hidden)] +#[macro_export] +pub use std::await as std_await; + +/// Like `tokio::run`, but takes an `async` block +pub fn run_async(future: F) +where F: Future03 + Send + 'static, +{ + use futures_util::future::FutureExt; + use crate::async_await::compat::backward; + + let future = future.map(|_| Ok(())); + run(backward::Compat::new(future)) +} + +/// Like `tokio::spawn`, but takes an `async` block +pub fn spawn_async(future: F) +where F: Future03 + Send + 'static, +{ + use futures_util::future::FutureExt; + use crate::async_await::compat::backward; + + let future = future.map(|_| Ok(())); + spawn(backward::Compat::new(future)); +} diff --git a/tokio-channel/CHANGELOG.md b/tokio-channel/CHANGELOG.md new file mode 100644 index 000000000..e69de29bb diff --git a/tokio-channel/Cargo.toml b/tokio-channel/Cargo.toml new file mode 100644 index 000000000..d38ac3753 --- /dev/null +++ b/tokio-channel/Cargo.toml @@ -0,0 +1,25 @@ +[package] +name = "tokio-channel" + +# When releasing to crates.io: +# - Update html_root_url. +# - Update CHANGELOG.md. +# - Create "v0.1.x" git tag. +version = "0.1.0" +authors = ["Carl Lerche "] +license = "MIT" +repository = "https://github.com/tokio-rs/tokio" +homepage = "https://tokio.rs" +documentation = "https://docs.rs/tokio-channel/0.1.0" +description = """ +Channels for asynchronous communication using Tokio. +""" +categories = ["asynchronous"] + +[features] +# This feature comes with no promise of stability. Things will break with each +# patch release. Use at your own risk. +async-await-preview = [] + +[dependencies] +futures = "0.1.23" diff --git a/tokio-channel/LICENSE b/tokio-channel/LICENSE new file mode 100644 index 000000000..e0c7ffadf --- /dev/null +++ b/tokio-channel/LICENSE @@ -0,0 +1,51 @@ +Copyright (c) 2018 Tokio Contributors + +Permission is hereby granted, free of charge, to any +person obtaining a copy of this software and associated +documentation files (the "Software"), to deal in the +Software without restriction, including without +limitation the rights to use, copy, modify, merge, +publish, distribute, sublicense, and/or sell copies of +the Software, and to permit persons to whom the Software +is furnished to do so, subject to the following +conditions: + +The above copyright notice and this permission notice +shall be included in all copies or substantial portions +of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF +ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED +TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT +SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR +IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +DEALINGS IN THE SOFTWARE. + +Copyright (c) 2016 futures-rs Contributors + +Permission is hereby granted, free of charge, to any +person obtaining a copy of this software and associated +documentation files (the "Software"), to deal in the +Software without restriction, including without +limitation the rights to use, copy, modify, merge, +publish, distribute, sublicense, and/or sell copies of +the Software, and to permit persons to whom the Software +is furnished to do so, subject to the following +conditions: + +The above copyright notice and this permission notice +shall be included in all copies or substantial portions +of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF +ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED +TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT +SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR +IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +DEALINGS IN THE SOFTWARE. diff --git a/tokio-channel/README.md b/tokio-channel/README.md new file mode 100644 index 000000000..e69de29bb diff --git a/tokio-channel/src/async_await.rs b/tokio-channel/src/async_await.rs new file mode 100644 index 000000000..ff501d683 --- /dev/null +++ b/tokio-channel/src/async_await.rs @@ -0,0 +1,10 @@ +use {oneshot, mpsc}; + +use std::marker::Unpin; + +impl Unpin for oneshot::Sender {} +impl Unpin for oneshot::Receiver {} + +impl Unpin for mpsc::Sender {} +impl Unpin for mpsc::UnboundedSender {} +impl Unpin for mpsc::Receiver {} diff --git a/tokio-channel/src/lib.rs b/tokio-channel/src/lib.rs new file mode 100644 index 000000000..ae385530c --- /dev/null +++ b/tokio-channel/src/lib.rs @@ -0,0 +1,22 @@ +#![doc(html_root_url = "https://docs.rs/tokio-channel/0.1.0")] +#![deny(missing_docs, warnings, missing_debug_implementations)] +#![cfg_attr(feature = "async-await-preview", feature( + pin, + ))] + +//! Asynchronous channels. +//! +//! This crate provides channels that can be used to communicate between +//! asynchronous tasks. + +extern crate futures; + +pub mod mpsc; +pub mod oneshot; + +mod lock; + +// ===== EXPERIMENTAL async / await support ===== + +#[cfg(feature = "async-await-preview")] +mod async_await; diff --git a/tokio-channel/src/lock.rs b/tokio-channel/src/lock.rs new file mode 100644 index 000000000..d2acf0a69 --- /dev/null +++ b/tokio-channel/src/lock.rs @@ -0,0 +1,105 @@ +//! A "mutex" which only supports `try_lock` +//! +//! As a futures library the eventual call to an event loop should be the only +//! thing that ever blocks, so this is assisted with a fast user-space +//! implementation of a lock that can only have a `try_lock` operation. + +use std::cell::UnsafeCell; +use std::ops::{Deref, DerefMut}; +use std::sync::atomic::Ordering::SeqCst; +use std::sync::atomic::AtomicBool; + +/// A "mutex" around a value, similar to `std::sync::Mutex`. +/// +/// This lock only supports the `try_lock` operation, however, and does not +/// implement poisoning. +#[derive(Debug)] +pub struct Lock { + locked: AtomicBool, + data: UnsafeCell, +} + +/// Sentinel representing an acquired lock through which the data can be +/// accessed. +pub struct TryLock<'a, T: 'a> { + __ptr: &'a Lock, +} + +// The `Lock` structure is basically just a `Mutex`, and these two impls are +// intended to mirror the standard library's corresponding impls for `Mutex`. +// +// If a `T` is sendable across threads, so is the lock, and `T` must be sendable +// across threads to be `Sync` because it allows mutable access from multiple +// threads. +unsafe impl Send for Lock {} +unsafe impl Sync for Lock {} + +impl Lock { + /// Creates a new lock around the given value. + pub fn new(t: T) -> Lock { + Lock { + locked: AtomicBool::new(false), + data: UnsafeCell::new(t), + } + } + + /// Attempts to acquire this lock, returning whether the lock was acquired or + /// not. + /// + /// If `Some` is returned then the data this lock protects can be accessed + /// through the sentinel. This sentinel allows both mutable and immutable + /// access. + /// + /// If `None` is returned then the lock is already locked, either elsewhere + /// on this thread or on another thread. + pub fn try_lock(&self) -> Option> { + if !self.locked.swap(true, SeqCst) { + Some(TryLock { __ptr: self }) + } else { + None + } + } +} + +impl<'a, T> Deref for TryLock<'a, T> { + type Target = T; + fn deref(&self) -> &T { + // The existence of `TryLock` represents that we own the lock, so we + // can safely access the data here. + unsafe { &*self.__ptr.data.get() } + } +} + +impl<'a, T> DerefMut for TryLock<'a, T> { + fn deref_mut(&mut self) -> &mut T { + // The existence of `TryLock` represents that we own the lock, so we + // can safely access the data here. + // + // Additionally, we're the *only* `TryLock` in existence so mutable + // access should be ok. + unsafe { &mut *self.__ptr.data.get() } + } +} + +impl<'a, T> Drop for TryLock<'a, T> { + fn drop(&mut self) { + self.__ptr.locked.store(false, SeqCst); + } +} + +#[cfg(test)] +mod tests { + use super::Lock; + + #[test] + fn smoke() { + let a = Lock::new(1); + let mut a1 = a.try_lock().unwrap(); + assert!(a.try_lock().is_none()); + assert_eq!(*a1, 1); + *a1 = 2; + drop(a1); + assert_eq!(*a.try_lock().unwrap(), 2); + assert_eq!(*a.try_lock().unwrap(), 2); + } +} diff --git a/tokio-channel/src/mpsc/mod.rs b/tokio-channel/src/mpsc/mod.rs new file mode 100644 index 000000000..21abd55d6 --- /dev/null +++ b/tokio-channel/src/mpsc/mod.rs @@ -0,0 +1,989 @@ +//! A multi-producer, single-consumer, futures-aware, FIFO queue with back pressure. +//! +//! A channel can be used as a communication primitive between tasks running on +//! `futures-rs` executors. 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 from the channel, the +//! current task will be notified when a new value is sent. `Sender` implements +//! the `Sink` trait and allows a task to send messages into the channel. If +//! the channel is at capacity, then send will be rejected and the task will be +//! notified when additional capacity is available. +//! +//! # 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, `Sender::poll` will return `Ok(Ready(None))`. +//! +//! If the receiver handle is dropped, then messages can no longer be read out +//! of the channel. In this case, a `send` will result in an error. +//! +//! # Clean Shutdown +//! +//! If the `Receiver` is simply dropped, then it is possible for there to be +//! messages still in the channel that will not be processed. As such, it is +//! usually desirable to perform a "clean" shutdown. To do this, the receiver +//! will first call `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. + +// At the core, the channel uses an atomic FIFO queue for message passing. This +// queue is used as the primary coordination primitive. In order to enforce +// capacity limits and handle back pressure, a secondary FIFO queue is used to +// send parked task handles. +// +// The general idea is that the channel is created with a `buffer` size of `n`. +// The channel capacity is `n + num-senders`. Each sender gets one "guaranteed" +// slot to hold a message. This allows `Sender` to know for a fact that a send +// will succeed *before* starting to do the actual work of sending the value. +// Since most of this work is lock-free, once the work starts, it is impossible +// to safely revert. +// +// If the sender is unable to process a send operation, then the current +// task is parked and the handle is sent on the parked task queue. +// +// Note that the implementation guarantees that the channel capacity will never +// exceed the configured limit, however there is no *strict* guarantee that the +// receiver will wake up a parked task *immediately* when a slot becomes +// available. However, it will almost always unpark a task when a slot becomes +// available and it is *guaranteed* that a sender will be unparked when the +// message that caused the sender to become parked is read out of the channel. +// +// The steps for sending a message are roughly: +// +// 1) Increment the channel message count +// 2) If the channel is at capacity, push the task handle onto the wait queue +// 3) Push the message onto the message queue. +// +// The steps for receiving a message are roughly: +// +// 1) Pop a message from the message queue +// 2) Pop a task handle from the wait queue +// 3) Decrement the channel message count. +// +// It's important for the order of operations on lock-free structures to happen +// in reverse order between the sender and receiver. This makes the message +// queue the primary coordination structure and establishes the necessary +// happens-before semantics required for the acquire / release semantics used +// by the queue structure. + + + +use mpsc::queue::{Queue, PopResult}; + +use futures::task::{self, Task}; +use futures::{Async, AsyncSink, Poll, StartSend, Sink, Stream}; + +use std::fmt; +use std::error::Error; +use std::any::Any; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering::SeqCst; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::usize; + +mod queue; + +/// The transmission end of a channel which is used to send values. +/// +/// This is created by the `channel` method. +#[derive(Debug)] +pub struct Sender { + // Channel state shared between the sender and receiver. + inner: Arc>, + + // Handle to the task that is blocked on this sender. This handle is sent + // to the receiver half in order to be notified when the sender becomes + // unblocked. + sender_task: Arc>, + + // True if the sender might be blocked. This is an optimization to avoid + // having to lock the mutex most of the time. + maybe_parked: bool, +} + +/// The transmission end of a channel which is used to send values. +/// +/// This is created by the `unbounded` method. +#[derive(Debug)] +pub struct UnboundedSender(Sender); + +trait AssertKinds: Send + Sync + Clone {} +impl AssertKinds for UnboundedSender {} + + +/// The receiving end of a channel which implements the `Stream` trait. +/// +/// This is a concrete implementation of a stream which can be used to represent +/// a stream of values being computed elsewhere. This is created by the +/// `channel` method. +#[derive(Debug)] +pub struct Receiver { + inner: Arc>, +} + +/// Error type for sending, used when the receiving end of a channel is +/// dropped +#[derive(Clone, PartialEq, Eq)] +pub struct SendError(T); + +/// Error type returned from `try_send` +#[derive(Clone, PartialEq, Eq)] +pub struct TrySendError { + kind: TrySendErrorKind, +} + +#[derive(Clone, PartialEq, Eq)] +enum TrySendErrorKind { + Full(T), + Disconnected(T), +} + +impl fmt::Debug for SendError { + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + fmt.debug_tuple("SendError") + .field(&"...") + .finish() + } +} + +impl fmt::Display for SendError { + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + write!(fmt, "send failed because receiver is gone") + } +} + +impl Error for SendError +{ + fn description(&self) -> &str { + "send failed because receiver is gone" + } +} + +impl SendError { + /// Returns the message that was attempted to be sent but failed. + pub fn into_inner(self) -> T { + self.0 + } +} + +impl fmt::Debug for TrySendError { + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + fmt.debug_tuple("TrySendError") + .field(&"...") + .finish() + } +} + +impl fmt::Display for TrySendError { + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + if self.is_full() { + write!(fmt, "send failed because channel is full") + } else { + write!(fmt, "send failed because receiver is gone") + } + } +} + +impl Error for TrySendError { + fn description(&self) -> &str { + if self.is_full() { + "send failed because channel is full" + } else { + "send failed because receiver is gone" + } + } +} + +impl TrySendError { + /// Returns true if this error is a result of the channel being full + pub fn is_full(&self) -> bool { + use self::TrySendErrorKind::*; + + match self.kind { + Full(_) => true, + _ => false, + } + } + + /// Returns true if this error is a result of the receiver being dropped + pub fn is_disconnected(&self) -> bool { + use self::TrySendErrorKind::*; + + match self.kind { + Disconnected(_) => true, + _ => false, + } + } + + /// Returns the message that was attempted to be sent but failed. + pub fn into_inner(self) -> T { + use self::TrySendErrorKind::*; + + match self.kind { + Full(v) | Disconnected(v) => v, + } + } +} + +#[derive(Debug)] +struct Inner { + // Max buffer size of the channel. If `None` then the channel is unbounded. + buffer: Option, + + // Internal channel state. Consists of the number of messages stored in the + // channel as well as a flag signalling that the channel is closed. + state: AtomicUsize, + + // Atomic, FIFO queue used to send messages to the receiver + message_queue: Queue>, + + // Atomic, FIFO queue used to send parked task handles to the receiver. + parked_queue: Queue>>, + + // Number of senders in existence + num_senders: AtomicUsize, + + // Handle to the receiver's task. + recv_task: Mutex, +} + +// Struct representation of `Inner::state`. +#[derive(Debug, Clone, Copy)] +struct State { + // `true` when the channel is open + is_open: bool, + + // Number of messages in the channel + num_messages: usize, +} + +#[derive(Debug)] +struct ReceiverTask { + unparked: bool, + task: Option, +} + +// Returned from Receiver::try_park() +enum TryPark { + Parked, + Closed, + NotEmpty, +} + +// The `is_open` flag is stored in the left-most bit of `Inner::state` +const OPEN_MASK: usize = usize::MAX - (usize::MAX >> 1); + +// When a new channel is created, it is created in the open state with no +// pending messages. +const INIT_STATE: usize = OPEN_MASK; + +// The maximum number of messages that a channel can track is `usize::MAX >> 1` +const MAX_CAPACITY: usize = !(OPEN_MASK); + +// The maximum requested buffer size must be less than the maximum capacity of +// a channel. This is because each sender gets a guaranteed slot. +const MAX_BUFFER: usize = MAX_CAPACITY >> 1; + +// Sent to the consumer to wake up blocked producers +#[derive(Debug)] +struct SenderTask { + task: Option, + is_parked: bool, +} + +impl SenderTask { + fn new() -> Self { + SenderTask { + task: None, + is_parked: false, + } + } + + fn notify(&mut self) { + self.is_parked = false; + + if let Some(task) = self.task.take() { + task.notify(); + } + } +} + +/// Creates an in-memory channel implementation of the `Stream` trait with +/// bounded capacity. +/// +/// This method creates a concrete implementation of the `Stream` trait which +/// can be used to send values across threads in a streaming fashion. This +/// channel is unique in that it implements back pressure to ensure that the +/// sender never outpaces the receiver. The channel capacity is equal to +/// `buffer + num-senders`. In other words, each sender gets a guaranteed slot +/// in the channel capacity, and on top of that there are `buffer` "first come, +/// first serve" slots available to all senders. +/// +/// The `Receiver` returned implements the `Stream` trait and has access to any +/// number of the associated combinators for transforming the result. +pub fn channel(buffer: usize) -> (Sender, Receiver) { + // Check that the requested buffer size does not exceed the maximum buffer + // size permitted by the system. + assert!(buffer < MAX_BUFFER, "requested buffer size too large"); + channel2(Some(buffer)) +} + +/// Creates an in-memory channel implementation of the `Stream` trait with +/// unbounded capacity. +/// +/// This method creates a concrete implementation of the `Stream` trait which +/// can be used to send values across threads in a streaming fashion. 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 buffered +/// internally. +/// +/// **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() -> (UnboundedSender, Receiver) { + let (tx, rx) = channel2(None); + (UnboundedSender(tx), rx) +} + +fn channel2(buffer: Option) -> (Sender, Receiver) { + let inner = Arc::new(Inner { + buffer: buffer, + state: AtomicUsize::new(INIT_STATE), + message_queue: Queue::new(), + parked_queue: Queue::new(), + num_senders: AtomicUsize::new(1), + recv_task: Mutex::new(ReceiverTask { + unparked: false, + task: None, + }), + }); + + let tx = Sender { + inner: inner.clone(), + sender_task: Arc::new(Mutex::new(SenderTask::new())), + maybe_parked: false, + }; + + let rx = Receiver { + inner: inner, + }; + + (tx, rx) +} + +/* + * + * ===== impl Sender ===== + * + */ + +impl Sender { + /// Attempts to send a message on this `Sender` without blocking. + /// + /// This function, unlike `start_send`, is safe to call whether it's being + /// called on a task or not. Note that this function, however, will *not* + /// attempt to block the current task if the message cannot be sent. + /// + /// It is not recommended to call this function from inside of a future, + /// only from an external thread where you've otherwise arranged to be + /// notified when the channel is no longer full. + pub fn try_send(&mut self, msg: T) -> Result<(), TrySendError> { + // If the sender is currently blocked, reject the message + if !self.poll_unparked(false).is_ready() { + return Err(TrySendError { + kind: TrySendErrorKind::Full(msg), + }); + } + + // The channel has capacity to accept the message, so send it + self.do_send(Some(msg), false) + .map_err(|SendError(v)| { + TrySendError { + kind: TrySendErrorKind::Disconnected(v), + } + }) + } + + // Do the send without failing + // None means close + fn do_send(&mut self, msg: Option, do_park: bool) -> Result<(), SendError> { + // First, increment the number of messages contained by the channel. + // This operation will also atomically determine if the sender task + // should be parked. + // + // None is returned in the case that the channel has been closed by the + // receiver. This happens when `Receiver::close` is called or the + // receiver is dropped. + let park_self = match self.inc_num_messages(msg.is_none()) { + Some(park_self) => park_self, + None => { + // The receiver has closed the channel. Only abort if actually + // sending a message. It is important that the stream + // termination (None) is always sent. This technically means + // that it is possible for the queue to contain the following + // number of messages: + // + // num-senders + buffer + 1 + // + if let Some(msg) = msg { + return Err(SendError(msg)); + } else { + return Ok(()); + } + } + }; + + // If the channel has reached capacity, then the sender task needs to + // be parked. This will send the task handle on the parked task queue. + // + // However, when `do_send` is called while dropping the `Sender`, + // `task::current()` can't be called safely. In this case, in order to + // maintain internal consistency, a blank message is pushed onto the + // parked task queue. + if park_self { + self.park(do_park); + } + + self.queue_push_and_signal(msg); + + Ok(()) + } + + // Do the send without parking current task. + // + // To be called from unbounded sender. + fn do_send_nb(&self, msg: T) -> Result<(), SendError> { + match self.inc_num_messages(false) { + Some(park_self) => assert!(!park_self), + None => return Err(SendError(msg)), + }; + + self.queue_push_and_signal(Some(msg)); + + Ok(()) + } + + // Push message to the queue and signal to the receiver + fn queue_push_and_signal(&self, msg: Option) { + // Push the message onto the message queue + self.inner.message_queue.push(msg); + + // Signal to the receiver that a message has been enqueued. If the + // receiver is parked, this will unpark the task. + self.signal(); + } + + // Increment the number of queued messages. Returns if the sender should + // block. + fn inc_num_messages(&self, close: bool) -> Option { + let mut curr = self.inner.state.load(SeqCst); + + loop { + let mut state = decode_state(curr); + + // The receiver end closed the channel. + if !state.is_open { + return None; + } + + // This probably is never hit? Odds are the process will run out of + // memory first. It may be worth to return something else in this + // case? + assert!(state.num_messages < MAX_CAPACITY, "buffer space exhausted; \ + sending this messages would overflow the state"); + + state.num_messages += 1; + + // The channel is closed by all sender handles being dropped. + if close { + state.is_open = false; + } + + let next = encode_state(&state); + match self.inner.state.compare_exchange(curr, next, SeqCst, SeqCst) { + Ok(_) => { + // Block if the current number of pending messages has exceeded + // the configured buffer size + let park_self = match self.inner.buffer { + Some(buffer) => state.num_messages > buffer, + None => false, + }; + + return Some(park_self) + } + Err(actual) => curr = actual, + } + } + } + + // Signal to the receiver task that a message has been enqueued + fn signal(&self) { + // TODO + // This logic can probably be improved by guarding the lock with an + // atomic. + // + // Do this step first so that the lock is dropped when + // `unpark` is called + let task = { + let mut recv_task = self.inner.recv_task.lock().unwrap(); + + // If the receiver has already been unparked, then there is nothing + // more to do + if recv_task.unparked { + return; + } + + // Setting this flag enables the receiving end to detect that + // an unpark event happened in order to avoid unnecessarily + // parking. + recv_task.unparked = true; + recv_task.task.take() + }; + + if let Some(task) = task { + task.notify(); + } + } + + fn park(&mut self, can_park: bool) { + // TODO: clean up internal state if the task::current will fail + + let task = if can_park { + Some(task::current()) + } else { + None + }; + + { + let mut sender = self.sender_task.lock().unwrap(); + sender.task = task; + sender.is_parked = true; + } + + // Send handle over queue + let t = self.sender_task.clone(); + self.inner.parked_queue.push(t); + + // Check to make sure we weren't closed after we sent our task on the + // queue + let state = decode_state(self.inner.state.load(SeqCst)); + self.maybe_parked = state.is_open; + } + + /// Polls the channel to determine if there is guaranteed to be capacity to send at least one + /// item without waiting. + /// + /// Returns `Ok(Async::Ready(_))` if there is sufficient capacity, or returns + /// `Ok(Async::NotReady)` if the channel is not guaranteed to have capacity. Returns + /// `Err(SendError(_))` if the receiver has been dropped. + /// + /// # Panics + /// + /// This method will panic if called from outside the context of a task or future. + pub fn poll_ready(&mut self) -> Poll<(), SendError<()>> { + let state = decode_state(self.inner.state.load(SeqCst)); + if !state.is_open { + return Err(SendError(())); + } + + Ok(self.poll_unparked(true)) + } + + fn poll_unparked(&mut self, do_park: bool) -> Async<()> { + // First check the `maybe_parked` variable. This avoids acquiring the + // lock in most cases + if self.maybe_parked { + // Get a lock on the task handle + let mut task = self.sender_task.lock().unwrap(); + + if !task.is_parked { + self.maybe_parked = false; + return Async::Ready(()) + } + + // At this point, an unpark request is pending, so there will be an + // unpark sometime in the future. We just need to make sure that + // the correct task will be notified. + // + // Update the task in case the `Sender` has been moved to another + // task + task.task = if do_park { + Some(task::current()) + } else { + None + }; + + Async::NotReady + } else { + Async::Ready(()) + } + } +} + +impl Sink for Sender { + type SinkItem = T; + type SinkError = SendError; + + fn start_send(&mut self, msg: T) -> StartSend> { + // If the sender is currently blocked, reject the message before doing + // any work. + if !self.poll_unparked(true).is_ready() { + return Ok(AsyncSink::NotReady(msg)); + } + + // The channel has capacity to accept the message, so send it. + self.do_send(Some(msg), true)?; + + Ok(AsyncSink::Ready) + } + + fn poll_complete(&mut self) -> Poll<(), SendError> { + Ok(Async::Ready(())) + } + + fn close(&mut self) -> Poll<(), SendError> { + Ok(Async::Ready(())) + } +} + +impl UnboundedSender { + /// Sends the provided message along this channel. + /// + /// This is an unbounded sender, so this function differs from `Sink::send` + /// by ensuring the return type reflects that the channel is always ready to + /// receive messages. + #[deprecated(note = "renamed to `unbounded_send`")] + #[doc(hidden)] + pub fn send(&self, msg: T) -> Result<(), SendError> { + self.unbounded_send(msg) + } + + /// Sends the provided message along this channel. + /// + /// This is an unbounded sender, so this function differs from `Sink::send` + /// by ensuring the return type reflects that the channel is always ready to + /// receive messages. + pub fn unbounded_send(&self, msg: T) -> Result<(), SendError> { + self.0.do_send_nb(msg) + } +} + +impl Sink for UnboundedSender { + type SinkItem = T; + type SinkError = SendError; + + fn start_send(&mut self, msg: T) -> StartSend> { + self.0.start_send(msg) + } + + fn poll_complete(&mut self) -> Poll<(), SendError> { + self.0.poll_complete() + } + + fn close(&mut self) -> Poll<(), SendError> { + Ok(Async::Ready(())) + } +} + +impl<'a, T> Sink for &'a UnboundedSender { + type SinkItem = T; + type SinkError = SendError; + + fn start_send(&mut self, msg: T) -> StartSend> { + self.0.do_send_nb(msg)?; + Ok(AsyncSink::Ready) + } + + fn poll_complete(&mut self) -> Poll<(), SendError> { + Ok(Async::Ready(())) + } + + fn close(&mut self) -> Poll<(), SendError> { + Ok(Async::Ready(())) + } +} + +impl Clone for UnboundedSender { + fn clone(&self) -> UnboundedSender { + UnboundedSender(self.0.clone()) + } +} + + +impl Clone for Sender { + fn clone(&self) -> Sender { + // Since this atomic op isn't actually guarding any memory and we don't + // care about any orderings besides the ordering on the single atomic + // variable, a relaxed ordering is acceptable. + let mut curr = self.inner.num_senders.load(SeqCst); + + loop { + // If the maximum number of senders has been reached, then fail + if curr == self.inner.max_senders() { + panic!("cannot clone `Sender` -- too many outstanding senders"); + } + + debug_assert!(curr < self.inner.max_senders()); + + let next = curr + 1; + let actual = self.inner.num_senders.compare_and_swap(curr, next, SeqCst); + + // The ABA problem doesn't matter here. We only care that the + // number of senders never exceeds the maximum. + if actual == curr { + return Sender { + inner: self.inner.clone(), + sender_task: Arc::new(Mutex::new(SenderTask::new())), + maybe_parked: false, + }; + } + + curr = actual; + } + } +} + +impl Drop for Sender { + fn drop(&mut self) { + // Ordering between variables don't matter here + let prev = self.inner.num_senders.fetch_sub(1, SeqCst); + + if prev == 1 { + let _ = self.do_send(None, false); + } + } +} + +/* + * + * ===== impl Receiver ===== + * + */ + +impl Receiver { + /// Closes the receiving half + /// + /// 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) { + let mut curr = self.inner.state.load(SeqCst); + + loop { + let mut state = decode_state(curr); + + if !state.is_open { + break + } + + state.is_open = false; + + let next = encode_state(&state); + match self.inner.state.compare_exchange(curr, next, SeqCst, SeqCst) { + Ok(_) => break, + Err(actual) => curr = actual, + } + } + + // Wake up any threads waiting as they'll see that we've closed the + // channel and will continue on their merry way. + loop { + match unsafe { self.inner.parked_queue.pop() } { + PopResult::Data(task) => { + task.lock().unwrap().notify(); + } + PopResult::Empty => break, + PopResult::Inconsistent => thread::yield_now(), + } + } + } + + fn next_message(&mut self) -> Async> { + // Pop off a message + loop { + match unsafe { self.inner.message_queue.pop() } { + PopResult::Data(msg) => { + return Async::Ready(msg); + } + PopResult::Empty => { + // The queue is empty, return NotReady + return Async::NotReady; + } + PopResult::Inconsistent => { + // Inconsistent means that there will be a message to pop + // in a short time. This branch can only be reached if + // values are being produced from another thread, so there + // are a few ways that we can deal with this: + // + // 1) Spin + // 2) thread::yield_now() + // 3) task::current().unwrap() & return NotReady + // + // For now, thread::yield_now() is used, but it would + // probably be better to spin a few times then yield. + thread::yield_now(); + } + } + } + } + + // Unpark a single task handle if there is one pending in the parked queue + fn unpark_one(&mut self) { + loop { + match unsafe { self.inner.parked_queue.pop() } { + PopResult::Data(task) => { + task.lock().unwrap().notify(); + return; + } + PopResult::Empty => { + // Queue empty, no task to wake up. + return; + } + PopResult::Inconsistent => { + // Same as above + thread::yield_now(); + } + } + } + } + + // Try to park the receiver task + fn try_park(&self) -> TryPark { + let curr = self.inner.state.load(SeqCst); + let state = decode_state(curr); + + // If the channel is closed, then there is no need to park. + if !state.is_open && state.num_messages == 0 { + return TryPark::Closed; + } + + // First, track the task in the `recv_task` slot + let mut recv_task = self.inner.recv_task.lock().unwrap(); + + if recv_task.unparked { + // Consume the `unpark` signal without actually parking + recv_task.unparked = false; + return TryPark::NotEmpty; + } + + recv_task.task = Some(task::current()); + TryPark::Parked + } + + fn dec_num_messages(&self) { + let mut curr = self.inner.state.load(SeqCst); + + loop { + let mut state = decode_state(curr); + + state.num_messages -= 1; + + let next = encode_state(&state); + match self.inner.state.compare_exchange(curr, next, SeqCst, SeqCst) { + Ok(_) => break, + Err(actual) => curr = actual, + } + } + } +} + +impl Stream for Receiver { + type Item = T; + type Error = (); + + fn poll(&mut self) -> Poll, ()> { + loop { + // Try to read a message off of the message queue. + let msg = match self.next_message() { + Async::Ready(msg) => msg, + Async::NotReady => { + // There are no messages to read, in this case, attempt to + // park. The act of parking will verify that the channel is + // still empty after the park operation has completed. + match self.try_park() { + TryPark::Parked => { + // The task was parked, and the channel is still + // empty, return NotReady. + return Ok(Async::NotReady); + } + TryPark::Closed => { + // The channel is closed, there will be no further + // messages. + return Ok(Async::Ready(None)); + } + TryPark::NotEmpty => { + // A message has been sent while attempting to + // park. Loop again, the next iteration is + // guaranteed to get the message. + continue; + } + } + } + }; + + // If there are any parked task handles in the parked queue, pop + // one and unpark it. + self.unpark_one(); + + // Decrement number of messages + self.dec_num_messages(); + + // Return the message + return Ok(Async::Ready(msg)); + } + } +} + +impl Drop for Receiver { + fn drop(&mut self) { + // Drain the channel of all pending messages + self.close(); + while self.next_message().is_ready() { + // ... + } + } +} + +/* + * + * ===== impl Inner ===== + * + */ + +impl Inner { + // The return value is such that the total number of messages that can be + // enqueued into the channel will never exceed MAX_CAPACITY + fn max_senders(&self) -> usize { + match self.buffer { + Some(buffer) => MAX_CAPACITY - buffer, + None => MAX_BUFFER, + } + } +} + +unsafe impl Send for Inner {} +unsafe impl Sync for Inner {} + +/* + * + * ===== Helpers ===== + * + */ + +fn decode_state(num: usize) -> State { + State { + is_open: num & OPEN_MASK == OPEN_MASK, + num_messages: num & MAX_CAPACITY, + } +} + +fn encode_state(state: &State) -> usize { + let mut num = state.num_messages; + + if state.is_open { + num |= OPEN_MASK; + } + + num +} diff --git a/tokio-channel/src/mpsc/queue.rs b/tokio-channel/src/mpsc/queue.rs new file mode 100644 index 000000000..9ff6bcf87 --- /dev/null +++ b/tokio-channel/src/mpsc/queue.rs @@ -0,0 +1,151 @@ +/* Copyright (c) 2010-2011 Dmitry Vyukov. All rights reserved. + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * 1. Redistributions of source code must retain the above copyright notice, + * this list of conditions and the following disclaimer. + * + * 2. Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * + * THIS SOFTWARE IS PROVIDED BY DMITRY VYUKOV "AS IS" AND ANY EXPRESS OR IMPLIED + * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF + * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT + * SHALL DMITRY VYUKOV OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, + * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR + * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF + * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE + * OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF + * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + * The views and conclusions contained in the software and documentation are + * those of the authors and should not be interpreted as representing official + * policies, either expressed or implied, of Dmitry Vyukov. + */ + +//! A mostly lock-free multi-producer, single consumer queue. +//! +//! This module contains an implementation of a concurrent MPSC queue. This +//! queue can be used to share data between threads, and is also used as the +//! building block of channels in rust. +//! +//! Note that the current implementation of this queue has a caveat of the `pop` +//! method, and see the method for more information about it. Due to this +//! caveat, this queue may not be appropriate for all use-cases. + +// http://www.1024cores.net/home/lock-free-algorithms +// /queues/non-intrusive-mpsc-node-based-queue + +// NOTE: this implementation is lifted from the standard library and only +// slightly modified + +pub use self::PopResult::*; +use std::prelude::v1::*; + +use std::cell::UnsafeCell; +use std::ptr; +use std::sync::atomic::{AtomicPtr, Ordering}; + +/// A result of the `pop` function. +pub enum PopResult { + /// Some data has been popped + Data(T), + /// The queue is empty + Empty, + /// The queue is in an inconsistent state. Popping data should succeed, but + /// some pushers have yet to make enough progress in order allow a pop to + /// succeed. It is recommended that a pop() occur "in the near future" in + /// order to see if the sender has made progress or not + Inconsistent, +} + +#[derive(Debug)] +struct Node { + next: AtomicPtr>, + value: Option, +} + +/// The multi-producer single-consumer structure. This is not cloneable, but it +/// may be safely shared so long as it is guaranteed that there is only one +/// popper at a time (many pushers are allowed). +#[derive(Debug)] +pub struct Queue { + head: AtomicPtr>, + tail: UnsafeCell<*mut Node>, +} + +unsafe impl Send for Queue { } +unsafe impl Sync for Queue { } + +impl Node { + unsafe fn new(v: Option) -> *mut Node { + Box::into_raw(Box::new(Node { + next: AtomicPtr::new(ptr::null_mut()), + value: v, + })) + } +} + +impl Queue { + /// Creates a new queue that is safe to share among multiple producers and + /// one consumer. + pub fn new() -> Queue { + let stub = unsafe { Node::new(None) }; + Queue { + head: AtomicPtr::new(stub), + tail: UnsafeCell::new(stub), + } + } + + /// Pushes a new value onto this queue. + pub fn push(&self, t: T) { + unsafe { + let n = Node::new(Some(t)); + let prev = self.head.swap(n, Ordering::AcqRel); + (*prev).next.store(n, Ordering::Release); + } + } + + /// Pops some data from this queue. + /// + /// Note that the current implementation means that this function cannot + /// return `Option`. It is possible for this queue to be in an + /// inconsistent state where many pushes have succeeded and completely + /// finished, but pops cannot return `Some(t)`. This inconsistent state + /// happens when a pusher is preempted at an inopportune moment. + /// + /// This inconsistent state means that this queue does indeed have data, but + /// it does not currently have access to it at this time. + /// + /// This function is unsafe because only one thread can call it at a time. + pub unsafe fn pop(&self) -> PopResult { + let tail = *self.tail.get(); + let next = (*tail).next.load(Ordering::Acquire); + + if !next.is_null() { + *self.tail.get() = next; + assert!((*tail).value.is_none()); + assert!((*next).value.is_some()); + let ret = (*next).value.take().unwrap(); + drop(Box::from_raw(tail)); + return Data(ret); + } + + if self.head.load(Ordering::Acquire) == tail {Empty} else {Inconsistent} + } +} + +impl Drop for Queue { + fn drop(&mut self) { + unsafe { + let mut cur = *self.tail.get(); + while !cur.is_null() { + let next = (*cur).next.load(Ordering::Relaxed); + drop(Box::from_raw(cur)); + cur = next; + } + } + } +} diff --git a/tokio-channel/src/oneshot.rs b/tokio-channel/src/oneshot.rs new file mode 100644 index 000000000..1e43e26ba --- /dev/null +++ b/tokio-channel/src/oneshot.rs @@ -0,0 +1,426 @@ +//! A one-shot, futures-aware channel + +use lock::Lock; + +use futures::{Future, Poll, Async}; +use futures::task::{self, Task}; + +use std::sync::Arc; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering::SeqCst; +use std::error::Error; +use std::fmt; + +/// A future representing the completion of a computation happening elsewhere in +/// memory. +/// +/// This is created by the `oneshot::channel` function. +#[must_use = "futures do nothing unless polled"] +#[derive(Debug)] +pub struct Receiver { + inner: Arc>, +} + +/// Represents the completion half of a oneshot through which the result of a +/// computation is signaled. +/// +/// This is created by the `oneshot::channel` function. +#[derive(Debug)] +pub struct Sender { + inner: Arc>, +} + +/// Internal state of the `Receiver`/`Sender` pair above. This is all used as +/// the internal synchronization between the two for send/recv operations. +#[derive(Debug)] +struct Inner { + /// Indicates whether this oneshot is complete yet. This is filled in both + /// by `Sender::drop` and by `Receiver::drop`, and both sides interpret it + /// appropriately. + /// + /// For `Receiver`, if this is `true`, then it's guaranteed that `data` is + /// unlocked and ready to be inspected. + /// + /// For `Sender` if this is `true` then the oneshot has gone away and it + /// can return ready from `poll_cancel`. + complete: AtomicBool, + + /// The actual data being transferred as part of this `Receiver`. This is + /// filled in by `Sender::complete` and read by `Receiver::poll`. + /// + /// Note that this is protected by `Lock`, but it is in theory safe to + /// replace with an `UnsafeCell` as it's actually protected by `complete` + /// above. I wouldn't recommend doing this, however, unless someone is + /// supremely confident in the various atomic orderings here and there. + data: Lock>, + + /// Field to store the task which is blocked in `Receiver::poll`. + /// + /// This is filled in when a oneshot is polled but not ready yet. Note that + /// the `Lock` here, unlike in `data` above, is important to resolve races. + /// Both the `Receiver` and the `Sender` halves understand that if they + /// can't acquire the lock then some important interference is happening. + rx_task: Lock>, + + /// Like `rx_task` above, except for the task blocked in + /// `Sender::poll_cancel`. Additionally, `Lock` cannot be `UnsafeCell`. + tx_task: Lock>, +} + +/// Creates a new futures-aware, one-shot channel. +/// +/// This function is similar to Rust's channels found in the standard library. +/// Two halves are returned, the first of which is a `Sender` handle, used to +/// signal the end of a computation and provide its value. The second half is a +/// `Receiver` which implements the `Future` trait, resolving to the value that +/// was given to the `Sender` handle. +/// +/// Each half can be separately owned and sent across threads/tasks. +/// +/// # Examples +/// +/// ``` +/// extern crate tokio_channel; +/// extern crate futures; +/// +/// use tokio_channel::oneshot; +/// use futures::*; +/// use std::thread; +/// +/// # fn main() { +/// let (p, c) = oneshot::channel::(); +/// +/// thread::spawn(|| { +/// c.map(|i| { +/// println!("got: {}", i); +/// }).wait(); +/// }); +/// +/// p.send(3).unwrap(); +/// # } +/// ``` +pub fn channel() -> (Sender, Receiver) { + let inner = Arc::new(Inner::new()); + let receiver = Receiver { + inner: inner.clone(), + }; + let sender = Sender { + inner: inner, + }; + (sender, receiver) +} + +impl Inner { + fn new() -> Inner { + Inner { + complete: AtomicBool::new(false), + data: Lock::new(None), + rx_task: Lock::new(None), + tx_task: Lock::new(None), + } + } + + fn send(&self, t: T) -> Result<(), T> { + if self.complete.load(SeqCst) { + return Err(t) + } + + // Note that this lock acquisition may fail if the receiver + // is closed and sets the `complete` flag to true, whereupon + // the receiver may call `poll()`. + if let Some(mut slot) = self.data.try_lock() { + assert!(slot.is_none()); + *slot = Some(t); + drop(slot); + + // If the receiver called `close()` between the check at the + // start of the function, and the lock being released, then + // the receiver may not be around to receive it, so try to + // pull it back out. + if self.complete.load(SeqCst) { + // If lock acquisition fails, then receiver is actually + // receiving it, so we're good. + if let Some(mut slot) = self.data.try_lock() { + if let Some(t) = slot.take() { + return Err(t); + } + } + } + Ok(()) + } else { + // Must have been closed + Err(t) + } + } + + fn poll_cancel(&self) -> Poll<(), ()> { + // Fast path up first, just read the flag and see if our other half is + // gone. This flag is set both in our destructor and the oneshot + // destructor, but our destructor hasn't run yet so if it's set then the + // oneshot is gone. + if self.complete.load(SeqCst) { + return Ok(Async::Ready(())) + } + + // If our other half is not gone then we need to park our current task + // and move it into the `notify_cancel` slot to get notified when it's + // actually gone. + // + // If `try_lock` fails, then the `Receiver` is in the process of using + // it, so we can deduce that it's now in the process of going away and + // hence we're canceled. If it succeeds then we just store our handle. + // + // Crucially we then check `oneshot_gone` *again* before we return. + // While we were storing our handle inside `notify_cancel` the `Receiver` + // may have been dropped. The first thing it does is set the flag, and + // if it fails to acquire the lock it assumes that we'll see the flag + // later on. So... we then try to see the flag later on! + let handle = task::current(); + match self.tx_task.try_lock() { + Some(mut p) => *p = Some(handle), + None => return Ok(Async::Ready(())), + } + if self.complete.load(SeqCst) { + Ok(Async::Ready(())) + } else { + Ok(Async::NotReady) + } + } + + fn is_canceled(&self) -> bool { + self.complete.load(SeqCst) + } + + fn drop_tx(&self) { + // Flag that we're a completed `Sender` and try to wake up a receiver. + // Whether or not we actually stored any data will get picked up and + // translated to either an item or cancellation. + // + // Note that if we fail to acquire the `rx_task` lock then that means + // we're in one of two situations: + // + // 1. The receiver is trying to block in `poll` + // 2. The receiver is being dropped + // + // In the first case it'll check the `complete` flag after it's done + // blocking to see if it succeeded. In the latter case we don't need to + // wake up anyone anyway. So in both cases it's ok to ignore the `None` + // case of `try_lock` and bail out. + // + // The first case crucially depends on `Lock` using `SeqCst` ordering + // under the hood. If it instead used `Release` / `Acquire` ordering, + // then it would not necessarily synchronize with `inner.complete` + // and deadlock might be possible, as was observed in + // https://github.com/rust-lang-nursery/futures-rs/pull/219. + self.complete.store(true, SeqCst); + if let Some(mut slot) = self.rx_task.try_lock() { + if let Some(task) = slot.take() { + drop(slot); + task.notify(); + } + } + } + + fn close_rx(&self) { + // Flag our completion and then attempt to wake up the sender if it's + // blocked. See comments in `drop` below for more info + self.complete.store(true, SeqCst); + if let Some(mut handle) = self.tx_task.try_lock() { + if let Some(task) = handle.take() { + drop(handle); + task.notify() + } + } + } + + fn recv(&self) -> Poll { + let mut done = false; + + // Check to see if some data has arrived. If it hasn't then we need to + // block our task. + // + // Note that the acquisition of the `rx_task` lock might fail below, but + // the only situation where this can happen is during `Sender::drop` + // when we are indeed completed already. If that's happening then we + // know we're completed so keep going. + if self.complete.load(SeqCst) { + done = true; + } else { + let task = task::current(); + match self.rx_task.try_lock() { + Some(mut slot) => *slot = Some(task), + None => done = true, + } + } + + // If we're `done` via one of the paths above, then look at the data and + // figure out what the answer is. If, however, we stored `rx_task` + // successfully above we need to check again if we're completed in case + // a message was sent while `rx_task` was locked and couldn't notify us + // otherwise. + // + // If we're not done, and we're not complete, though, then we've + // successfully blocked our task and we return `NotReady`. + if done || self.complete.load(SeqCst) { + // If taking the lock fails, the sender will realise that the we're + // `done` when it checks the `complete` flag on the way out, and will + // treat the send as a failure. + if let Some(mut slot) = self.data.try_lock() { + if let Some(data) = slot.take() { + return Ok(data.into()); + } + } + Err(Canceled) + } else { + Ok(Async::NotReady) + } + } + + fn drop_rx(&self) { + // Indicate to the `Sender` that we're done, so any future calls to + // `poll_cancel` are weeded out. + self.complete.store(true, SeqCst); + + // If we've blocked a task then there's no need for it to stick around, + // so we need to drop it. If this lock acquisition fails, though, then + // it's just because our `Sender` is trying to take the task, so we + // let them take care of that. + if let Some(mut slot) = self.rx_task.try_lock() { + let task = slot.take(); + drop(slot); + drop(task); + } + + // Finally, if our `Sender` wants to get notified of us going away, it + // would have stored something in `tx_task`. Here we try to peel that + // out and unpark it. + // + // Note that the `try_lock` here may fail, but only if the `Sender` is + // in the process of filling in the task. If that happens then we + // already flagged `complete` and they'll pick that up above. + if let Some(mut handle) = self.tx_task.try_lock() { + if let Some(task) = handle.take() { + drop(handle); + task.notify() + } + } + } +} + +impl Sender { + #[deprecated(note = "renamed to `send`", since = "0.1.11")] + #[doc(hidden)] + #[cfg(feature = "with-deprecated")] + pub fn complete(self, t: T) { + drop(self.send(t)); + } + + /// Completes this oneshot with a successful result. + /// + /// This function will consume `self` and indicate to the other end, the + /// `Receiver`, that the value provided is the result of the computation this + /// represents. + /// + /// If the value is successfully enqueued for the remote end to receive, + /// then `Ok(())` is returned. If the receiving end was deallocated before + /// this function was called, however, then `Err` is returned with the value + /// provided. + pub fn send(self, t: T) -> Result<(), T> { + self.inner.send(t) + } + + /// Polls this `Sender` half to detect whether the `Receiver` this has + /// paired with has gone away. + /// + /// This function can be used to learn about when the `Receiver` (consumer) + /// half has gone away and nothing will be able to receive a message sent + /// from `send`. + /// + /// If `Ready` is returned then it means that the `Receiver` has disappeared + /// and the result this `Sender` would otherwise produce should no longer + /// be produced. + /// + /// If `NotReady` is returned then the `Receiver` is still alive and may be + /// able to receive a message if sent. The current task, however, is + /// scheduled to receive a notification if the corresponding `Receiver` goes + /// away. + /// + /// # Panics + /// + /// Like `Future::poll`, this function will panic if it's not called from + /// within the context of a task. In other words, this should only ever be + /// called from inside another future. + /// + /// If you're calling this function from a context that does not have a + /// task, then you can use the `is_canceled` API instead. + pub fn poll_cancel(&mut self) -> Poll<(), ()> { + self.inner.poll_cancel() + } + + /// Tests to see whether this `Sender`'s corresponding `Receiver` + /// has gone away. + /// + /// This function can be used to learn about when the `Receiver` (consumer) + /// half has gone away and nothing will be able to receive a message sent + /// from `send`. + /// + /// Note that this function is intended to *not* be used in the context of a + /// future. If you're implementing a future you probably want to call the + /// `poll_cancel` function which will block the current task if the + /// cancellation hasn't happened yet. This can be useful when working on a + /// non-futures related thread, though, which would otherwise panic if + /// `poll_cancel` were called. + pub fn is_canceled(&self) -> bool { + self.inner.is_canceled() + } +} + +impl Drop for Sender { + fn drop(&mut self) { + self.inner.drop_tx() + } +} + +/// Error returned from a `Receiver` whenever the corresponding `Sender` +/// is dropped. +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub struct Canceled; + +impl fmt::Display for Canceled { + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + write!(fmt, "oneshot canceled") + } +} + +impl Error for Canceled { + fn description(&self) -> &str { + "oneshot canceled" + } +} + +impl Receiver { + /// Gracefully close this receiver, preventing sending any future messages. + /// + /// Any `send` operation which happens after this method returns is + /// guaranteed to fail. Once this method is called the normal `poll` method + /// can be used to determine whether a message was actually sent or not. If + /// `Canceled` is returned from `poll` then no message was sent. + pub fn close(&mut self) { + self.inner.close_rx() + } +} + +impl Future for Receiver { + type Item = T; + type Error = Canceled; + + fn poll(&mut self) -> Poll { + self.inner.recv() + } +} + +impl Drop for Receiver { + fn drop(&mut self) { + self.inner.drop_rx() + } +} diff --git a/tokio-channel/tests/mpsc-close.rs b/tokio-channel/tests/mpsc-close.rs new file mode 100644 index 000000000..3290dae4c --- /dev/null +++ b/tokio-channel/tests/mpsc-close.rs @@ -0,0 +1,22 @@ +extern crate tokio_channel; +extern crate futures; + + +use tokio_channel::mpsc::*; +use futures::prelude::*; +use std::thread; + +#[test] +fn smoke() { + let (mut sender, receiver) = channel(1); + + let t = thread::spawn(move ||{ + while let Ok(s) = sender.send(42).wait() { + sender = s; + } + }); + + receiver.take(3).for_each(|_| Ok(())).wait().unwrap(); + + t.join().unwrap() +} diff --git a/tokio-channel/tests/mpsc.rs b/tokio-channel/tests/mpsc.rs new file mode 100644 index 000000000..781cc971b --- /dev/null +++ b/tokio-channel/tests/mpsc.rs @@ -0,0 +1,481 @@ +extern crate tokio_channel; +#[macro_use] +extern crate futures; + +mod support; +use support::*; + +use tokio_channel::mpsc; +use tokio_channel::oneshot; + +use futures::prelude::*; +use futures::future::lazy; + +use std::thread; +use std::sync::{Arc, Mutex}; +use std::sync::atomic::{AtomicUsize, Ordering}; + +trait AssertSend: Send {} +impl AssertSend for mpsc::Sender {} +impl AssertSend for mpsc::Receiver {} + +#[test] +fn send_recv() { + let (tx, rx) = mpsc::channel::(16); + let mut rx = rx.wait(); + + tx.send(1).wait().unwrap(); + + assert_eq!(rx.next().unwrap(), Ok(1)); +} + +#[test] +fn send_recv_no_buffer() { + let (mut tx, mut rx) = mpsc::channel::(0); + + // Run on a task context + lazy(move || { + assert!(tx.poll_complete().unwrap().is_ready()); + assert!(tx.poll_ready().unwrap().is_ready()); + + // Send first message + let res = tx.start_send(1).unwrap(); + assert!(is_ready(&res)); + assert!(tx.poll_ready().unwrap().is_not_ready()); + + // Send second message + let res = tx.start_send(2).unwrap(); + assert!(!is_ready(&res)); + + // Take the value + assert_eq!(rx.poll().unwrap(), Async::Ready(Some(1))); + assert!(tx.poll_ready().unwrap().is_ready()); + + let res = tx.start_send(2).unwrap(); + assert!(is_ready(&res)); + assert!(tx.poll_ready().unwrap().is_not_ready()); + + // Take the value + assert_eq!(rx.poll().unwrap(), Async::Ready(Some(2))); + assert!(tx.poll_ready().unwrap().is_ready()); + + Ok::<(), ()>(()) + }).wait().unwrap(); +} + +#[test] +fn send_shared_recv() { + let (tx1, rx) = mpsc::channel::(16); + let tx2 = tx1.clone(); + let mut rx = rx.wait(); + + tx1.send(1).wait().unwrap(); + assert_eq!(rx.next().unwrap(), Ok(1)); + + tx2.send(2).wait().unwrap(); + assert_eq!(rx.next().unwrap(), Ok(2)); +} + +#[test] +fn send_recv_threads() { + let (tx, rx) = mpsc::channel::(16); + let mut rx = rx.wait(); + + thread::spawn(move|| { + tx.send(1).wait().unwrap(); + }); + + assert_eq!(rx.next().unwrap(), Ok(1)); +} + +#[test] +fn send_recv_threads_no_capacity() { + let (tx, rx) = mpsc::channel::(0); + let mut rx = rx.wait(); + + let (readytx, readyrx) = mpsc::channel::<()>(2); + let mut readyrx = readyrx.wait(); + let t = thread::spawn(move|| { + let readytx = readytx.sink_map_err(|_| panic!()); + let (a, b) = tx.send(1).join(readytx.send(())).wait().unwrap(); + a.send(2).join(b.send(())).wait().unwrap(); + }); + + drop(readyrx.next().unwrap()); + assert_eq!(rx.next().unwrap(), Ok(1)); + drop(readyrx.next().unwrap()); + assert_eq!(rx.next().unwrap(), Ok(2)); + + t.join().unwrap(); +} + +#[test] +fn recv_close_gets_none() { + let (mut tx, mut rx) = mpsc::channel::(10); + + // Run on a task context + lazy(move || { + rx.close(); + + assert_eq!(rx.poll(), Ok(Async::Ready(None))); + assert!(tx.poll_ready().is_err()); + + drop(tx); + + Ok::<(), ()>(()) + }).wait().unwrap(); +} + + +#[test] +fn tx_close_gets_none() { + let (_, mut rx) = mpsc::channel::(10); + + // Run on a task context + lazy(move || { + assert_eq!(rx.poll(), Ok(Async::Ready(None))); + assert_eq!(rx.poll(), Ok(Async::Ready(None))); + + Ok::<(), ()>(()) + }).wait().unwrap(); +} + +#[test] +fn stress_shared_unbounded() { + const AMT: u32 = 10000; + const NTHREADS: u32 = 8; + let (tx, rx) = mpsc::unbounded::(); + let mut rx = rx.wait(); + + let t = thread::spawn(move|| { + for _ in 0..AMT * NTHREADS { + assert_eq!(rx.next().unwrap(), Ok(1)); + } + + if rx.next().is_some() { + panic!(); + } + }); + + for _ in 0..NTHREADS { + let tx = tx.clone(); + + thread::spawn(move|| { + for _ in 0..AMT { + tx.unbounded_send(1).unwrap(); + } + }); + } + + drop(tx); + + t.join().ok().unwrap(); +} + +#[test] +fn stress_shared_bounded_hard() { + const AMT: u32 = 10000; + const NTHREADS: u32 = 8; + let (tx, rx) = mpsc::channel::(0); + let mut rx = rx.wait(); + + let t = thread::spawn(move|| { + for _ in 0..AMT * NTHREADS { + assert_eq!(rx.next().unwrap(), Ok(1)); + } + + if rx.next().is_some() { + panic!(); + } + }); + + for _ in 0..NTHREADS { + let mut tx = tx.clone(); + + thread::spawn(move|| { + for _ in 0..AMT { + tx = tx.send(1).wait().unwrap(); + } + }); + } + + drop(tx); + + t.join().ok().unwrap(); +} + +#[test] +fn stress_receiver_multi_task_bounded_hard() { + const AMT: usize = 10_000; + const NTHREADS: u32 = 2; + + let (mut tx, rx) = mpsc::channel::(0); + let rx = Arc::new(Mutex::new(Some(rx))); + let n = Arc::new(AtomicUsize::new(0)); + + let mut th = vec![]; + + for _ in 0..NTHREADS { + let rx = rx.clone(); + let n = n.clone(); + + let t = thread::spawn(move || { + let mut i = 0; + + loop { + i += 1; + let mut lock = rx.lock().ok().unwrap(); + + match lock.take() { + Some(mut rx) => { + if i % 5 == 0 { + let (item, rest) = rx.into_future().wait().ok().unwrap(); + + if item.is_none() { + break; + } + + n.fetch_add(1, Ordering::Relaxed); + *lock = Some(rest); + } else { + // Just poll + let n = n.clone(); + let r = lazy(move || { + let r = match rx.poll().unwrap() { + Async::Ready(Some(_)) => { + n.fetch_add(1, Ordering::Relaxed); + *lock = Some(rx); + false + } + Async::Ready(None) => { + true + } + Async::NotReady => { + *lock = Some(rx); + false + } + }; + + Ok::(r) + }).wait().unwrap(); + + if r { + break; + } + } + } + None => break, + } + } + }); + + th.push(t); + } + + for i in 0..AMT { + tx = tx.send(i).wait().unwrap(); + } + + drop(tx); + + for t in th { + t.join().unwrap(); + } + + assert_eq!(AMT, n.load(Ordering::Relaxed)); +} + +/// Stress test that receiver properly receives all the messages +/// after sender dropped. +#[test] +fn stress_drop_sender() { + fn list() -> Box> { + let (tx, rx) = mpsc::channel(1); + tx.send(Ok(1)) + .and_then(|tx| tx.send(Ok(2))) + .and_then(|tx| tx.send(Ok(3))) + .forget(); + Box::new(rx.then(|r| r.unwrap())) + } + + for _ in 0..10000 { + assert_eq!(list().wait().collect::, _>>(), + Ok(vec![1, 2, 3])); + } +} + +/// Stress test that after receiver dropped, +/// no messages are lost. +fn stress_close_receiver_iter() { + let (tx, rx) = mpsc::unbounded(); + let (unwritten_tx, unwritten_rx) = std::sync::mpsc::channel(); + let th = thread::spawn(move || { + for i in 1.. { + if let Err(_) = tx.unbounded_send(i) { + unwritten_tx.send(i).expect("unwritten_tx"); + return; + } + } + }); + + let mut rx = rx.wait(); + + // Read one message to make sure thread effectively started + assert_eq!(Some(Ok(1)), rx.next()); + + rx.get_mut().close(); + + for i in 2.. { + match rx.next() { + Some(Ok(r)) => assert!(i == r), + Some(Err(_)) => unreachable!(), + None => { + let unwritten = unwritten_rx.recv().expect("unwritten_rx"); + assert_eq!(unwritten, i); + th.join().unwrap(); + return; + } + } + } +} + +#[test] +fn stress_close_receiver() { + for _ in 0..10000 { + stress_close_receiver_iter(); + } +} + +/// Tests that after `poll_ready` indicates capacity a channel can always send without waiting. +#[test] +fn stress_poll_ready() { + // A task which checks channel capacity using poll_ready, and pushes items onto the channel when + // ready. + struct SenderTask { + sender: mpsc::Sender, + count: u32, + } + impl Future for SenderTask { + type Item = (); + type Error = (); + fn poll(&mut self) -> Poll<(), ()> { + // In a loop, check if the channel is ready. If so, push an item onto the channel + // (asserting that it doesn't attempt to block). + while self.count > 0 { + try_ready!(self.sender.poll_ready().map_err(|_| ())); + assert!(self.sender.start_send(self.count).unwrap().is_ready()); + self.count -= 1; + } + Ok(Async::Ready(())) + } + } + + const AMT: u32 = 1000; + const NTHREADS: u32 = 8; + + /// Run a stress test using the specified channel capacity. + fn stress(capacity: usize) { + let (tx, rx) = mpsc::channel(capacity); + let mut threads = Vec::new(); + for _ in 0..NTHREADS { + let sender = tx.clone(); + threads.push(thread::spawn(move || { + SenderTask { + sender: sender, + count: AMT, + }.wait() + })); + } + drop(tx); + + let mut rx = rx.wait(); + for _ in 0..AMT * NTHREADS { + assert!(rx.next().is_some()); + } + + assert!(rx.next().is_none()); + + for thread in threads { + thread.join().unwrap().unwrap(); + } + } + + stress(0); + stress(1); + stress(8); + stress(16); +} + +fn is_ready(res: &AsyncSink) -> bool { + match *res { + AsyncSink::Ready => true, + _ => false, + } +} + +#[test] +fn try_send_1() { + const N: usize = 3000; + let (mut tx, rx) = mpsc::channel(0); + + let t = thread::spawn(move || { + for i in 0..N { + loop { + if tx.try_send(i).is_ok() { + break + } + } + } + }); + for (i, j) in rx.wait().enumerate() { + assert_eq!(i, j.unwrap()); + } + t.join().unwrap(); +} + +#[test] +fn try_send_2() { + let (mut tx, rx) = mpsc::channel(0); + + tx.try_send("hello").unwrap(); + + let (readytx, readyrx) = oneshot::channel::<()>(); + + let th = thread::spawn(|| { + lazy(|| { + assert!(tx.start_send("fail").unwrap().is_not_ready()); + Ok::<_, ()>(()) + }).wait().unwrap(); + + drop(readytx); + tx.send("goodbye").wait().unwrap(); + }); + + let mut rx = rx.wait(); + + drop(readyrx.wait()); + assert_eq!(rx.next(), Some(Ok("hello"))); + assert_eq!(rx.next(), Some(Ok("goodbye"))); + assert!(rx.next().is_none()); + + th.join().unwrap(); +} + +#[test] +fn try_send_fail() { + let (mut tx, rx) = mpsc::channel(0); + let mut rx = rx.wait(); + + tx.try_send("hello").unwrap(); + + // This should fail + assert!(tx.try_send("fail").is_err()); + + assert_eq!(rx.next(), Some(Ok("hello"))); + + tx.try_send("goodbye").unwrap(); + drop(tx); + + assert_eq!(rx.next(), Some(Ok("goodbye"))); + assert!(rx.next().is_none()); +} diff --git a/tokio-channel/tests/oneshot.rs b/tokio-channel/tests/oneshot.rs new file mode 100644 index 000000000..2997f8246 --- /dev/null +++ b/tokio-channel/tests/oneshot.rs @@ -0,0 +1,124 @@ +extern crate tokio_channel; +extern crate futures; + +mod support; +use support::*; + +use tokio_channel::oneshot::*; + +use futures::prelude::*; +use futures::future::{lazy, ok}; + +use std::sync::mpsc; +use std::thread; + +#[test] +fn smoke_poll() { + let (mut tx, rx) = channel::(); + + lazy(|| { + assert!(tx.poll_cancel().unwrap().is_not_ready()); + assert!(tx.poll_cancel().unwrap().is_not_ready()); + drop(rx); + assert!(tx.poll_cancel().unwrap().is_ready()); + assert!(tx.poll_cancel().unwrap().is_ready()); + ok::<(), ()>(()) + }).wait().unwrap(); +} + +#[test] +fn cancel_notifies() { + let (tx, rx) = channel::(); + let (tx2, rx2) = mpsc::channel(); + + WaitForCancel { tx: tx }.then(move |v| tx2.send(v)).forget(); + drop(rx); + rx2.recv().unwrap().unwrap(); +} + +struct WaitForCancel { + tx: Sender, +} + +impl Future for WaitForCancel { + type Item = (); + type Error = (); + + fn poll(&mut self) -> Poll<(), ()> { + self.tx.poll_cancel() + } +} + +#[test] +fn cancel_lots() { + let (tx, rx) = mpsc::channel::<(Sender<_>, mpsc::Sender<_>)>(); + let t = thread::spawn(move || { + for (tx, tx2) in rx { + WaitForCancel { tx: tx }.then(move |v| tx2.send(v)).forget(); + } + + }); + + for _ in 0..20000 { + let (otx, orx) = channel::(); + let (tx2, rx2) = mpsc::channel(); + tx.send((otx, tx2)).unwrap(); + drop(orx); + rx2.recv().unwrap().unwrap(); + } + drop(tx); + + t.join().unwrap(); +} + +#[test] +fn close() { + let (mut tx, mut rx) = channel::(); + rx.close(); + assert!(rx.poll().is_err()); + assert!(tx.poll_cancel().unwrap().is_ready()); +} + +#[test] +fn close_wakes() { + let (tx, mut rx) = channel::(); + let (tx2, rx2) = mpsc::channel(); + let t = thread::spawn(move || { + rx.close(); + rx2.recv().unwrap(); + }); + WaitForCancel { tx: tx }.wait().unwrap(); + tx2.send(()).unwrap(); + t.join().unwrap(); +} + +#[test] +fn is_canceled() { + let (tx, rx) = channel::(); + assert!(!tx.is_canceled()); + drop(rx); + assert!(tx.is_canceled()); +} + +#[test] +fn cancel_sends() { + let (tx, rx) = mpsc::channel::>(); + let t = thread::spawn(move || { + for otx in rx { + let _ = otx.send(42); + } + }); + + for _ in 0..20000 { + let (otx, mut orx) = channel::(); + tx.send(otx).unwrap(); + + orx.close(); + // Not necessary to wrap in a task because the implementation of oneshot + // never calls `task::current()` if the channel has been closed already. + let _ = orx.poll(); + } + + drop(tx); + t.join().unwrap(); +} diff --git a/tokio-channel/tests/support/mod.rs b/tokio-channel/tests/support/mod.rs new file mode 100644 index 000000000..2dccfc779 --- /dev/null +++ b/tokio-channel/tests/support/mod.rs @@ -0,0 +1,16 @@ +use futures::Future; + +pub trait ForgetExt { + fn forget(self); +} + +impl ForgetExt for F + where F: Future + Sized + Send + 'static, + F::Item: Send, + F::Error: Send +{ + fn forget(self) { + use std::thread; + thread::spawn(|| self.wait()); + } +} diff --git a/tokio-reactor/CHANGELOG.md b/tokio-reactor/CHANGELOG.md index e7bd54d8c..344a6928c 100644 --- a/tokio-reactor/CHANGELOG.md +++ b/tokio-reactor/CHANGELOG.md @@ -1,3 +1,7 @@ +# 0.1.5 (August 27, 2018) + +* Experimental async / await support. + # 0.1.4 (August 23, 2018) * Use a scalable RW lock (#517) diff --git a/tokio-reactor/Cargo.toml b/tokio-reactor/Cargo.toml index 85901c5d8..7861204e6 100644 --- a/tokio-reactor/Cargo.toml +++ b/tokio-reactor/Cargo.toml @@ -6,18 +6,23 @@ name = "tokio-reactor" # - Update CHANGELOG.md. # - Update doc URL. # - Create "v0.1.x" git tag. -version = "0.1.4" +version = "0.1.5" authors = ["Carl Lerche "] license = "MIT" readme = "README.md" repository = "https://github.com/tokio-rs/tokio" homepage = "https://tokio.rs" -documentation = "https://docs.rs/tokio-reactor/0.1.4/tokio_reactor" +documentation = "https://docs.rs/tokio-reactor/0.1.5/tokio_reactor" description = """ Event loop that drives Tokio I/O resources. """ categories = ["asynchronous", "network-programming"] +[features] +# This feature comes with no promise of stability. Things will break with each +# patch release. Use at your own risk. +async-await-preview = [] + [dependencies] crossbeam-utils = "0.5.0" futures = "0.1.19" diff --git a/tokio-reactor/src/async_await.rs b/tokio-reactor/src/async_await.rs new file mode 100644 index 000000000..698f285ce --- /dev/null +++ b/tokio-reactor/src/async_await.rs @@ -0,0 +1,5 @@ +use Registration; + +use std::marker::Unpin; + +impl Unpin for Registration {} diff --git a/tokio-reactor/src/lib.rs b/tokio-reactor/src/lib.rs index 3dff57829..eeefb7d02 100644 --- a/tokio-reactor/src/lib.rs +++ b/tokio-reactor/src/lib.rs @@ -1,5 +1,8 @@ -#![doc(html_root_url = "https://docs.rs/tokio-reactor/0.1.4")] +#![doc(html_root_url = "https://docs.rs/tokio-reactor/0.1.5")] #![deny(missing_docs, warnings, missing_debug_implementations)] +#![cfg_attr(feature = "async-await-preview", feature( + pin, + ))] //! Event loop that drives Tokio I/O resources. //! @@ -756,3 +759,8 @@ impl Error for SetFallbackError { "attempted to set fallback reactor while already configured" } } + +// ===== EXPERIMENTAL async / await support ===== + +#[cfg(feature = "async-await-preview")] +mod async_await;