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.

Author: carllerche

Cargo.toml

@@ -24,6 +24,7 @@ keywords = ["io", "async", "non-blocking", "futures"]
 
 members = [
   "./",
+  "tokio-channel",
   "tokio-codec",
   "tokio-current-thread",
   "tokio-executor",

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 <[email protected]>"]
+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" }

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.
+

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.

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<String>;
+
+struct Shared {
+    peers: HashMap<SocketAddr, Tx>,
+}
+
+impl Shared {
+    /// Create a new, empty, instance of `Shared`.
+    fn new() -> Self {
+        Shared {
+            peers: HashMap::new(),
+        }
+    }
+}
+
+async fn process(stream: TcpStream, state: Arc<Mutex<Shared>>) -> 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");
+                }
+            });
+        }
+    });
+}
+

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::<SocketAddr>().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),
+        }
+    });
+}

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::<SocketAddr>().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);
+        }
+    });
+}