From d304791c0e3a34cce91f6bcc7178d10c1b3ec9e8 Mon Sep 17 00:00:00 2001 From: Aaron Turon Date: Tue, 13 Mar 2018 13:57:35 -0700 Subject: [PATCH] Simultaneous futures compat (#172) This patch adds opt-in support for futures 0.2. --- .travis.yml | 3 + Cargo.toml | 11 ++ futures2/Cargo.toml | 11 ++ futures2/src/lib.rs | 2 + src/executor/current_thread/mod.rs | 17 +++ src/executor/mod.rs | 1 - src/lib.rs | 19 +++ src/net/tcp/incoming.rs | 15 ++ src/net/tcp/listener.rs | 38 +++++ src/net/tcp/stream.rs | 111 +++++++++++++- src/runtime.rs | 50 +++++++ tests/current_thread.rs | 2 + tokio-executor/Cargo.toml | 5 + tokio-executor/src/enter.rs | 9 ++ tokio-executor/src/global.rs | 29 ++++ tokio-executor/src/lib.rs | 13 +- tokio-io/Cargo.toml | 5 + tokio-reactor/Cargo.toml | 5 + tokio-reactor/src/atomic_task.rs | 11 +- tokio-reactor/src/background.rs | 7 +- tokio-reactor/src/lib.rs | 41 ++++- tokio-reactor/src/poll_evented.rs | 215 +++++++++++++++++++++++++-- tokio-reactor/src/registration.rs | 56 +++++-- tokio-threadpool/Cargo.toml | 5 + tokio-threadpool/src/lib.rs | 182 +++++++++++++++++++++-- tokio-threadpool/src/task.rs | 77 ++++++++-- tokio-threadpool/tests/threadpool.rs | 210 ++++++++++++++++++++------ 27 files changed, 1045 insertions(+), 105 deletions(-) create mode 100644 futures2/Cargo.toml create mode 100644 futures2/src/lib.rs mode change 100644 => 100755 tests/current_thread.rs diff --git a/.travis.yml b/.travis.yml index 00d28a7ca..225fbb5fc 100644 --- a/.travis.yml +++ b/.travis.yml @@ -26,6 +26,9 @@ script: cargo check --tests --all --target $TARGET else cargo test --all + cargo test --features unstable-futures + cargo test --manifest-path tokio-threadpool/Cargo.toml --features unstable-futures + cargo test --manifest-path tokio-reactor/Cargo.toml --features unstable-futures fi deploy: diff --git a/Cargo.toml b/Cargo.toml index 6ff608160..7af8cc430 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,6 +27,7 @@ members = [ "tokio-io", "tokio-reactor", "tokio-threadpool", + "futures2", ] [badges] @@ -44,6 +45,7 @@ mio = "0.6.14" slab = "0.4" iovec = "0.1" futures = "0.1.18" +futures2 = { version = "0.1", path = "futures2", optional = true } [dev-dependencies] env_logger = { version = "0.4", default-features = false } @@ -60,3 +62,12 @@ time = "0.1" [patch.crates-io] tokio-io = { path = "tokio-io" } + +[features] +unstable-futures = [ + "futures2", + "tokio-reactor/unstable-futures", + "tokio-threadpool/unstable-futures", + "tokio-executor/unstable-futures" +] +default = [] diff --git a/futures2/Cargo.toml b/futures2/Cargo.toml new file mode 100644 index 000000000..a78a42d8c --- /dev/null +++ b/futures2/Cargo.toml @@ -0,0 +1,11 @@ +[package] +name = "futures2" + +version = "0.1.0" +authors = ["Aaron Turon "] +license = "MIT/Apache-2.0" +repository = "https://github.com/tokio-rs/tokio" +homepage = "https://tokio.rs" + +[dependencies] +futures = "0.2.0-alpha" diff --git a/futures2/src/lib.rs b/futures2/src/lib.rs new file mode 100644 index 000000000..af0c9dc41 --- /dev/null +++ b/futures2/src/lib.rs @@ -0,0 +1,2 @@ +extern crate futures; +pub use futures::*; diff --git a/src/executor/current_thread/mod.rs b/src/executor/current_thread/mod.rs index a5035d64c..d4d2ed295 100644 --- a/src/executor/current_thread/mod.rs +++ b/src/executor/current_thread/mod.rs @@ -119,6 +119,9 @@ use std::marker::PhantomData; use std::rc::Rc; use std::time::{Duration, Instant}; +#[cfg(feature = "unstable-futures")] +use futures2; + /// Executes tasks on the current thread pub struct CurrentThread { /// Execute futures and receive unpark notifications. @@ -386,6 +389,13 @@ impl tokio_executor::Executor for CurrentThread { self.borrow().spawn_local(future); Ok(()) } + + #[cfg(feature = "unstable-futures")] + fn spawn2(&mut self, _future: Box + Send>) + -> Result<(), futures2::executor::SpawnError> + { + panic!("Futures 0.2 integration is not available for current_thread"); + } } impl fmt::Debug for CurrentThread

{ @@ -591,6 +601,13 @@ impl tokio_executor::Executor for TaskExecutor { self.spawn_local(future) } + #[cfg(feature = "unstable-futures")] + fn spawn2(&mut self, _future: Box + Send>) + -> Result<(), futures2::executor::SpawnError> + { + panic!("Futures 0.2 integration is not available for current_thread"); + } + fn status(&self) -> Result<(), SpawnError> { CURRENT.with(|current| { if current.spawn.get().is_some() { diff --git a/src/executor/mod.rs b/src/executor/mod.rs index 465439bb6..c69558032 100644 --- a/src/executor/mod.rs +++ b/src/executor/mod.rs @@ -49,7 +49,6 @@ //! [`Executor`]: # //! [`spawn`]: # - pub mod current_thread; pub mod thread_pool { diff --git a/src/lib.rs b/src/lib.rs index 963e71c5b..6bbf24168 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -79,6 +79,9 @@ extern crate tokio_threadpool; #[macro_use] extern crate log; +#[cfg(feature = "unstable-futures")] +extern crate futures2; + pub mod executor; pub mod net; pub mod reactor; @@ -187,3 +190,19 @@ pub mod prelude { task, }; } + +#[cfg(feature = "unstable-futures")] +fn lift_async(old: futures::Async) -> futures2::Async { + match old { + futures::Async::Ready(x) => futures2::Async::Ready(x), + futures::Async::NotReady => futures2::Async::Pending, + } +} + +#[cfg(feature = "unstable-futures")] +fn lower_async(new: futures2::Async) -> futures::Async { + match new { + futures2::Async::Ready(x) => futures::Async::Ready(x), + futures2::Async::Pending => futures::Async::NotReady, + } +} diff --git a/src/net/tcp/incoming.rs b/src/net/tcp/incoming.rs index 0e5e5bb81..591acc204 100644 --- a/src/net/tcp/incoming.rs +++ b/src/net/tcp/incoming.rs @@ -5,6 +5,9 @@ use std::io; use futures::stream::Stream; use futures::{Poll, Async}; +#[cfg(feature = "unstable-futures")] +use futures2; + /// Stream returned by the `TcpListener::incoming` function representing the /// stream of sockets received from a listener. #[must_use = "streams do nothing unless polled"] @@ -28,3 +31,15 @@ impl Stream for Incoming { Ok(Async::Ready(Some(socket))) } } + +#[cfg(feature = "unstable-futures")] +impl futures2::Stream for Incoming { + type Item = TcpStream; + type Error = io::Error; + + fn poll_next(&mut self, cx: &mut futures2::task::Context) + -> futures2::Poll, io::Error> + { + Ok(self.inner.poll_accept2(cx)?.map(|(sock, _)| Some(sock))) + } +} diff --git a/src/net/tcp/listener.rs b/src/net/tcp/listener.rs index de9e38ea9..bc9a736a6 100644 --- a/src/net/tcp/listener.rs +++ b/src/net/tcp/listener.rs @@ -10,6 +10,9 @@ use mio; use reactor::{Handle, PollEvented2}; +#[cfg(feature = "unstable-futures")] +use futures2; + /// An I/O object representing a TCP socket listening for incoming connections. /// /// This object can be converted into a stream of incoming connections for @@ -64,6 +67,22 @@ impl TcpListener { Ok((io, addr).into()) } + /// Like `poll_accept`, but for futures 0.2 + #[cfg(feature = "unstable-futures")] + pub fn poll_accept2(&mut self, cx: &mut futures2::task::Context) + -> futures2::Poll<(TcpStream, SocketAddr), io::Error> + { + let (io, addr) = match self.poll_accept_std2(cx)? { + futures2::Async::Ready(x) => x, + futures2::Async::Pending => return Ok(futures2::Async::Pending), + }; + + let io = mio::net::TcpStream::from_stream(io)?; + let io = TcpStream::new(io); + + Ok((io, addr).into()) + } + #[deprecated(since = "0.1.2", note = "use poll_accept_std instead")] #[doc(hidden)] pub fn accept_std(&mut self) -> io::Result<(net::TcpStream, SocketAddr)> { @@ -105,6 +124,25 @@ impl TcpListener { } } + /// Like `poll_accept_std`, but for futures 0.2. + #[cfg(feature = "unstable-futures")] + pub fn poll_accept_std2(&mut self, cx: &mut futures2::task::Context) + -> futures2::Poll<(net::TcpStream, SocketAddr), io::Error> + { + if let futures2::Async::Pending = self.io.poll_read_ready2(cx, mio::Ready::readable())? { + return Ok(futures2::Async::Pending); + } + + match self.io.get_ref().accept_std() { + Ok(pair) => Ok(pair.into()), + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { + self.io.clear_read_ready2(cx, mio::Ready::readable())?; + Ok(futures2::Async::Pending) + } + Err(e) => Err(e), + } + } + /// Create a new TCP listener from the standard library's TCP listener. /// /// This method can be used when the `Handle::tcp_listen` method isn't diff --git a/src/net/tcp/stream.rs b/src/net/tcp/stream.rs index d37a1a435..602bd4bb2 100644 --- a/src/net/tcp/stream.rs +++ b/src/net/tcp/stream.rs @@ -12,6 +12,9 @@ use tokio_io::{AsyncRead, AsyncWrite}; use reactor::{Handle, PollEvented2}; +#[cfg(feature = "unstable-futures")] +use futures2; + /// An I/O object representing a TCP stream connected to a remote endpoint. /// /// A TCP stream can either be created by connecting to an endpoint, via the @@ -208,6 +211,25 @@ impl TcpStream { } } + /// Like `poll_peek` but compatible with futures 0.2 + #[cfg(feature = "unstable-futures")] + pub fn poll_peek2(&mut self, cx: &mut futures2::task::Context, buf: &mut [u8]) + -> futures2::Poll + { + if let futures2::Async::Pending = self.io.poll_read_ready2(cx, mio::Ready::readable())? { + return Ok(futures2::Async::Pending); + } + + match self.io.get_ref().peek(buf) { + Ok(ret) => Ok(ret.into()), + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { + self.io.clear_read_ready2(cx, mio::Ready::readable())?; + Ok(futures2::Async::Pending) + } + Err(e) => Err(e), + } + } + /// Shuts down the read, write, or both halves of this connection. /// /// This function will cause all pending and future I/O on the specified @@ -367,6 +389,15 @@ impl AsyncRead for TcpStream { } } +#[cfg(feature = "unstable-futures")] +impl futures2::io::AsyncRead for TcpStream { + fn poll_read(&mut self, cx: &mut futures2::task::Context, buf: &mut [u8]) + -> futures2::Poll + { + futures2::io::AsyncRead::poll_read(&mut self.io, cx, buf) + } +} + impl AsyncWrite for TcpStream { fn shutdown(&mut self) -> Poll<(), io::Error> { <&TcpStream>::shutdown(&mut &*self) @@ -377,6 +408,23 @@ impl AsyncWrite for TcpStream { } } +#[cfg(feature = "unstable-futures")] +impl futures2::io::AsyncWrite for TcpStream { + fn poll_write(&mut self, cx: &mut futures2::task::Context, buf: &[u8]) + -> futures2::Poll + { + futures2::io::AsyncWrite::poll_write(&mut self.io, cx, buf) + } + + fn poll_flush(&mut self, cx: &mut futures2::task::Context) -> futures2::Poll<(), io::Error> { + futures2::io::AsyncWrite::poll_flush(&mut self.io, cx) + } + + fn poll_close(&mut self, cx: &mut futures2::task::Context) -> futures2::Poll<(), io::Error> { + futures2::io::AsyncWrite::poll_close(&mut self.io, cx) + } +} + // ===== impl Read / Write for &'a ===== impl<'a> Read for &'a TcpStream { @@ -449,6 +497,15 @@ impl<'a> AsyncRead for &'a TcpStream { } } +#[cfg(feature = "unstable-futures")] +impl<'a> futures2::io::AsyncRead for &'a TcpStream { + fn poll_read(&mut self, cx: &mut futures2::task::Context, buf: &mut [u8]) + -> futures2::Poll + { + futures2::io::AsyncRead::poll_read(&mut &self.io, cx, buf) + } +} + impl<'a> AsyncWrite for &'a TcpStream { fn shutdown(&mut self) -> Poll<(), io::Error> { Ok(().into()) @@ -483,13 +540,29 @@ impl<'a> AsyncWrite for &'a TcpStream { } } +#[cfg(feature = "unstable-futures")] +impl<'a> futures2::io::AsyncWrite for &'a TcpStream { + fn poll_write(&mut self, cx: &mut futures2::task::Context, buf: &[u8]) + -> futures2::Poll + { + futures2::io::AsyncWrite::poll_write(&mut &self.io, cx, buf) + } + + fn poll_flush(&mut self, cx: &mut futures2::task::Context) -> futures2::Poll<(), io::Error> { + futures2::io::AsyncWrite::poll_flush(&mut &self.io, cx) + } + + fn poll_close(&mut self, cx: &mut futures2::task::Context) -> futures2::Poll<(), io::Error> { + futures2::io::AsyncWrite::poll_close(&mut &self.io, cx) + } +} + impl fmt::Debug for TcpStream { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { self.io.get_ref().fmt(f) } } - impl Future for ConnectFuture { type Item = TcpStream; type Error = io::Error; @@ -499,11 +572,20 @@ impl Future for ConnectFuture { } } -impl Future for ConnectFutureState { +#[cfg(feature = "unstable-futures")] +impl futures2::Future for ConnectFuture { type Item = TcpStream; type Error = io::Error; - fn poll(&mut self) -> Poll { + fn poll(&mut self, cx: &mut futures2::task::Context) -> futures2::Poll { + futures2::Future::poll(&mut self.inner, cx) + } +} + +impl ConnectFutureState { + fn poll_inner(&mut self, f: F) -> Poll + where F: FnOnce(&mut PollEvented2) -> Poll + { { let stream = match *self { ConnectFutureState::Waiting(ref mut s) => s, @@ -523,7 +605,7 @@ impl Future for ConnectFutureState { // actually hit an error or not. // // If all that succeeded then we ship everything on up. - if let Async::NotReady = stream.io.poll_write_ready()? { + if let Async::NotReady = f(&mut stream.io)? { return Ok(Async::NotReady) } @@ -531,6 +613,7 @@ impl Future for ConnectFutureState { return Err(e) } } + match mem::replace(self, ConnectFutureState::Empty) { ConnectFutureState::Waiting(stream) => Ok(Async::Ready(stream)), _ => panic!(), @@ -538,6 +621,26 @@ impl Future for ConnectFutureState { } } +impl Future for ConnectFutureState { + type Item = TcpStream; + type Error = io::Error; + + fn poll(&mut self) -> Poll { + self.poll_inner(|io| io.poll_write_ready()) + } +} + +#[cfg(feature = "unstable-futures")] +impl futures2::Future for ConnectFutureState { + type Item = TcpStream; + type Error = io::Error; + + fn poll(&mut self, cx: &mut futures2::task::Context) -> futures2::Poll { + self.poll_inner(|io| io.poll_write_ready2(cx).map(::lower_async)) + .map(::lift_async) + } +} + #[cfg(all(unix, not(target_os = "fuchsia")))] mod sys { use std::os::unix::prelude::*; diff --git a/src/runtime.rs b/src/runtime.rs index 3fb3c9aa0..c277bb788 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -112,6 +112,9 @@ use futures::future::{self, Future}; use std::{fmt, io}; +#[cfg(feature = "unstable-futures")] +use futures2; + /// Handle to the Tokio runtime. /// /// The Tokio runtime includes a reactor as well as an executor for running @@ -205,6 +208,18 @@ where F: Future + Send + 'static, runtime.shutdown_on_idle().wait().unwrap(); } +/// Start the Tokio runtime using the supplied future to bootstrap execution. +/// +/// Identical to `run` but works with futures 0.2-style futures. +#[cfg(feature = "unstable-futures")] +pub fn run2(future: F) + where F: futures2::Future + Send + 'static, +{ + let mut runtime = Runtime::new().unwrap(); + runtime.spawn2(future); + runtime.shutdown_on_idle().wait().unwrap(); +} + impl Runtime { /// Create a new runtime instance with default configuration values. /// @@ -287,6 +302,19 @@ impl Runtime { self } + /// Spawn a futures 0.2-style future onto the Tokio runtime. + /// + /// Otherwise identical to `spawn` + #[cfg(feature = "unstable-futures")] + pub fn spawn2(&mut self, future: F) -> &mut Self + where F: futures2::Future + Send + 'static, + { + futures2::executor::Executor::spawn( + self.inner_mut().pool.sender_mut(), Box::new(future) + ).unwrap(); + self + } + /// Signals the runtime to shutdown once it becomes idle. /// /// Returns a future that completes once the shutdown operation has @@ -420,8 +448,30 @@ impl ::executor::Executor for TaskExecutor { { self.inner.spawn(future) } + + #[cfg(feature = "unstable-futures")] + fn spawn2(&mut self, future: Box + Send>) + -> Result<(), futures2::executor::SpawnError> + { + self.inner.spawn2(future) + } } +#[cfg(feature = "unstable-futures")] +type Task2 = Box + Send>; + +#[cfg(feature = "unstable-futures")] +impl futures2::executor::Executor for TaskExecutor { + fn spawn(&mut self, f: Task2) -> Result<(), futures2::executor::SpawnError> { + futures2::executor::Executor::spawn(&mut self.inner, f) + } + + fn status(&self) -> Result<(), futures2::executor::SpawnError> { + futures2::executor::Executor::status(&self.inner) + } +} + + // ===== impl Shutdown ===== impl Shutdown { diff --git a/tests/current_thread.rs b/tests/current_thread.rs old mode 100644 new mode 100755 index dfc803345..1124f727a --- a/tests/current_thread.rs +++ b/tests/current_thread.rs @@ -1,3 +1,5 @@ +#![cfg(not(feature = "unstable-futures"))] + extern crate tokio; extern crate tokio_executor; extern crate futures; diff --git a/tokio-executor/Cargo.toml b/tokio-executor/Cargo.toml index 8d65dcb22..a3ba1b3bd 100644 --- a/tokio-executor/Cargo.toml +++ b/tokio-executor/Cargo.toml @@ -14,3 +14,8 @@ categories = ["concurrency", "asynchronous"] [dependencies] futures = "0.1.18" +futures2 = { version = "0.1", path = "../futures2", optional = true } + +[features] +unstable-futures = ["futures2"] +default = [] diff --git a/tokio-executor/src/enter.rs b/tokio-executor/src/enter.rs index 3c9f3cfa1..39a445ad3 100644 --- a/tokio-executor/src/enter.rs +++ b/tokio-executor/src/enter.rs @@ -2,6 +2,9 @@ use std::prelude::v1::*; use std::cell::Cell; use std::fmt; +#[cfg(feature = "unstable-futures")] +use futures2; + thread_local!(static ENTERED: Cell = Cell::new(false)); /// Represents an executor context. @@ -10,6 +13,9 @@ thread_local!(static ENTERED: Cell = Cell::new(false)); pub struct Enter { on_exit: Vec>, permanent: bool, + + #[cfg(feature = "unstable-futures")] + _enter2: futures2::executor::Enter, } /// An error returned by `enter` if an execution scope has already been @@ -40,6 +46,9 @@ pub fn enter() -> Result { Ok(Enter { on_exit: Vec::new(), permanent: false, + + #[cfg(feature = "unstable-futures")] + _enter2: futures2::executor::enter().unwrap(), }) } }) diff --git a/tokio-executor/src/global.rs b/tokio-executor/src/global.rs index 239af6eba..ab4d526dc 100644 --- a/tokio-executor/src/global.rs +++ b/tokio-executor/src/global.rs @@ -6,6 +6,9 @@ use std::cell::Cell; use std::marker::PhantomData; use std::rc::Rc; +#[cfg(feature = "unstable-futures")] +use futures2; + /// Executes futures on the default executor for the current execution context. /// /// `DefaultExecutor` implements `Executor` and can be used to spawn futures @@ -59,6 +62,23 @@ impl super::Executor for DefaultExecutor { } }) } + + #[cfg(feature = "unstable-futures")] + fn spawn2(&mut self, future: Box + Send>) + -> Result<(), futures2::executor::SpawnError> + { + EXECUTOR.with(|current_executor| { + match current_executor.get() { + Some(executor) => { + let executor = unsafe { &mut *executor }; + executor.spawn2(future) + } + None => { + Err(futures2::executor::SpawnError::shutdown()) + } + } + }) + } } // ===== global spawn fns ===== @@ -109,6 +129,15 @@ pub fn spawn(future: T) .unwrap() } +/// Like `spawn` but compatible with futures 0.2 +#[cfg(feature = "unstable-futures")] +pub fn spawn2(future: T) + where T: futures2::Future + Send + 'static, +{ + DefaultExecutor::current().spawn2(Box::new(future)) + .unwrap() +} + /// Set the default executor for the duration of the closure /// /// # Panics diff --git a/tokio-executor/src/lib.rs b/tokio-executor/src/lib.rs index 7aab4d868..690c1e489 100644 --- a/tokio-executor/src/lib.rs +++ b/tokio-executor/src/lib.rs @@ -35,6 +35,9 @@ extern crate futures; +#[cfg(feature = "unstable-futures")] +extern crate futures2; + mod enter; mod global; pub mod park; @@ -42,6 +45,9 @@ pub mod park; pub use enter::{enter, Enter, EnterError}; pub use global::{spawn, with_default, DefaultExecutor}; +#[cfg(feature = "unstable-futures")] +pub use global::spawn2; + use futures::Future; /// A value that executes futures. @@ -129,7 +135,12 @@ pub trait Executor { /// # fn main() {} /// ``` fn spawn(&mut self, future: Box + Send>) - -> Result<(), SpawnError>; + -> Result<(), SpawnError>; + + /// Like `spawn`, but compatible with futures 0.2 + #[cfg(feature = "unstable-futures")] + fn spawn2(&mut self, future: Box + Send>) + -> Result<(), futures2::executor::SpawnError>; /// Provides a best effort **hint** to whether or not `spawn` will succeed. /// diff --git a/tokio-io/Cargo.toml b/tokio-io/Cargo.toml index 95a9db3ed..bf8232574 100644 --- a/tokio-io/Cargo.toml +++ b/tokio-io/Cargo.toml @@ -20,3 +20,8 @@ categories = ["asynchronous"] bytes = "0.4" futures = "0.1.18" log = "0.4" +futures2 = { version = "0.1", path = "../futures2", optional = true } + +[features] +unstable-futures = ["futures2"] +default = [] diff --git a/tokio-reactor/Cargo.toml b/tokio-reactor/Cargo.toml index 453b774e7..0e2ab916e 100644 --- a/tokio-reactor/Cargo.toml +++ b/tokio-reactor/Cargo.toml @@ -24,3 +24,8 @@ mio = "0.6.14" slab = "0.4.0" tokio-executor = { version = "0.1.0", path = "../tokio-executor" } tokio-io = { version = "0.1.6", path = "../tokio-io" } +futures2 = { version = "0.1", path = "../futures2", optional = true } + +[features] +unstable-futures = ["futures2"] +default = [] diff --git a/tokio-reactor/src/atomic_task.rs b/tokio-reactor/src/atomic_task.rs index 9b4ba00ca..6a4788e6c 100644 --- a/tokio-reactor/src/atomic_task.rs +++ b/tokio-reactor/src/atomic_task.rs @@ -1,10 +1,10 @@ -use futures::task::{self, Task}; - use std::fmt; use std::cell::UnsafeCell; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering::{Acquire, Release}; +use Task; + /// A synchronization primitive for task notification. /// /// `AtomicTask` will coordinate concurrent notifications with the consumer @@ -69,11 +69,6 @@ impl AtomicTask { } } - /// Registers the **current** task to be notified on calls to `notify`. - pub fn register(&self) { - self.register_task(task::current()); - } - /// Registers the task to be notified on calls to `notify`. /// /// The new task will take place of any previous tasks that were registered @@ -89,7 +84,7 @@ impl AtomicTask { /// idea. Concurrent calls to `register` will attempt to register different /// tasks to be notified. One of the callers will win and have its task set, /// but there is no guarantee as to which caller will succeed. - pub fn register_task(&self, task: Task) { + pub(crate) fn register(&self, task: Task) { match self.state.compare_and_swap(WAITING, LOCKED_WRITE, Acquire) { WAITING => { unsafe { diff --git a/tokio-reactor/src/background.rs b/tokio-reactor/src/background.rs index 1f07c7b89..03f057cbe 100644 --- a/tokio-reactor/src/background.rs +++ b/tokio-reactor/src/background.rs @@ -1,7 +1,7 @@ -use {Reactor, Handle}; +use {Reactor, Handle, Task}; use atomic_task::AtomicTask; -use futures::{Future, Async, Poll}; +use futures::{Future, Async, Poll, task}; use std::io; use std::thread; @@ -136,7 +136,8 @@ impl Future for Shutdown { type Error = (); fn poll(&mut self) -> Poll<(), ()> { - self.inner.shared.shutdown_task.register(); + let task = Task::Futures1(task::current()); + self.inner.shared.shutdown_task.register(task); if !self.inner.is_shutdown() { return Ok(Async::NotReady); diff --git a/tokio-reactor/src/lib.rs b/tokio-reactor/src/lib.rs index ae105824a..a0ce1788c 100644 --- a/tokio-reactor/src/lib.rs +++ b/tokio-reactor/src/lib.rs @@ -39,6 +39,9 @@ extern crate slab; extern crate tokio_executor; extern crate tokio_io; +#[cfg(feature = "unstable-futures")] +extern crate futures2; + pub(crate) mod background; mod atomic_task; mod poll_evented; @@ -69,7 +72,6 @@ use std::time::{Duration, Instant}; use log::Level; use mio::event::Evented; use slab::Slab; -use futures::task::Task; /// The core reactor, or event loop. /// @@ -155,6 +157,14 @@ fn _assert_kinds() { _assert::(); } +/// A wakeup handle for a task, which may be either a futures 0.1 or 0.2 task +#[derive(Debug, Clone)] +pub(crate) enum Task { + Futures1(futures::task::Task), + #[cfg(feature = "unstable-futures")] + Futures2(futures2::task::Waker), +} + // ===== impl Reactor ===== /// Set the default reactor for the duration of the closure @@ -578,7 +588,7 @@ impl Inner { Direction::Write => (&sched.writer, mio::Ready::writable()), }; - task.register_task(t); + task.register(t); if sched.readiness.load(SeqCst) & ready.as_usize() != 0 { task.notify(); @@ -611,6 +621,17 @@ impl Direction { } } +impl Task { + fn notify(&self) { + match *self { + Task::Futures1(ref task) => task.notify(), + + #[cfg(feature = "unstable-futures")] + Task::Futures2(ref waker) => waker.wake(), + } + } +} + #[cfg(all(unix, not(target_os = "fuchsia")))] mod platform { use mio::Ready; @@ -637,3 +658,19 @@ mod platform { false } } + +#[cfg(feature = "unstable-futures")] +fn lift_async(old: futures::Async) -> futures2::Async { + match old { + futures::Async::Ready(x) => futures2::Async::Ready(x), + futures::Async::NotReady => futures2::Async::Pending, + } +} + +#[cfg(feature = "unstable-futures")] +fn lower_async(new: futures2::Async) -> futures::Async { + match new { + futures2::Async::Ready(x) => futures::Async::Ready(x), + futures2::Async::Pending => futures::Async::NotReady, + } +} diff --git a/tokio-reactor/src/poll_evented.rs b/tokio-reactor/src/poll_evented.rs index bddf94c79..c9f5b77d2 100644 --- a/tokio-reactor/src/poll_evented.rs +++ b/tokio-reactor/src/poll_evented.rs @@ -5,6 +5,9 @@ use mio; use mio::event::Evented; use tokio_io::{AsyncRead, AsyncWrite}; +#[cfg(feature = "unstable-futures")] +use futures2; + use std::fmt; use std::io::{self, Read, Write}; use std::sync::atomic::AtomicUsize; @@ -99,7 +102,7 @@ struct Inner { // ===== impl PollEvented ===== macro_rules! poll_ready { - ($me:expr, $mask:expr, $cache:ident, $poll:ident, $take:ident) => {{ + ($me:expr, $mask:expr, $cache:ident, $take:ident, $poll:expr) => {{ $me.register()?; // Load cached & encoded readiness. @@ -114,7 +117,7 @@ macro_rules! poll_ready { // stream. This happens in a loop to ensure that the stream gets // drained. loop { - let ready = try_ready!($me.inner.registration.$poll()); + let ready = try_ready!($poll); cached |= ready.as_usize(); // Update the cache store @@ -210,7 +213,23 @@ where E: Evented /// * called from outside of a task context. pub fn poll_read_ready(&self, mask: mio::Ready) -> Poll { assert!(!mask.is_writable(), "cannot poll for write readiness"); - poll_ready!(self, mask, read_readiness, poll_read_ready, take_read_ready) + poll_ready!( + self, mask, read_readiness, take_read_ready, + self.inner.registration.poll_read_ready() + ) + } + + /// Like `poll_read_ready` but compatible with futures 0.2. + #[cfg(feature = "unstable-futures")] + pub fn poll_read_ready2(&self, cx: &mut futures2::task::Context, mask: mio::Ready) + -> futures2::Poll + { + assert!(!mask.is_writable(), "cannot poll for write readiness"); + let mut res = || poll_ready!( + self, mask, read_readiness, take_read_ready, + self.inner.registration.poll_read_ready2(cx).map(::lower_async) + ); + res().map(::lift_async) } /// Clears the I/O resource's read readiness state and registers the current @@ -243,6 +262,25 @@ where E: Evented Ok(()) } + /// Like `clear_read_ready` but compatible with futures 0.2. + #[cfg(feature = "unstable-futures")] + pub fn clear_read_ready2(&self, cx: &mut futures2::task::Context, ready: mio::Ready) + -> io::Result<()> + { + // Cannot clear write readiness + assert!(!ready.is_writable(), "cannot clear write readiness"); + assert!(!::platform::is_hup(&ready), "cannot clear HUP readiness"); + + self.inner.read_readiness.fetch_and(!ready.as_usize(), Relaxed); + + if self.poll_read_ready2(cx, ready)?.is_ready() { + // Notify the current task + cx.waker().wake() + } + + Ok(()) + } + /// Check the I/O resource's write readiness state. /// /// This always checks for writable readiness and also checks for HUP @@ -263,13 +301,31 @@ where E: Evented /// * `ready` contains bits besides `writable` and `hup`. /// * called from outside of a task context. pub fn poll_write_ready(&self) -> Poll { - poll_ready!(self, - mio::Ready::writable(), - write_readiness, - poll_write_ready, - take_write_ready) + poll_ready!( + self, + mio::Ready::writable(), + write_readiness, + take_write_ready, + self.inner.registration.poll_write_ready() + ) } + /// Like `poll_write_ready` but compatible with futures 0.2. + #[cfg(feature = "unstable-futures")] + pub fn poll_write_ready2(&self, cx: &mut futures2::task::Context) + -> futures2::Poll + { + let mut res = || poll_ready!( + self, + mio::Ready::writable(), + write_readiness, + take_write_ready, + self.inner.registration.poll_write_ready2(cx).map(::lower_async) + ); + res().map(::lift_async) + } + + /// Resets the I/O resource's write readiness state and registers the current /// task to be notified once a write readiness event is received. /// @@ -295,6 +351,21 @@ where E: Evented Ok(()) } + /// Like `clear_write_ready`, but compatible with futures 0.2. + #[cfg(feature = "unstable-futures")] + pub fn clear_write_ready2(&self, cx: &mut futures2::task::Context) -> io::Result<()> { + let ready = mio::Ready::writable(); + + self.inner.read_readiness.fetch_and(!ready.as_usize(), Relaxed); + + if self.poll_write_ready2(cx)?.is_ready() { + // Notify the current task + cx.waker().wake() + } + + Ok(()) + } + /// Ensure that the I/O resource is registered with the reactor. fn register(&self) -> io::Result<()> { self.inner.registration.register(self.io.as_ref().unwrap())?; @@ -322,6 +393,28 @@ where E: Evented + Read, } } +#[cfg(feature = "unstable-futures")] +impl futures2::io::AsyncRead for PollEvented + where E: Evented, E: Read, +{ + fn poll_read(&mut self, cx: &mut futures2::task::Context, buf: &mut [u8]) + -> futures2::Poll + { + if let futures2::Async::Pending = self.poll_read_ready2(cx, mio::Ready::readable())? { + return Ok(futures2::Async::Pending); + } + + match self.get_mut().read(buf) { + Ok(n) => Ok(futures2::Async::Ready(n)), + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { + self.clear_read_ready2(cx, mio::Ready::readable())?; + Ok(futures2::Async::Pending) + } + Err(e) => Err(e), + } + } +} + impl Write for PollEvented where E: Evented + Write, { @@ -354,6 +447,48 @@ where E: Evented + Write, } } +#[cfg(feature = "unstable-futures")] +impl futures2::io::AsyncWrite for PollEvented + where E: Evented, E: Write, +{ + fn poll_write(&mut self, cx: &mut futures2::task::Context, buf: &[u8]) + -> futures2::Poll + { + if let futures2::Async::Pending = self.poll_write_ready2(cx)? { + return Ok(futures2::Async::Pending); + } + + match self.get_mut().write(buf) { + Ok(n) => Ok(futures2::Async::Ready(n)), + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { + self.clear_write_ready2(cx)?; + Ok(futures2::Async::Pending) + } + Err(e) => Err(e), + } + } + + fn poll_flush(&mut self, cx: &mut futures2::task::Context) -> futures2::Poll<(), io::Error> { + if let futures2::Async::Pending = self.poll_write_ready2(cx)? { + return Ok(futures2::Async::Pending); + } + + match self.get_mut().flush() { + Ok(_) => Ok(futures2::Async::Ready(())), + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { + self.clear_write_ready2(cx)?; + Ok(futures2::Async::Pending) + } + Err(e) => Err(e), + } + } + + fn poll_close(&mut self, cx: &mut futures2::task::Context) -> futures2::Poll<(), io::Error> { + futures2::io::AsyncWrite::poll_flush(self, cx) + } +} + + impl AsyncRead for PollEvented where E: Evented + Read, { @@ -387,6 +522,28 @@ where E: Evented, &'a E: Read, } } +#[cfg(feature = "unstable-futures")] +impl<'a, E> futures2::io::AsyncRead for &'a PollEvented + where E: Evented, &'a E: Read, +{ + fn poll_read(&mut self, cx: &mut futures2::task::Context, buf: &mut [u8]) + -> futures2::Poll + { + if let futures2::Async::Pending = self.poll_read_ready2(cx, mio::Ready::readable())? { + return Ok(futures2::Async::Pending); + } + + match self.get_ref().read(buf) { + Ok(n) => Ok(futures2::Async::Ready(n)), + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { + self.clear_read_ready2(cx, mio::Ready::readable())?; + Ok(futures2::Async::Pending) + } + Err(e) => Err(e), + } + } +} + impl<'a, E> Write for &'a PollEvented where E: Evented, &'a E: Write, { @@ -419,6 +576,47 @@ where E: Evented, &'a E: Write, } } +#[cfg(feature = "unstable-futures")] +impl<'a, E> futures2::io::AsyncWrite for &'a PollEvented + where E: Evented, &'a E: Write, +{ + fn poll_write(&mut self, cx: &mut futures2::task::Context, buf: &[u8]) + -> futures2::Poll + { + if let futures2::Async::Pending = self.poll_write_ready2(cx)? { + return Ok(futures2::Async::Pending); + } + + match self.get_ref().write(buf) { + Ok(n) => Ok(futures2::Async::Ready(n)), + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { + self.clear_write_ready2(cx)?; + Ok(futures2::Async::Pending) + } + Err(e) => Err(e), + } + } + + fn poll_flush(&mut self, cx: &mut futures2::task::Context) -> futures2::Poll<(), io::Error> { + if let futures2::Async::Pending = self.poll_write_ready2(cx)? { + return Ok(futures2::Async::Pending); + } + + match self.get_ref().flush() { + Ok(_) => Ok(futures2::Async::Ready(())), + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { + self.clear_write_ready2(cx)?; + Ok(futures2::Async::Pending) + } + Err(e) => Err(e), + } + } + + fn poll_close(&mut self, cx: &mut futures2::task::Context) -> futures2::Poll<(), io::Error> { + futures2::io::AsyncWrite::poll_flush(self, cx) + } +} + impl<'a, E> AsyncRead for &'a PollEvented where E: Evented, &'a E: Read, { @@ -439,7 +637,6 @@ fn is_wouldblock(r: &io::Result) -> bool { } } - impl fmt::Debug for PollEvented { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { f.debug_struct("PollEvented") diff --git a/tokio-reactor/src/registration.rs b/tokio-reactor/src/registration.rs index a726bdd33..84199ea23 100644 --- a/tokio-reactor/src/registration.rs +++ b/tokio-reactor/src/registration.rs @@ -1,9 +1,11 @@ -use {Handle, Direction}; +use {Handle, Direction, Task}; -use futures::{Async, Poll}; -use futures::task::{self, Task}; +use futures::{Async, Poll, task}; use mio::{self, Evented}; +#[cfg(feature = "unstable-futures")] +use futures2; + use std::{io, mem, usize}; use std::cell::UnsafeCell; use std::sync::atomic::AtomicUsize; @@ -271,13 +273,26 @@ impl Registration { /// /// This function will panic if called from outside of a task context. pub fn poll_read_ready(&self) -> Poll { - self.poll_ready(Direction::Read, true) + self.poll_ready(Direction::Read, true, || Task::Futures1(task::current())) .map(|v| match v { Some(v) => Async::Ready(v), _ => Async::NotReady, }) } + /// Like `poll_ready_ready`, but compatible with futures 0.2 + #[cfg(feature = "unstable-futures")] + pub fn poll_read_ready2(&self, cx: &mut futures2::task::Context) + -> futures2::Poll + { + use futures2::Async as Async2; + self.poll_ready(Direction::Read, true, || Task::Futures2(cx.waker().clone())) + .map(|v| match v { + Some(v) => Async2::Ready(v), + _ => Async2::Pending, + }) + } + /// Consume any pending read readiness event. /// /// This function is identical to [`poll_read_ready`] **except** that it @@ -286,7 +301,7 @@ impl Registration { /// /// [`poll_read_ready`]: #method.poll_read_ready pub fn take_read_ready(&self) -> io::Result> { - self.poll_ready(Direction::Read, false) + self.poll_ready(Direction::Read, false, || panic!()) } @@ -323,13 +338,26 @@ impl Registration { /// /// This function will panic if called from outside of a task context. pub fn poll_write_ready(&self) -> Poll { - self.poll_ready(Direction::Write, true) + self.poll_ready(Direction::Write, true, || Task::Futures1(task::current())) .map(|v| match v { Some(v) => Async::Ready(v), _ => Async::NotReady, }) } + /// Like `poll_write_ready`, but compatible with futures 0.2 + #[cfg(feature = "unstable-futures")] + pub fn poll_write_ready2(&self, cx: &mut futures2::task::Context) + -> futures2::Poll + { + use futures2::Async as Async2; + self.poll_ready(Direction::Write, true, || Task::Futures2(cx.waker().clone())) + .map(|v| match v { + Some(v) => Async2::Ready(v), + _ => Async2::Pending, + }) + } + /// Consume any pending write readiness event. /// /// This function is identical to [`poll_write_ready`] **except** that it @@ -338,11 +366,12 @@ impl Registration { /// /// [`poll_write_ready`]: #method.poll_write_ready pub fn take_write_ready(&self) -> io::Result> { - self.poll_ready(Direction::Write, false) + self.poll_ready(Direction::Write, false, || unreachable!()) } - fn poll_ready(&self, direction: Direction, notify: bool) + fn poll_ready(&self, direction: Direction, notify: bool, task: F) -> io::Result> + where F: Fn() -> Task { let mut state = self.state.load(SeqCst); @@ -357,7 +386,7 @@ impl Registration { } READY => { let inner = unsafe { (*self.inner.get()).as_ref().unwrap() }; - return inner.poll_ready(direction, notify); + return inner.poll_ready(direction, notify, task); } _ => { if !notify { @@ -371,7 +400,7 @@ impl Registration { let mut n = node.take().unwrap_or_else(|| { Box::new(Node { direction, - task: task::current(), + task: task(), next: None, }) }); @@ -472,8 +501,9 @@ impl Inner { inner.deregister_source(io) } - fn poll_ready(&self, direction: Direction, notify: bool) + fn poll_ready(&self, direction: Direction, notify: bool, task: F) -> io::Result> + where F: FnOnce() -> Task { if self.token == ERROR { return Err(io::Error::new(io::ErrorKind::Other, "failed to associate with reactor")); @@ -504,8 +534,8 @@ impl Inner { if ready.is_empty() && notify { // Update the task info match direction { - Direction::Read => sched.reader.register(), - Direction::Write => sched.writer.register(), + Direction::Read => sched.reader.register(task()), + Direction::Write => sched.writer.register(task()), } // Try again diff --git a/tokio-threadpool/Cargo.toml b/tokio-threadpool/Cargo.toml index 3a68043a0..f781f9f93 100644 --- a/tokio-threadpool/Cargo.toml +++ b/tokio-threadpool/Cargo.toml @@ -19,8 +19,13 @@ crossbeam-deque = "0.3" num_cpus = "1.2" rand = "0.4" log = "0.3" +futures2 = { version = "0.1", path = "../futures2", optional = true } [dev-dependencies] tokio-timer = "0.1" env_logger = "0.4" futures-cpupool = "0.1.7" + +[features] +unstable-futures = ["futures2", "tokio-executor/unstable-futures"] +default = [] diff --git a/tokio-threadpool/src/lib.rs b/tokio-threadpool/src/lib.rs index d5b6e09fe..d809f27fc 100644 --- a/tokio-threadpool/src/lib.rs +++ b/tokio-threadpool/src/lib.rs @@ -12,6 +12,9 @@ extern crate rand; #[macro_use] extern crate log; +#[cfg(feature = "unstable-futures")] +extern crate futures2; + mod task; use tokio_executor::{Enter, SpawnError}; @@ -33,6 +36,14 @@ use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering::{AcqRel, Acquire, Release, Relaxed}; use std::time::{Instant, Duration}; +#[derive(Debug)] +struct ShutdownTask { + task1: AtomicTask, + + #[cfg(feature = "unstable-futures")] + task2: futures2::task::AtomicWaker, +} + /// Work-stealing based thread pool for executing futures. /// /// If a `ThreadPool` instance is dropped without explicitly being shutdown, @@ -160,7 +171,7 @@ struct Inner { workers: Box<[WorkerEntry]>, // Task notified when the worker shuts down - shutdown_task: AtomicTask, + shutdown_task: ShutdownTask, // Configuration config: Config, @@ -180,6 +191,12 @@ struct Notifier { inner: Weak, } +#[cfg(feature = "unstable-futures")] +struct Futures2Wake { + notifier: Arc, + id: usize, +} + /// ThreadPool state. /// /// The two least significant bits are the shutdown flags. (0 for active, 1 for @@ -532,7 +549,11 @@ impl Builder { num_workers: AtomicUsize::new(self.pool_size), next_thread_id: AtomicUsize::new(0), workers: workers.into_boxed_slice(), - shutdown_task: AtomicTask::new(), + shutdown_task: ShutdownTask { + task1: AtomicTask::new(), + #[cfg(feature = "unstable-futures")] + task2: futures2::task::AtomicWaker::new(), + }, config: self.config.clone(), }); @@ -772,6 +793,11 @@ impl tokio_executor::Executor for Sender { let mut s = &*self; tokio_executor::Executor::spawn(&mut s, future) } + + #[cfg(feature = "unstable-futures")] + fn spawn2(&mut self, f: Task2) -> Result<(), futures2::executor::SpawnError> { + futures2::executor::Executor::spawn(self, f) + } } impl<'a> tokio_executor::Executor for &'a Sender { @@ -806,6 +832,11 @@ impl<'a> tokio_executor::Executor for &'a Sender { Ok(()) } + + #[cfg(feature = "unstable-futures")] + fn spawn2(&mut self, f: Task2) -> Result<(), futures2::executor::SpawnError> { + futures2::executor::Executor::spawn(self, f) + } } impl future::Executor for Sender @@ -827,6 +858,48 @@ where T: Future + Send + 'static, } } +#[cfg(feature = "unstable-futures")] +type Task2 = Box + Send>; + +#[cfg(feature = "unstable-futures")] +impl futures2::executor::Executor for Sender { + fn spawn(&mut self, f: Task2) -> Result<(), futures2::executor::SpawnError> { + let mut s = &*self; + futures2::executor::Executor::spawn(&mut s, f) + } + + fn status(&self) -> Result<(), futures2::executor::SpawnError> { + let s = &*self; + futures2::executor::Executor::status(&s) + } +} + +#[cfg(feature = "unstable-futures")] +impl<'a> futures2::executor::Executor for &'a Sender { + fn spawn(&mut self, f: Task2) -> Result<(), futures2::executor::SpawnError> { + self.prepare_for_spawn() + // TODO: get rid of this once the futures crate adds more error types + .map_err(|_| futures2::executor::SpawnError::shutdown())?; + + // At this point, the pool has accepted the future, so schedule it for + // execution. + + // Create a new task for the future + let task = Task::new2(f, |id| into_waker(Arc::new(Futures2Wake::new(id, &self.inner)))); + + self.inner.submit(task, &self.inner); + + Ok(()) + } + + fn status(&self) -> Result<(), futures2::executor::SpawnError> { + tokio_executor::Executor::status(self) + // TODO: get rid of this once the futures crate adds more error types + .map_err(|_| futures2::executor::SpawnError::shutdown()) + } +} + + impl Clone for Sender { #[inline] fn clone(&self) -> Sender { @@ -835,6 +908,21 @@ impl Clone for Sender { } } +// ===== impl ShutdownTask ===== + +impl ShutdownTask { + #[cfg(not(feature = "unstable-futures"))] + fn notify(&self) { + self.task1.notify(); + } + + #[cfg(feature = "unstable-futures")] + fn notify(&self) { + self.task1.notify(); + self.task2.wake(); + } +} + // ===== impl Shutdown ===== impl Shutdown { @@ -850,7 +938,7 @@ impl Future for Shutdown { fn poll(&mut self) -> Poll<(), ()> { trace!("Shutdown::poll"); - self.inner().shutdown_task.register(); + self.inner().shutdown_task.task1.register(); if 0 != self.inner().num_workers.load(Acquire) { return Ok(Async::NotReady); @@ -860,6 +948,24 @@ impl Future for Shutdown { } } +#[cfg(feature = "unstable-futures")] +impl futures2::Future for Shutdown { + type Item = (); + type Error = (); + + fn poll(&mut self, cx: &mut futures2::task::Context) -> futures2::Poll<(), ()> { + trace!("Shutdown::poll"); + + self.inner().shutdown_task.task2.register(cx.waker()); + + if 0 != self.inner().num_workers.load(Acquire) { + return Ok(futures2::Async::Pending); + } + + Ok(().into()) + } +} + // ===== impl Inner ===== impl Inner { @@ -1346,6 +1452,7 @@ impl Worker { let notify = Arc::new(Notifier { inner: Arc::downgrade(&self.inner), }); + let mut sender = Sender { inner: self.inner.clone() }; let mut first = true; let mut spin_cnt = 0; @@ -1358,14 +1465,14 @@ impl Worker { let consistent = self.drain_inbound(); // Run the next available task - if self.try_run_task(¬ify) { + if self.try_run_task(¬ify, &mut sender) { spin_cnt = 0; // As long as there is work, keep looping. continue; } // No work in this worker's queue, it is time to try stealing. - if self.try_steal_task(¬ify) { + if self.try_steal_task(¬ify, &mut sender) { spin_cnt = 0; continue; } @@ -1448,13 +1555,13 @@ impl Worker { /// /// Returns `true` if work was found. #[inline] - fn try_run_task(&self, notify: &Arc) -> bool { + fn try_run_task(&self, notify: &Arc, sender: &mut Sender) -> bool { use deque::Steal::*; // Poll the internal queue for a task to run match self.entry().deque.steal() { Data(task) => { - self.run_task(task, notify); + self.run_task(task, notify, sender); true } Empty => false, @@ -1466,7 +1573,7 @@ impl Worker { /// /// Returns `true` if work was found #[inline] - fn try_steal_task(&self, notify: &Arc) -> bool { + fn try_steal_task(&self, notify: &Arc, sender: &mut Sender) -> bool { use deque::Steal::*; let len = self.inner.workers.len(); @@ -1480,7 +1587,7 @@ impl Worker { Data(task) => { trace!("stole task"); - self.run_task(task, notify); + self.run_task(task, notify, sender); trace!("try_steal_task -- signal_work; self={}; from={}", self.idx, idx); @@ -1507,10 +1614,10 @@ impl Worker { found_work } - fn run_task(&self, task: Task, notify: &Arc) { + fn run_task(&self, task: Task, notify: &Arc, sender: &mut Sender) { use task::Run::*; - match task.run(notify) { + match task.run(notify, sender) { Idle => {} Schedule => { self.entry().push_internal(task); @@ -2111,3 +2218,56 @@ impl fmt::Debug for Callback { write!(fmt, "Fn") } } + +// ===== impl Futures2Wake ===== + +#[cfg(feature = "unstable-futures")] +impl Futures2Wake { + fn new(id: usize, inner: &Arc) -> Futures2Wake { + let notifier = Arc::new(Notifier { + inner: Arc::downgrade(inner), + }); + Futures2Wake { id, notifier } + } +} + +#[cfg(feature = "unstable-futures")] +impl Drop for Futures2Wake { + fn drop(&mut self) { + self.notifier.drop_id(self.id) + } +} + +#[cfg(feature = "unstable-futures")] +struct ArcWrapped(PhantomData); + +#[cfg(feature = "unstable-futures")] +unsafe impl futures2::task::UnsafeWake for ArcWrapped { + unsafe fn clone_raw(&self) -> futures2::task::Waker { + let me: *const ArcWrapped = self; + let arc = (*(&me as *const *const ArcWrapped as *const Arc)).clone(); + arc.notifier.clone_id(arc.id); + into_waker(arc) + } + + unsafe fn drop_raw(&self) { + let mut me: *const ArcWrapped = self; + let me = &mut me as *mut *const ArcWrapped as *mut Arc; + (*me).notifier.drop_id((*me).id); + ::std::ptr::drop_in_place(me); + } + + unsafe fn wake(&self) { + let me: *const ArcWrapped = self; + let me = &me as *const *const ArcWrapped as *const Arc; + (*me).notifier.notify((*me).id) + } +} + +#[cfg(feature = "unstable-futures")] +fn into_waker(rc: Arc) -> futures2::task::Waker { + unsafe { + let ptr = mem::transmute::, *mut ArcWrapped>(rc); + futures2::task::Waker::new(ptr) + } +} diff --git a/tokio-threadpool/src/task.rs b/tokio-threadpool/src/task.rs index 9a2167feb..59deca07b 100644 --- a/tokio-threadpool/src/task.rs +++ b/tokio-threadpool/src/task.rs @@ -1,6 +1,6 @@ -use Notifier; +use {Notifier, Sender}; -use futures::{future, Future, Async}; +use futures::{self, future, Future, Async}; use futures::executor::{self, Spawn}; use std::{fmt, mem, panic, ptr}; @@ -9,6 +9,9 @@ use std::sync::Arc; use std::sync::atomic::{self, AtomicUsize, AtomicPtr}; use std::sync::atomic::Ordering::{AcqRel, Acquire, Release, Relaxed}; +#[cfg(feature = "unstable-futures")] +use futures2; + pub(crate) struct Task { ptr: *mut Inner, } @@ -34,6 +37,22 @@ pub(crate) enum Run { Complete, } +type BoxFuture = Box + Send + 'static>; + +#[cfg(feature = "unstable-futures")] +type BoxFuture2 = Box + Send>; + +enum TaskFuture { + Futures1(Spawn), + + #[cfg(feature = "unstable-futures")] + Futures2 { + tls: futures2::task::LocalMap, + waker: futures2::task::Waker, + fut: BoxFuture2, + } +} + struct Inner { // Next pointer in the queue that submits tasks to a worker. next: AtomicPtr, @@ -47,7 +66,7 @@ struct Inner { // Store the future at the head of the struct // // The future is dropped immediately when it transitions to Complete - future: Option>, + future: Option, } #[derive(Debug, Clone, Copy, Eq, PartialEq)] @@ -64,23 +83,41 @@ enum State { Complete, } -type BoxFuture = Box + Send + 'static>; - // ===== impl Task ===== impl Task { /// Create a new task handle pub fn new(future: BoxFuture) -> Task { + let task_fut = TaskFuture::Futures1(executor::spawn(future)); let inner = Box::new(Inner { next: AtomicPtr::new(ptr::null_mut()), state: AtomicUsize::new(State::new().into()), ref_count: AtomicUsize::new(1), - future: Some(executor::spawn(future)), + future: Some(task_fut), }); Task { ptr: Box::into_raw(inner) } } + /// Create a new task handle for a futures 0.2 future + #[cfg(feature = "unstable-futures")] + pub fn new2(fut: BoxFuture2, make_waker: F) -> Task + where F: FnOnce(usize) -> futures2::task::Waker + { + let mut inner = Box::new(Inner { + next: AtomicPtr::new(ptr::null_mut()), + state: AtomicUsize::new(State::new().into()), + ref_count: AtomicUsize::new(1), + future: None, + }); + + let waker = make_waker((&*inner) as *const _ as usize); + let tls = futures2::task::LocalMap::new(); + inner.future = Some(TaskFuture::Futures2 { waker, tls, fut }); + + Task { ptr: Box::into_raw(inner) } + } + /// Transmute a u64 to a Task pub unsafe fn from_notify_id(unpark_id: usize) -> Task { mem::transmute(unpark_id) @@ -93,7 +130,7 @@ impl Task { /// Execute the task returning `Run::Schedule` if the task needs to be /// scheduled again. - pub fn run(&self, unpark: &Arc) -> Run { + pub fn run(&self, unpark: &Arc, exec: &mut Sender) -> Run { use self::State::*; // Transition task to running state. At this point, the task must be @@ -118,7 +155,7 @@ impl Task { // `thread::panicking() -> true`. To do this, the future is dropped from // within the catch_unwind block. let res = panic::catch_unwind(panic::AssertUnwindSafe(|| { - struct Guard<'a>(&'a mut Option>, bool); + struct Guard<'a>(&'a mut Option, bool); impl<'a> Drop for Guard<'a> { fn drop(&mut self) { @@ -132,7 +169,7 @@ impl Task { let mut g = Guard(fut, true); let ret = g.0.as_mut().unwrap() - .poll_future_notify(unpark, self.ptr as usize); + .poll(unpark, self.ptr as usize, exec); g.1 = false; @@ -302,7 +339,7 @@ impl Inner { next: AtomicPtr::new(ptr::null_mut()), state: AtomicUsize::new(State::stub().into()), ref_count: AtomicUsize::new(0), - future: Some(executor::spawn(Box::new(future::empty()))), + future: Some(TaskFuture::Futures1(executor::spawn(Box::new(future::empty())))), } } @@ -454,3 +491,23 @@ impl From for usize { } } } + +// ===== impl TaskFuture ===== + +impl TaskFuture { + #[allow(unused_variables)] + fn poll(&mut self, unpark: &Arc, id: usize, exec: &mut Sender) -> futures::Poll<(), ()> { + match *self { + TaskFuture::Futures1(ref mut fut) => fut.poll_future_notify(unpark, id), + + #[cfg(feature = "unstable-futures")] + TaskFuture::Futures2 { ref mut fut, ref waker, ref mut tls } => { + let mut cx = futures2::task::Context::new(tls, waker, exec); + match fut.poll(&mut cx).unwrap() { + futures2::Async::Pending => Ok(Async::NotReady), + futures2::Async::Ready(x) => Ok(Async::Ready(x)), + } + } + } + } +} diff --git a/tokio-threadpool/tests/threadpool.rs b/tokio-threadpool/tests/threadpool.rs index c76031a2c..61dd5cf22 100644 --- a/tokio-threadpool/tests/threadpool.rs +++ b/tokio-threadpool/tests/threadpool.rs @@ -3,9 +3,20 @@ extern crate tokio_executor; extern crate futures; extern crate env_logger; +#[cfg(feature = "unstable-futures")] +extern crate futures2; + use tokio_threadpool::*; -use futures::{Poll, Sink, Stream, Async}; -use futures::future::{Future, lazy}; + +#[cfg(not(feature = "unstable-futures"))] +use futures::{Poll, Sink, Stream, Async, Future}; +#[cfg(not(feature = "unstable-futures"))] +use futures::future::lazy; + +#[cfg(feature = "unstable-futures")] +use futures2::prelude::*; +#[cfg(feature = "unstable-futures")] +use futures2::future::lazy; use std::cell::Cell; use std::sync::{mpsc, Arc}; @@ -15,6 +26,57 @@ use std::time::Duration; thread_local!(static FOO: Cell = Cell::new(0)); +#[cfg(not(feature = "unstable-futures"))] +fn spawn_pool(pool: &mut Sender, f: F) + where F: Future + Send + 'static +{ + pool.spawn(f).unwrap() +} +#[cfg(feature = "unstable-futures")] +fn spawn_pool(pool: &mut Sender, f: F) + where F: Future + Send + 'static +{ + futures2::executor::Executor::spawn( + pool, + Box::new(f.map_err(|_| panic!())) + ).unwrap() +} + +#[cfg(not(feature = "unstable-futures"))] +fn spawn_default(f: F) + where F: Future + Send + 'static +{ + tokio_executor::spawn(f) +} +#[cfg(feature = "unstable-futures")] +fn spawn_default(f: F) + where F: Future + Send + 'static +{ + tokio_executor::spawn2(Box::new(f.map_err(|_| panic!()))) +} + +fn ignore_results(f: F) -> Box + Send> { + Box::new(f.map(|_| ()).map_err(|_| ())) +} + +#[cfg(feature = "unstable-futures")] +fn await_shutdown(shutdown: Shutdown) { + futures::Future::wait(shutdown).unwrap() +} +#[cfg(not(feature = "unstable-futures"))] +fn await_shutdown(shutdown: Shutdown) { + shutdown.wait().unwrap() +} + +#[cfg(not(feature = "unstable-futures"))] +fn block_on(f: F) -> Result { + f.wait() +} +#[cfg(feature = "unstable-futures")] +fn block_on(f: F) -> Result { + futures2::executor::block_on(f) +} + #[test] fn natural_shutdown_simple_futures() { let _ = ::env_logger::init(); @@ -33,29 +95,29 @@ fn natural_shutdown_simple_futures() { NUM_DEC.fetch_add(1, Relaxed); }) .build(); - let tx = pool.sender().clone(); + let mut tx = pool.sender().clone(); let a = { let (t, rx) = mpsc::channel(); - tx.spawn(lazy(move || { + spawn_pool(&mut tx, lazy(move || { // Makes sure this runs on a worker thread FOO.with(|f| assert_eq!(f.get(), 0)); t.send("one").unwrap(); Ok(()) - })).unwrap(); + })); rx }; let b = { let (t, rx) = mpsc::channel(); - tx.spawn(lazy(move || { + spawn_pool(&mut tx, lazy(move || { // Makes sure this runs on a worker thread FOO.with(|f| assert_eq!(f.get(), 0)); t.send("two").unwrap(); Ok(()) - })).unwrap(); + })); rx }; @@ -65,7 +127,7 @@ fn natural_shutdown_simple_futures() { assert_eq!("two", b.recv().unwrap()); // Wait for the pool to shutdown - pool.shutdown().wait().unwrap(); + await_shutdown(pool.shutdown()); // Assert that at least one thread started let num_inc = NUM_INC.load(Relaxed); @@ -89,6 +151,7 @@ fn force_shutdown_drops_futures() { struct Never(Arc); + #[cfg(not(feature = "unstable-futures"))] impl Future for Never { type Item = (); type Error = (); @@ -98,6 +161,16 @@ fn force_shutdown_drops_futures() { } } + #[cfg(feature = "unstable-futures")] + impl Future for Never { + type Item = (); + type Error = (); + + fn poll(&mut self, _: &mut futures2::task::Context) -> Poll<(), ()> { + Ok(Async::Pending) + } + } + impl Drop for Never { fn drop(&mut self) { self.0.fetch_add(1, Relaxed); @@ -116,10 +189,10 @@ fn force_shutdown_drops_futures() { .build(); let mut tx = pool.sender().clone(); - tx.spawn(Never(num_drop.clone())).unwrap(); + spawn_pool(&mut tx, Never(num_drop.clone())); // Wait for the pool to shutdown - pool.shutdown_now().wait().unwrap(); + await_shutdown(pool.shutdown_now()); // Assert that only a single thread was spawned. let a = num_inc.load(Relaxed); @@ -146,6 +219,7 @@ fn drop_threadpool_drops_futures() { struct Never(Arc); + #[cfg(not(feature = "unstable-futures"))] impl Future for Never { type Item = (); type Error = (); @@ -155,6 +229,16 @@ fn drop_threadpool_drops_futures() { } } + #[cfg(feature = "unstable-futures")] + impl Future for Never { + type Item = (); + type Error = (); + + fn poll(&mut self, _: &mut futures2::task::Context) -> Poll<(), ()> { + Ok(Async::Pending) + } + } + impl Drop for Never { fn drop(&mut self) { self.0.fetch_add(1, Relaxed); @@ -173,7 +257,7 @@ fn drop_threadpool_drops_futures() { .build(); let mut tx = pool.sender().clone(); - tx.spawn(Never(num_drop.clone())).unwrap(); + spawn_pool(&mut tx, Never(num_drop.clone())); // Wait for the pool to shutdown drop(pool); @@ -211,13 +295,13 @@ fn thread_shutdown_timeout() { let _ = t.lock().unwrap().send(()); }) .build(); - let tx = pool.sender().clone(); + let mut tx = pool.sender().clone(); let t = complete_tx.clone(); - tx.spawn(lazy(move || { + spawn_pool(&mut tx, lazy(move || { t.send(()).unwrap(); Ok(()) - })).unwrap(); + })); // The future completes complete_rx.recv().unwrap(); @@ -226,14 +310,14 @@ fn thread_shutdown_timeout() { shutdown_rx.recv().unwrap(); // Futures can still be run - tx.spawn(lazy(move || { + spawn_pool(&mut tx, lazy(move || { complete_tx.send(()).unwrap(); Ok(()) - })).unwrap(); + })); complete_rx.recv().unwrap(); - pool.shutdown().wait().unwrap(); + await_shutdown(pool.shutdown()); } #[test] @@ -249,14 +333,14 @@ fn many_oneshot_futures() { for _ in 0..NUM { let cnt = cnt.clone(); - tx.spawn(lazy(move || { + spawn_pool(&mut tx, lazy(move || { cnt.fetch_add(1, Relaxed); Ok(()) - })).unwrap(); + })); } // Wait for the pool to shutdown - pool.shutdown().wait().unwrap(); + await_shutdown(pool.shutdown()); let num = cnt.load(Relaxed); assert_eq!(num, NUM); @@ -265,8 +349,12 @@ fn many_oneshot_futures() { #[test] fn many_multishot_futures() { + #[cfg(not(feature = "unstable-futures"))] use futures::sync::mpsc; + #[cfg(feature = "unstable-futures")] + use futures2::channel::mpsc; + const CHAIN: usize = 200; const CYCLES: usize = 5; const TRACKS: usize = 50; @@ -290,11 +378,11 @@ fn many_multishot_futures() { .map_err(|e| panic!("{:?}", e)); // Forward all the messages - pool_tx.spawn(next_tx + spawn_pool(&mut pool_tx, next_tx .send_all(rx) .map(|_| ()) .map_err(|e| panic!("{:?}", e)) - ).unwrap(); + ); chain_rx = next_rx; } @@ -304,7 +392,7 @@ fn many_multishot_futures() { let cycle_tx = start_tx.clone(); let mut rem = CYCLES; - pool_tx.spawn(chain_rx.take(CYCLES as u64).for_each(move |msg| { + let task = chain_rx.take(CYCLES as u64).for_each(move |msg| { rem -= 1; let send = if rem == 0 { final_tx.clone().send(msg) @@ -316,83 +404,109 @@ fn many_multishot_futures() { res.unwrap(); Ok(()) }) - })).unwrap(); + }); + spawn_pool(&mut pool_tx, ignore_results(task)); start_txs.push(start_tx); final_rxs.push(final_rx); } for start_tx in start_txs { - start_tx.send("ping").wait().unwrap(); + block_on(start_tx.send("ping")).unwrap(); } for final_rx in final_rxs { - final_rx.wait().next().unwrap().unwrap(); + block_on(final_rx.into_future()).unwrap(); } // Shutdown the pool - pool.shutdown().wait().unwrap(); + await_shutdown(pool.shutdown()); } } #[test] fn global_executor_is_configured() { let pool = ThreadPool::new(); - let tx = pool.sender().clone(); + let mut tx = pool.sender().clone(); let (signal_tx, signal_rx) = mpsc::channel(); - tx.spawn(lazy(move || { - tokio_executor::spawn(lazy(move || { + spawn_pool(&mut tx, lazy(move || { + spawn_default(lazy(move || { signal_tx.send(()).unwrap(); Ok(()) })); Ok(()) - })).unwrap(); + })); signal_rx.recv().unwrap(); - pool.shutdown().wait().unwrap(); + await_shutdown(pool.shutdown()); } #[test] fn new_threadpool_is_idle() { let pool = ThreadPool::new(); - pool.shutdown_on_idle().wait().unwrap(); + await_shutdown(pool.shutdown_on_idle()); } #[test] fn busy_threadpool_is_not_idle() { + #[cfg(not(feature = "unstable-futures"))] use futures::sync::oneshot; + #[cfg(feature = "unstable-futures")] + use futures2::channel::oneshot; + let pool = ThreadPool::new(); - let tx = pool.sender().clone(); + let mut tx = pool.sender().clone(); let (term_tx, term_rx) = oneshot::channel(); - tx.spawn(term_rx.then(|_| { + spawn_pool(&mut tx, term_rx.then(|_| { Ok(()) - })).unwrap(); + })); let mut idle = pool.shutdown_on_idle(); - futures::lazy(|| { - assert!(idle.poll().unwrap().is_not_ready()); - Ok::<_, ()>(()) - }).wait().unwrap(); + struct IdleFut<'a>(&'a mut Shutdown); + + #[cfg(not(feature = "unstable-futures"))] + impl<'a> Future for IdleFut<'a> { + type Item = (); + type Error = (); + fn poll(&mut self) -> Poll<(), ()> { + assert!(self.0.poll().unwrap().is_not_ready()); + Ok(Async::Ready(())) + } + } + + #[cfg(feature = "unstable-futures")] + impl<'a> Future for IdleFut<'a> { + type Item = (); + type Error = (); + fn poll(&mut self, cx: &mut futures2::task::Context) -> Poll<(), ()> { + assert!(self.0.poll(cx).unwrap().is_pending()); + Ok(Async::Ready(())) + } + } + + block_on(IdleFut(&mut idle)).unwrap(); term_tx.send(()).unwrap(); - idle.wait().unwrap(); + await_shutdown(idle); } #[test] fn panic_in_task() { let pool = ThreadPool::new(); + let mut tx = pool.sender().clone(); struct Boom; + #[cfg(not(feature = "unstable-futures"))] impl Future for Boom { type Item = (); type Error = (); @@ -402,13 +516,23 @@ fn panic_in_task() { } } + #[cfg(feature = "unstable-futures")] + impl Future for Boom { + type Item = (); + type Error = (); + + fn poll(&mut self, _cx: &mut futures2::task::Context) -> Poll<(), ()> { + panic!(); + } + } + impl Drop for Boom { fn drop(&mut self) { assert!(::std::thread::panicking()); } } - pool.spawn(Boom); + spawn_pool(&mut tx, Boom); - pool.shutdown_on_idle().wait().unwrap(); + await_shutdown(pool.shutdown_on_idle()); }