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

add experimental async/await support. (#582)

Browse Source 
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.
This commit is contained in:
Carl Lerche
2018-08-27 12:24:51 -07:00
committed by GitHub
parent 6e45e0ac61
commit b479ce78d3
42 changed files with 3736 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
+1
View File
@@ -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
+45
View File
@@ -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
+52
View File
@@ -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
+57
View File
@@ -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
+133
View File
@@ -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
+53
View File
@@ -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
+45
View File
@@ -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);
}
});
}
tokio-async-await/examples/hyper.rs
+25
View File
@@ -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());
});
}
tokio-async-await/src/async_await/await.rs
+14
View File
@@ -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)
}}
}
tokio-async-await/src/async_await/compat/backward.rs
+112
View File
@@ -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<T>(PinBox<T>);
impl<T> Compat<T> {
pub fn new(data: T) -> Compat<T> {
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<T> IntoAwaitable for T
where T: Future03,
{
type Awaitable = Self;
fn into_awaitable(self) -> Self {
self
}
}
impl<T, Item, Error> Future01 for Compat<T>
where T: Future03<Output = Result<Item, Error>>,
{
type Item = Item;
type Error = Error;
fn poll(&mut self) -> Poll01<Item, Error> {
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<NoopWaker> = NonNull::dangling();
unsafe { LocalWaker::new(w) }
}
fn noop_waker() -> Waker {
let w: NonNull<NoopWaker> = 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,
})
}
}
tokio-async-await/src/async_await/compat/forward.rs
+69
View File
@@ -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>(T);
pub(crate) fn convert_poll<T, E>(poll: Result<Async<T>, E>) -> Poll03<Result<T, E>> {
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<T, E>(
poll: Result<Async<Option<T>>, E>) -> Poll03<Option<Result<T, E>>>
{
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<T: Future + Unpin> IntoAwaitable for T {
type Awaitable = Compat<T>;
fn into_awaitable(self) -> Self::Awaitable {
Compat(self)
}
}
impl<T> Future03 for Compat<T>
where T: Future + Unpin
{
type Output = Result<T::Item, T::Error>;
fn poll(self: PinMut<Self>, _cx: &mut Context) -> Poll03<Self::Output> {
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)),
}
}
}
tokio-async-await/src/async_await/compat/mod.rs
+4
View File
@@ -0,0 +1,4 @@
#![doc(hidden)]
pub mod forward;
pub mod backward;
tokio-async-await/src/async_await/io/flush.rs
+32
View File
@@ -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<Self>, _cx: &mut task::Context) -> Poll<Self::Output> {
use crate::async_await::compat::forward::convert_poll;
convert_poll(self.writer.poll_flush())
}
}
tokio-async-await/src/async_await/io/mod.rs
+192
View File
@@ -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<Self> {
Flush::new(self)
}
}
impl<T: AsyncRead + ?Sized> AsyncReadExt for T {}
impl<T: AsyncWrite + ?Sized> AsyncWriteExt for T {}
tokio-async-await/src/async_await/io/read.rs
+38
View File
@@ -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<usize>;
fn poll(mut self: PinMut<Self>, _cx: &mut task::Context) -> Poll<Self::Output> {
use crate::async_await::compat::forward::convert_poll;
let this = &mut *self;
convert_poll(this.reader.poll_read(this.buf))
}
}
tokio-async-await/src/async_await/io/read_exact.rs
+56
View File
@@ -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<Self>, _cx: &mut task::Context) -> Poll<Self::Output> {
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(()))
}
}
tokio-async-await/src/async_await/io/write.rs
+38
View File
@@ -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<usize>;
fn poll(mut self: PinMut<Self>, _cx: &mut task::Context) -> Poll<io::Result<usize>> {
use crate::async_await::compat::forward::convert_poll;
let this = &mut *self;
convert_poll(this.writer.poll_write(this.buf))
}
}
tokio-async-await/src/async_await/io/write_all.rs
+57
View File
@@ -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<Self>, _cx: &mut task::Context) -> Poll<io::Result<()>> {
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(()))
}
}
tokio-async-await/src/async_await/mod.rs
+8
View File
@@ -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;
tokio-async-await/src/async_await/sink/mod.rs
+26
View File
@@ -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<Self>
where
Self: Sized + Unpin,
{
Send::new(self, item)
}
}
impl<T: Sink> SinkExt for T {}
tokio-async-await/src/async_await/sink/send.rs
+59
View File
@@ -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<T::SinkItem>,
}
impl<T: Sink + Unpin + ?Sized> 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<T: Sink + Unpin + ?Sized> Future for Send<'_, T> {
type Output = Result<(), T::SinkError>;
fn poll(mut self: PinMut<Self>, _cx: &mut task::Context) -> Poll<Self::Output> {
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(()))
}
}
tokio-async-await/src/async_await/stream/mod.rs
+40
View File
@@ -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<Self>
where
Self: Sized + Unpin,
{
Next::new(self)
}
}
impl<T: Stream> StreamExt for T {}
tokio-async-await/src/async_await/stream/next.rs
+31
View File
@@ -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<Result<T::Item, T::Error>>;
fn poll(self: PinMut<Self>, _cx: &mut task::Context) -> Poll<Self::Output> {
use crate::async_await::compat::forward::convert_poll_stream;
convert_poll_stream(
PinMut::get_mut(self).stream.poll())
}
}
tokio-async-await/src/lib.rs
+102
View File
@@ -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<F>(future: F)
where F: Future03<Output = ()> + 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<F>(future: F)
where F: Future03<Output = ()> + Send + 'static,
{
use futures_util::future::FutureExt;
use crate::async_await::compat::backward;
let future = future.map(|_| Ok(()));
spawn(backward::Compat::new(future));
}
tokio-channel/CHANGELOG.md
View File
tokio-channel/Cargo.toml
+25
View File
@@ -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 <[email protected]>"]
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"
tokio-channel/LICENSE
+51
View File
@@ -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.
tokio-channel/README.md
View File
tokio-channel/src/async_await.rs
+10
View File
@@ -0,0 +1,10 @@
use {oneshot, mpsc};
use std::marker::Unpin;
impl<T> Unpin for oneshot::Sender<T> {}
impl<T> Unpin for oneshot::Receiver<T> {}
impl<T> Unpin for mpsc::Sender<T> {}
impl<T> Unpin for mpsc::UnboundedSender<T> {}
impl<T> Unpin for mpsc::Receiver<T> {}
tokio-channel/src/lib.rs
+22
View File
@@ -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;
tokio-channel/src/lock.rs
+105
View File
@@ -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<T>`.
///
/// This lock only supports the `try_lock` operation, however, and does not
/// implement poisoning.
#[derive(Debug)]
pub struct Lock<T> {
locked: AtomicBool,
data: UnsafeCell<T>,
}
/// Sentinel representing an acquired lock through which the data can be
/// accessed.
pub struct TryLock<'a, T: 'a> {
__ptr: &'a Lock<T>,
}
// The `Lock` structure is basically just a `Mutex<T>`, and these two impls are
// intended to mirror the standard library's corresponding impls for `Mutex<T>`.
//
// 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<T: Send> Send for Lock<T> {}
unsafe impl<T: Send> Sync for Lock<T> {}
impl<T> Lock<T> {
/// Creates a new lock around the given value.
pub fn new(t: T) -> Lock<T> {
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<TryLock<T>> {
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);
}
}
tokio-channel/src/mpsc/mod.rs
+989
View File
@@ -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<T> {
// Channel state shared between the sender and receiver.
inner: Arc<Inner<T>>,
// 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<Mutex<SenderTask>>,
// 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<T>(Sender<T>);
trait AssertKinds: Send + Sync + Clone {}
impl AssertKinds for UnboundedSender<u32> {}
/// 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<T> {
inner: Arc<Inner<T>>,
}
/// Error type for sending, used when the receiving end of a channel is
/// dropped
#[derive(Clone, PartialEq, Eq)]
pub struct SendError<T>(T);
/// Error type returned from `try_send`
#[derive(Clone, PartialEq, Eq)]
pub struct TrySendError<T> {
kind: TrySendErrorKind<T>,
}
#[derive(Clone, PartialEq, Eq)]
enum TrySendErrorKind<T> {
Full(T),
Disconnected(T),
}
impl<T> fmt::Debug for SendError<T> {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
fmt.debug_tuple("SendError")
.field(&"...")
.finish()
}
}
impl<T> fmt::Display for SendError<T> {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
write!(fmt, "send failed because receiver is gone")
}
}
impl<T: Any> Error for SendError<T>
{
fn description(&self) -> &str {
"send failed because receiver is gone"
}
}
impl<T> SendError<T> {
/// Returns the message that was attempted to be sent but failed.
pub fn into_inner(self) -> T {
self.0
}
}
impl<T> fmt::Debug for TrySendError<T> {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
fmt.debug_tuple("TrySendError")
.field(&"...")
.finish()
}
}
impl<T> fmt::Display for TrySendError<T> {
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<T: Any> Error for TrySendError<T> {
fn description(&self) -> &str {
if self.is_full() {
"send failed because channel is full"
} else {
"send failed because receiver is gone"
}
}
}
impl<T> TrySendError<T> {
/// 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<T> {
// Max buffer size of the channel. If `None` then the channel is unbounded.
buffer: Option<usize>,
// 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<Option<T>>,
// Atomic, FIFO queue used to send parked task handles to the receiver.
parked_queue: Queue<Arc<Mutex<SenderTask>>>,
// Number of senders in existence
num_senders: AtomicUsize,
// Handle to the receiver's task.
recv_task: Mutex<ReceiverTask>,
}
// 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<Task>,
}
// 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<Task>,
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<T>(buffer: usize) -> (Sender<T>, Receiver<T>) {
// 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<T>() -> (UnboundedSender<T>, Receiver<T>) {
let (tx, rx) = channel2(None);
(UnboundedSender(tx), rx)
}
fn channel2<T>(buffer: Option<usize>) -> (Sender<T>, Receiver<T>) {
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<T> Sender<T> {
/// Attempts to send a message on this `Sender<T>` 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<T>> {
// 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<T>, do_park: bool) -> Result<(), SendError<T>> {
// 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<T>> {
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<T>) {
// 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<bool> {
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<T> Sink for Sender<T> {
type SinkItem = T;
type SinkError = SendError<T>;
fn start_send(&mut self, msg: T) -> StartSend<T, SendError<T>> {
// 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<T>> {
Ok(Async::Ready(()))
}
fn close(&mut self) -> Poll<(), SendError<T>> {
Ok(Async::Ready(()))
}
}
impl<T> UnboundedSender<T> {
/// 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<T>> {
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<T>> {
self.0.do_send_nb(msg)
}
}
impl<T> Sink for UnboundedSender<T> {
type SinkItem = T;
type SinkError = SendError<T>;
fn start_send(&mut self, msg: T) -> StartSend<T, SendError<T>> {
self.0.start_send(msg)
}
fn poll_complete(&mut self) -> Poll<(), SendError<T>> {
self.0.poll_complete()
}
fn close(&mut self) -> Poll<(), SendError<T>> {
Ok(Async::Ready(()))
}
}
impl<'a, T> Sink for &'a UnboundedSender<T> {
type SinkItem = T;
type SinkError = SendError<T>;
fn start_send(&mut self, msg: T) -> StartSend<T, SendError<T>> {
self.0.do_send_nb(msg)?;
Ok(AsyncSink::Ready)
}
fn poll_complete(&mut self) -> Poll<(), SendError<T>> {
Ok(Async::Ready(()))
}
fn close(&mut self) -> Poll<(), SendError<T>> {
Ok(Async::Ready(()))
}
}
impl<T> Clone for UnboundedSender<T> {
fn clone(&self) -> UnboundedSender<T> {
UnboundedSender(self.0.clone())
}
}
impl<T> Clone for Sender<T> {
fn clone(&self) -> Sender<T> {
// 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<T> Drop for Sender<T> {
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<T> Receiver<T> {
/// 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<Option<T>> {
// 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<T> Stream for Receiver<T> {
type Item = T;
type Error = ();
fn poll(&mut self) -> Poll<Option<T>, ()> {
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<T> Drop for Receiver<T> {
fn drop(&mut self) {
// Drain the channel of all pending messages
self.close();
while self.next_message().is_ready() {
// ...
}
}
}
/*
*
* ===== impl Inner =====
*
*/
impl<T> Inner<T> {
// 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<T: Send> Send for Inner<T> {}
unsafe impl<T: Send> Sync for Inner<T> {}
/*
*
* ===== 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
}
tokio-channel/src/mpsc/queue.rs
+151
View File
@@ -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<T> {
/// 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<T> {
next: AtomicPtr<Node<T>>,
value: Option<T>,
}
/// 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<T> {
head: AtomicPtr<Node<T>>,
tail: UnsafeCell<*mut Node<T>>,
}
unsafe impl<T: Send> Send for Queue<T> { }
unsafe impl<T: Send> Sync for Queue<T> { }
impl<T> Node<T> {
unsafe fn new(v: Option<T>) -> *mut Node<T> {
Box::into_raw(Box::new(Node {
next: AtomicPtr::new(ptr::null_mut()),
value: v,
}))
}
}
impl<T> Queue<T> {
/// Creates a new queue that is safe to share among multiple producers and
/// one consumer.
pub fn new() -> Queue<T> {
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<T>`. 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<T> {
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<T> Drop for Queue<T> {
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;
}
}
}
}
tokio-channel/src/oneshot.rs
+426
View File
@@ -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<T> {
inner: Arc<Inner<T>>,
}
/// 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<T> {
inner: Arc<Inner<T>>,
}
/// 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<T> {
/// 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<Option<T>>,
/// 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<Option<Task>>,
/// Like `rx_task` above, except for the task blocked in
/// `Sender::poll_cancel`. Additionally, `Lock` cannot be `UnsafeCell`.
tx_task: Lock<Option<Task>>,
}
/// 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::<i32>();
///
/// thread::spawn(|| {
/// c.map(|i| {
/// println!("got: {}", i);
/// }).wait();
/// });
///
/// p.send(3).unwrap();
/// # }
/// ```
pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
let inner = Arc::new(Inner::new());
let receiver = Receiver {
inner: inner.clone(),
};
let sender = Sender {
inner: inner,
};
(sender, receiver)
}
impl<T> Inner<T> {
fn new() -> Inner<T> {
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<T, Canceled> {
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<T> Sender<T> {
#[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<T> Drop for Sender<T> {
fn drop(&mut self) {
self.inner.drop_tx()
}
}
/// Error returned from a `Receiver<T>` whenever the corresponding `Sender<T>`
/// 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<T> Receiver<T> {
/// 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<T> Future for Receiver<T> {
type Item = T;
type Error = Canceled;
fn poll(&mut self) -> Poll<T, Canceled> {
self.inner.recv()
}
}
impl<T> Drop for Receiver<T> {
fn drop(&mut self) {
self.inner.drop_rx()
}
}
tokio-channel/tests/mpsc-close.rs
+22
View File
@@ -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()
}
tokio-channel/tests/mpsc.rs
+481
View File
@@ -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<i32> {}
impl AssertSend for mpsc::Receiver<i32> {}
#[test]
fn send_recv() {
let (tx, rx) = mpsc::channel::<i32>(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::<i32>(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::<i32>(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::<i32>(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::<i32>(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::<i32>(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::<i32>(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::<i32>();
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::<i32>(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::<usize>(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::<bool, ()>(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<Stream<Item=i32, Error=u32>> {
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::<Result<Vec<_>, _>>(),
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<u32>,
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<T>(res: &AsyncSink<T>) -> 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());
}
tokio-channel/tests/oneshot.rs
+124
View File
@@ -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::<u32>();
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::<u32>();
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<u32>,
}
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::<u32>();
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::<u32>();
rx.close();
assert!(rx.poll().is_err());
assert!(tx.poll_cancel().unwrap().is_ready());
}
#[test]
fn close_wakes() {
let (tx, mut rx) = channel::<u32>();
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::<u32>();
assert!(!tx.is_canceled());
drop(rx);
assert!(tx.is_canceled());
}
#[test]
fn cancel_sends() {
let (tx, rx) = mpsc::channel::<Sender<_>>();
let t = thread::spawn(move || {
for otx in rx {
let _ = otx.send(42);
}
});
for _ in 0..20000 {
let (otx, mut orx) = channel::<u32>();
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();
}
tokio-channel/tests/support/mod.rs
+16
View File
@@ -0,0 +1,16 @@
use futures::Future;
pub trait ForgetExt {
fn forget(self);
}
impl<F> 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());
}
}
tokio-reactor/CHANGELOG.md
+4
View File
@@ -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)
tokio-reactor/Cargo.toml
+7 -2
View File
@@ -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 <[email protected]>"]
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"
tokio-reactor/src/async_await.rs
+5
View File
@@ -0,0 +1,5 @@
use Registration;
use std::marker::Unpin;
impl Unpin for Registration {}
tokio-reactor/src/lib.rs
+9 -1
View File
@@ -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;