From d91c775f360d875afb36b56a812ff5f77940981a Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Thu, 9 Aug 2018 21:56:53 -0700 Subject: [PATCH] Remove dead futures2 code. (#538) The futures 0.2 crate is not intended for widespread usage. Also, the futures team is exploring the compat shim route. If futures 0.3 support is added to Tokio 0.1, then a different integration route will be explored, making the current code unhelpful. --- src/executor/mod.rs | 23 --- src/lib.rs | 6 - src/runtime/mod.rs | 27 --- src/runtime/task_executor.rs | 23 --- tests/echo2.rs | 53 ----- tests/global2.rs | 122 ------------ tests/tcp2.rs | 136 ------------- tokio-current-thread/src/lib.rs | 27 --- tokio-current-thread/tests/current_thread.rs | 2 - tokio-executor/src/enter.rs | 9 - tokio-executor/src/global.rs | 22 +-- tokio-executor/src/lib.rs | 18 -- tokio-reactor/src/background.rs | 4 +- tokio-reactor/src/lib.rs | 39 +--- tokio-reactor/src/poll_evented.rs | 193 ------------------ tokio-reactor/src/registration.rs | 33 +--- tokio-tcp/src/incoming.rs | 15 -- tokio-tcp/src/lib.rs | 19 -- tokio-tcp/src/listener.rs | 38 ---- tokio-tcp/src/stream.rs | 173 ---------------- tokio-threadpool/src/builder.rs | 3 - tokio-threadpool/src/futures2_wake.rs | 60 ------ tokio-threadpool/src/lib.rs | 5 - tokio-threadpool/src/pool/mod.rs | 4 +- tokio-threadpool/src/sender.rs | 55 ------ tokio-threadpool/src/shutdown.rs | 22 +-- tokio-threadpool/src/shutdown_task.rs | 16 +- tokio-threadpool/src/task/mod.rs | 73 +------ tokio-threadpool/src/worker/mod.rs | 26 ++- tokio-threadpool/tests/threadpool.rs | 198 ++++--------------- tokio-udp/src/lib.rs | 3 - 31 files changed, 67 insertions(+), 1380 deletions(-) delete mode 100644 tests/echo2.rs delete mode 100644 tests/global2.rs delete mode 100644 tests/tcp2.rs delete mode 100644 tokio-threadpool/src/futures2_wake.rs diff --git a/src/executor/mod.rs b/src/executor/mod.rs index 5528d1572..3f238f8fa 100644 --- a/src/executor/mod.rs +++ b/src/executor/mod.rs @@ -61,9 +61,6 @@ pub use tokio_executor::{Executor, DefaultExecutor, SpawnError}; use futures::{Future, IntoFuture}; use futures::future::{self, FutureResult}; -#[cfg(feature = "unstable-futures")] -use futures2; - /// Return value from the `spawn` function. /// /// Currently this value doesn't actually provide any functionality. However, it @@ -133,15 +130,6 @@ where F: Future + 'static + Send Spawn(()) } -/// Like `spawn`, but compatible with futures 0.2 -#[cfg(feature = "unstable-futures")] -pub fn spawn2(f: F) -> Spawn - where F: futures2::Future + 'static + Send -{ - ::tokio_executor::spawn2(f); - Spawn(()) -} - impl IntoFuture for Spawn { type Future = FutureResult<(), ()>; type Item = (); @@ -151,14 +139,3 @@ impl IntoFuture for Spawn { future::ok(()) } } - -#[cfg(feature = "unstable-futures")] -impl futures2::IntoFuture for Spawn { - type Future = futures2::future::FutureResult<(), ()>; - type Item = (); - type Error = (); - - fn into_future(self) -> Self::Future { - futures2::future::ok(()) - } -} diff --git a/src/lib.rs b/src/lib.rs index ea6582fd8..c0a7d8a09 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -80,9 +80,6 @@ extern crate tokio_timer; extern crate tokio_tcp; extern crate tokio_udp; -#[cfg(feature = "unstable-futures")] -extern crate futures2; - pub mod clock; pub mod executor; pub mod fs; @@ -93,9 +90,6 @@ pub mod timer; pub mod util; pub use executor::spawn; -#[cfg(feature = "unstable-futures")] -pub use executor::spawn2; - pub use runtime::run; pub mod io { diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs index 38feb53f7..2e5b344d4 100644 --- a/src/runtime/mod.rs +++ b/src/runtime/mod.rs @@ -129,8 +129,6 @@ use tokio_threadpool as threadpool; use futures; use futures::future::Future; -#[cfg(feature = "unstable-futures")] -use futures2; /// Handle to the Tokio runtime. /// @@ -215,18 +213,6 @@ 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. /// @@ -353,19 +339,6 @@ 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 - } - /// Run a future to completion on the Tokio runtime. /// /// This runs the given future on the runtime, blocking until it is diff --git a/src/runtime/task_executor.rs b/src/runtime/task_executor.rs index ed918be5f..e213201ab 100644 --- a/src/runtime/task_executor.rs +++ b/src/runtime/task_executor.rs @@ -2,8 +2,6 @@ use tokio_threadpool::Sender; use futures::future::{self, Future}; -#[cfg(feature = "unstable-futures")] -use futures2; /// Executes futures on the runtime /// @@ -74,25 +72,4 @@ 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) - } } diff --git a/tests/echo2.rs b/tests/echo2.rs deleted file mode 100644 index 6ead07d81..000000000 --- a/tests/echo2.rs +++ /dev/null @@ -1,53 +0,0 @@ -#![cfg(feature = "unstable-futures")] - -// This test is the same as `echo.rs`, but ported to futures 0.2 - -extern crate env_logger; -extern crate futures2; -extern crate tokio; -extern crate tokio_io; - -use std::io::{Read, Write}; -use std::net::TcpStream; -use std::thread; - -use futures2::prelude::*; -use futures2::executor::block_on; -use tokio::net::TcpListener; - -macro_rules! t { - ($e:expr) => (match $e { - Ok(e) => e, - Err(e) => panic!("{} failed with {:?}", stringify!($e), e), - }) -} - -#[test] -fn echo_server() { - drop(env_logger::init()); - - let srv = t!(TcpListener::bind(&t!("127.0.0.1:0".parse()))); - let addr = t!(srv.local_addr()); - - let msg = "foo bar baz"; - let t = thread::spawn(move || { - let mut s = TcpStream::connect(&addr).unwrap(); - - for _i in 0..1024 { - assert_eq!(t!(s.write(msg.as_bytes())), msg.len()); - let mut buf = [0; 1024]; - assert_eq!(t!(s.read(&mut buf)), msg.len()); - assert_eq!(&buf[..msg.len()], msg.as_bytes()); - } - }); - - let clients = srv.incoming(); - let client = clients.next().map(|e| e.0.unwrap()).map_err(|e| e.0); - let halves = client.map(|s| s.split()); - let copied = halves.and_then(|(a, b)| a.copy_into(b)); - - let (amt, _, _) = t!(block_on(copied)); - t.join().unwrap(); - - assert_eq!(amt, msg.len() as u64 * 1024); -} diff --git a/tests/global2.rs b/tests/global2.rs deleted file mode 100644 index 24244abb1..000000000 --- a/tests/global2.rs +++ /dev/null @@ -1,122 +0,0 @@ -#![cfg(feature = "unstable-futures")] - -// This test is the same as `global.rs`, but ported to futures 0.2 - -extern crate futures; -extern crate futures2; -extern crate tokio; -extern crate tokio_io; -extern crate env_logger; - -use std::{io, thread}; -use std::sync::Arc; - -use futures2::prelude::*; -use futures2::executor::block_on; -use futures2::task; - -use tokio::net::{TcpStream, TcpListener}; -use tokio::runtime::Runtime; - -macro_rules! t { - ($e:expr) => (match $e { - Ok(e) => e, - Err(e) => panic!("{} failed with {:?}", stringify!($e), e), - }) -} - -#[test] -fn hammer() { - let _ = env_logger::init(); - - let threads = (0..10).map(|_| { - thread::spawn(|| { - let srv = t!(TcpListener::bind(&"127.0.0.1:0".parse().unwrap())); - let addr = t!(srv.local_addr()); - let mine = TcpStream::connect(&addr); - let theirs = srv.incoming().next() - .map(|(s, _)| s.unwrap()) - .map_err(|(s, _)| s); - let (mine, theirs) = t!(block_on(mine.join(theirs))); - - assert_eq!(t!(mine.local_addr()), t!(theirs.peer_addr())); - assert_eq!(t!(theirs.local_addr()), t!(mine.peer_addr())); - }) - }).collect::>(); - for thread in threads { - thread.join().unwrap(); - } -} - -struct Rd(Arc); -struct Wr(Arc); - -impl AsyncRead for Rd { - fn poll_read(&mut self, cx: &mut task::Context, dst: &mut [u8]) -> Poll { - <&TcpStream>::poll_read(&mut &*self.0, cx, dst) - } -} - -impl AsyncWrite for Wr { - fn poll_write(&mut self, cx: &mut task::Context, src: &[u8]) -> Poll { - <&TcpStream>::poll_write(&mut &*self.0, cx, src) - } - - fn poll_flush(&mut self, _cx: &mut task::Context) -> Poll<(), io::Error> { - Ok(().into()) - } - - fn poll_close(&mut self, _cx: &mut task::Context) -> Poll<(), io::Error> { - Ok(().into()) - } -} - -#[test] -fn hammer_split() { - const N: usize = 100; - - let _ = env_logger::init(); - - let srv = t!(TcpListener::bind(&"127.0.0.1:0".parse().unwrap())); - let addr = t!(srv.local_addr()); - - let mut rt = Runtime::new().unwrap(); - - fn split(socket: TcpStream) { - let socket = Arc::new(socket); - let rd = Rd(socket.clone()); - let wr = Wr(socket); - - let rd = rd.read(vec![0; 1]) - .map(|_| ()) - .map_err(|e| panic!("read error = {:?}", e)); - - let wr = wr.write_all(b"1") - .map(|_| ()) - .map_err(|e| panic!("write error = {:?}", e)); - - tokio::spawn2(rd); - tokio::spawn2(wr); - } - - rt.spawn2({ - srv.incoming() - .map_err(|e| panic!("accept error = {:?}", e)) - .take(N as u64) - .for_each(|socket| { - split(socket); - Ok(()) - }) - .map(|_| ()) - }); - - for _ in 0..N { - rt.spawn2({ - TcpStream::connect(&addr) - .map_err(|e| panic!("connect error = {:?}", e)) - .map(|socket| split(socket)) - }); - } - - futures::Future::wait(rt.shutdown_on_idle()).unwrap(); -} diff --git a/tests/tcp2.rs b/tests/tcp2.rs deleted file mode 100644 index 4fbc978cb..000000000 --- a/tests/tcp2.rs +++ /dev/null @@ -1,136 +0,0 @@ -#![cfg(feature = "unstable-futures")] - -// This test is the same as `tcp.rs`, but ported to futures 0.2 - -extern crate env_logger; -extern crate tokio; -extern crate mio; -extern crate futures2; - -use std::{net, thread}; -use std::sync::mpsc::channel; - -use tokio::net::{TcpListener, TcpStream}; -use futures2::executor::block_on; -use futures2::prelude::*; - -macro_rules! t { - ($e:expr) => (match $e { - Ok(e) => e, - Err(e) => panic!("{} failed with {:?}", stringify!($e), e), - }) -} - -#[test] -fn connect() { - drop(env_logger::init()); - let srv = t!(net::TcpListener::bind("127.0.0.1:0")); - let addr = t!(srv.local_addr()); - let t = thread::spawn(move || { - t!(srv.accept()).0 - }); - - let stream = TcpStream::connect(&addr); - let mine = t!(block_on(stream)); - let theirs = t.join().unwrap(); - - assert_eq!(t!(mine.local_addr()), t!(theirs.peer_addr())); - assert_eq!(t!(theirs.local_addr()), t!(mine.peer_addr())); -} - -#[test] -fn accept() { - drop(env_logger::init()); - let srv = t!(TcpListener::bind(&t!("127.0.0.1:0".parse()))); - let addr = t!(srv.local_addr()); - - let (tx, rx) = channel(); - let client = srv.incoming().map(move |t| { - tx.send(()).unwrap(); - t - }).next().map_err(|e| e.0); - assert!(rx.try_recv().is_err()); - let t = thread::spawn(move || { - net::TcpStream::connect(&addr).unwrap() - }); - - let (mine, _remaining) = t!(block_on(client)); - let mine = mine.unwrap(); - let theirs = t.join().unwrap(); - - assert_eq!(t!(mine.local_addr()), t!(theirs.peer_addr())); - assert_eq!(t!(theirs.local_addr()), t!(mine.peer_addr())); -} - -#[test] -fn accept2() { - drop(env_logger::init()); - let srv = t!(TcpListener::bind(&t!("127.0.0.1:0".parse()))); - let addr = t!(srv.local_addr()); - - let t = thread::spawn(move || { - net::TcpStream::connect(&addr).unwrap() - }); - - let (tx, rx) = channel(); - let client = srv.incoming().map(move |t| { - tx.send(()).unwrap(); - t - }).next().map_err(|e| e.0); - assert!(rx.try_recv().is_err()); - - let (mine, _remaining) = t!(block_on(client)); - mine.unwrap(); - t.join().unwrap(); -} - -#[cfg(unix)] -mod unix { - use tokio::net::TcpStream; - use tokio::prelude::*; - - use env_logger; - use futures2::future; - use futures2::executor::block_on; - use futures2::io::AsyncRead; - use mio::unix::UnixReady; - - use std::{net, thread}; - use std::time::Duration; - - #[test] - fn poll_hup() { - drop(env_logger::init()); - - let srv = t!(net::TcpListener::bind("127.0.0.1:0")); - let addr = t!(srv.local_addr()); - let t = thread::spawn(move || { - let mut client = t!(srv.accept()).0; - client.write(b"hello world").unwrap(); - thread::sleep(Duration::from_millis(200)); - }); - - let mut stream = t!(block_on(TcpStream::connect(&addr))); - - // Poll for HUP before reading. - block_on(future::poll_fn(|cx| { - stream.poll_read_ready2(cx, UnixReady::hup().into()) - })).unwrap(); - - // Same for write half - block_on(future::poll_fn(|cx| { - stream.poll_write_ready2(cx) - })).unwrap(); - - let mut buf = vec![0; 11]; - - // Read the data - block_on(future::poll_fn(|cx| { - stream.poll_read(cx, &mut buf) - })).unwrap(); - - assert_eq!(b"hello world", &buf[..]); - - t.join().unwrap(); - } -} diff --git a/tokio-current-thread/src/lib.rs b/tokio-current-thread/src/lib.rs index e8794c9bb..08f614e7a 100644 --- a/tokio-current-thread/src/lib.rs +++ b/tokio-current-thread/src/lib.rs @@ -45,9 +45,6 @@ use std::rc::Rc; use std::sync::{atomic, mpsc, Arc}; 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. @@ -410,13 +407,6 @@ impl tokio_executor::Executor for CurrentThread { self.borrow().spawn_local(future, false); 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

{ @@ -699,23 +689,6 @@ 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() { - Ok(()) - } else { - Err(SpawnError::shutdown()) - } - }) - } } impl Executor for TaskExecutor diff --git a/tokio-current-thread/tests/current_thread.rs b/tokio-current-thread/tests/current_thread.rs index 5d4d8124e..d40cf6f43 100644 --- a/tokio-current-thread/tests/current_thread.rs +++ b/tokio-current-thread/tests/current_thread.rs @@ -1,5 +1,3 @@ -#![cfg(not(feature = "unstable-futures"))] - extern crate tokio_current_thread; extern crate tokio_executor; extern crate futures; diff --git a/tokio-executor/src/enter.rs b/tokio-executor/src/enter.rs index cb2879cec..c33cf6492 100644 --- a/tokio-executor/src/enter.rs +++ b/tokio-executor/src/enter.rs @@ -3,9 +3,6 @@ use std::cell::Cell; use std::error::Error; use std::fmt; -#[cfg(feature = "unstable-futures")] -use futures2; - thread_local!(static ENTERED: Cell = Cell::new(false)); /// Represents an executor context. @@ -14,9 +11,6 @@ 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 @@ -66,9 +60,6 @@ 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 ca7dbcfaf..24867cd57 100644 --- a/tokio-executor/src/global.rs +++ b/tokio-executor/src/global.rs @@ -4,9 +4,6 @@ use futures::Future; use std::cell::Cell; -#[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 @@ -65,7 +62,7 @@ enum State { Ready(*mut Executor), // default executor is currently active (used to detect recursive calls) Active -} +} /// Thread-local tracking the current executor thread_local!(static EXECUTOR: Cell = Cell::new(State::Empty)); @@ -80,14 +77,6 @@ impl super::Executor for DefaultExecutor { .unwrap_or_else(|| Err(SpawnError::shutdown())) } - #[cfg(feature = "unstable-futures")] - fn spawn2(&mut self, future: Box + Send>) - -> Result<(), futures2::executor::SpawnError> - { - DefaultExecutor::with_current(|executor| executor.spawn2(future)) - .unwrap_or_else(|| Err(futures2::executor::SpawnError::shutdown())) - } - fn status(&self) -> Result<(), SpawnError> { DefaultExecutor::with_current(|executor| executor.status()) .unwrap_or_else(|| Err(SpawnError::shutdown())) @@ -142,15 +131,6 @@ 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 356832232..aee48891d 100644 --- a/tokio-executor/src/lib.rs +++ b/tokio-executor/src/lib.rs @@ -37,9 +37,6 @@ extern crate futures; -#[cfg(feature = "unstable-futures")] -extern crate futures2; - mod enter; mod global; pub mod park; @@ -47,9 +44,6 @@ 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; use std::error::Error; @@ -142,11 +136,6 @@ pub trait Executor { fn spawn(&mut self, future: Box + Send>) -> 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. /// /// This function may return both false positives **and** false negatives. @@ -194,13 +183,6 @@ impl Executor for Box { (**self).spawn(future) } - #[cfg(feature = "unstable-futures")] - fn spawn2(&mut self, future: Box + Send>) - -> Result<(), futures2::executor::SpawnError> - { - (**self).spawn2(future) - } - fn status(&self) -> Result<(), SpawnError> { (**self).status() } diff --git a/tokio-reactor/src/background.rs b/tokio-reactor/src/background.rs index 88f78a8bc..6544b738f 100644 --- a/tokio-reactor/src/background.rs +++ b/tokio-reactor/src/background.rs @@ -1,4 +1,4 @@ -use {Reactor, Handle, Task}; +use {Reactor, Handle}; use atomic_task::AtomicTask; use futures::{Future, Async, Poll, task}; @@ -136,7 +136,7 @@ impl Future for Shutdown { type Error = (); fn poll(&mut self) -> Poll<(), ()> { - let task = Task::Futures1(task::current()); + let task = task::current(); self.inner.shared.shutdown_task.register_task(task); if !self.inner.is_shutdown() { diff --git a/tokio-reactor/src/lib.rs b/tokio-reactor/src/lib.rs index f7735f5e2..cae4373bb 100644 --- a/tokio-reactor/src/lib.rs +++ b/tokio-reactor/src/lib.rs @@ -44,9 +44,6 @@ extern crate slab; extern crate tokio_executor; extern crate tokio_io; -#[cfg(feature = "unstable-futures")] -extern crate futures2; - mod atomic_task; pub(crate) mod background; mod poll_evented; @@ -64,6 +61,7 @@ pub use self::poll_evented::PollEvented; use atomic_task::AtomicTask; use sharded_rwlock::RwLock; +use futures::task::Task; use tokio_executor::Enter; use tokio_executor::park::{Park, Unpark}; @@ -184,14 +182,6 @@ 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 @@ -726,17 +716,6 @@ 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(unix)] mod platform { use mio::Ready; @@ -764,22 +743,6 @@ mod platform { } } -#[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, - } -} - // ===== impl SetFallbackError ===== impl fmt::Display for SetFallbackError { diff --git a/tokio-reactor/src/poll_evented.rs b/tokio-reactor/src/poll_evented.rs index 3723c98a1..2352a874a 100644 --- a/tokio-reactor/src/poll_evented.rs +++ b/tokio-reactor/src/poll_evented.rs @@ -5,9 +5,6 @@ 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; @@ -228,19 +225,6 @@ where E: Evented ) } - /// 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 /// task to be notified once a read readiness event is received. /// @@ -271,25 +255,6 @@ 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 @@ -319,22 +284,6 @@ where E: Evented ) } - /// 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. /// @@ -360,21 +309,6 @@ 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.write_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())?; @@ -402,28 +336,6 @@ 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, { @@ -456,48 +368,6 @@ 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, { @@ -531,28 +401,6 @@ 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, { @@ -585,47 +433,6 @@ 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, { diff --git a/tokio-reactor/src/registration.rs b/tokio-reactor/src/registration.rs index 981dd5524..9b79f00e5 100644 --- a/tokio-reactor/src/registration.rs +++ b/tokio-reactor/src/registration.rs @@ -3,9 +3,6 @@ use {Handle, HandlePriv, Direction, Task}; use futures::{Async, Poll, task}; use mio::{self, Evented}; -#[cfg(feature = "unstable-futures")] -use futures2; - use std::{io, ptr, usize}; use std::cell::UnsafeCell; use std::sync::atomic::AtomicUsize; @@ -279,26 +276,13 @@ 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, || Task::Futures1(task::current())) + self.poll_ready(Direction::Read, true, || 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 @@ -344,26 +328,13 @@ 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, || Task::Futures1(task::current())) + self.poll_ready(Direction::Write, true, || 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 diff --git a/tokio-tcp/src/incoming.rs b/tokio-tcp/src/incoming.rs index 6726224b8..7db6414c0 100644 --- a/tokio-tcp/src/incoming.rs +++ b/tokio-tcp/src/incoming.rs @@ -5,9 +5,6 @@ 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"] @@ -31,15 +28,3 @@ 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/tokio-tcp/src/lib.rs b/tokio-tcp/src/lib.rs index c7713ee21..1a6d8ef2a 100644 --- a/tokio-tcp/src/lib.rs +++ b/tokio-tcp/src/lib.rs @@ -30,9 +30,6 @@ extern crate mio; extern crate tokio_io; extern crate tokio_reactor; -#[cfg(feature = "unstable-futures")] -extern crate futures2; - mod incoming; mod listener; mod stream; @@ -41,19 +38,3 @@ pub use self::incoming::Incoming; pub use self::listener::TcpListener; pub use self::stream::TcpStream; pub use self::stream::ConnectFuture; - -#[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-tcp/src/listener.rs b/tokio-tcp/src/listener.rs index 1eff35567..4e386478e 100644 --- a/tokio-tcp/src/listener.rs +++ b/tokio-tcp/src/listener.rs @@ -9,9 +9,6 @@ use futures::{Poll, Async}; use mio; use tokio_reactor::{Handle, PollEvented}; -#[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 @@ -66,22 +63,6 @@ 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)> { @@ -123,25 +104,6 @@ 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/tokio-tcp/src/stream.rs b/tokio-tcp/src/stream.rs index a1ed6bca7..02ebbea9e 100644 --- a/tokio-tcp/src/stream.rs +++ b/tokio-tcp/src/stream.rs @@ -11,9 +11,6 @@ use mio; use tokio_io::{AsyncRead, AsyncWrite}; use tokio_reactor::{Handle, PollEvented}; -#[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 @@ -138,14 +135,6 @@ impl TcpStream { self.io.poll_read_ready(mask) } - /// 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 - { - self.io.poll_read_ready2(cx, mask) - } - /// Check the TCP stream's write readiness state. /// /// This always checks for writable readiness and also checks for HUP @@ -164,14 +153,6 @@ impl TcpStream { self.io.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 - { - self.io.poll_write_ready2(cx) - } - /// Returns the local address that this stream is bound to. pub fn local_addr(&self) -> io::Result { self.io.get_ref().local_addr() @@ -222,25 +203,6 @@ 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 @@ -411,25 +373,6 @@ 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) - } - - fn poll_vectored_read(&mut self, cx: &mut futures2::task::Context, vec: &mut [&mut IoVec]) - -> futures2::Poll - { - futures2::io::AsyncRead::poll_vectored_read(&mut &*self, cx, vec) - } - - unsafe fn initializer(&self) -> futures2::io::Initializer { - futures2::io::Initializer::nop() - } -} - impl AsyncWrite for TcpStream { fn shutdown(&mut self) -> Poll<(), io::Error> { <&TcpStream>::shutdown(&mut &*self) @@ -440,29 +383,6 @@ 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_vectored_write(&mut self, cx: &mut futures2::task::Context, vec: &[&IoVec]) - -> futures2::Poll - { - futures2::io::AsyncWrite::poll_vectored_write(&mut &*self, cx, vec) - } - - 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 { @@ -535,40 +455,6 @@ 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) - } - - fn poll_vectored_read(&mut self, cx: &mut futures2::task::Context, vec: &mut [&mut IoVec]) - -> futures2::Poll - { - if let futures2::Async::Pending = self.io.poll_read_ready2(cx, mio::Ready::readable())? { - return Ok(futures2::Async::Pending) - } - - let r = self.io.get_ref().read_bufs(vec); - - match r { - Ok(n) => { - Ok(futures2::Async::Ready(n)) - } - 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), - } - } - - unsafe fn initializer(&self) -> futures2::io::Initializer { - futures2::io::Initializer::nop() - } -} - impl<'a> AsyncWrite for &'a TcpStream { fn shutdown(&mut self) -> Poll<(), io::Error> { Ok(().into()) @@ -603,44 +489,6 @@ 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_vectored_write(&mut self, cx: &mut futures2::task::Context, vec: &[&IoVec]) - -> futures2::Poll - { - if let futures2::Async::Pending = self.io.poll_write_ready2(cx)? { - return Ok(futures2::Async::Pending) - } - - let r = self.io.get_ref().write_bufs(vec); - - match r { - Ok(n) => { - Ok(futures2::Async::Ready(n)) - } - Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { - self.io.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> { - 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) @@ -656,16 +504,6 @@ impl Future for ConnectFuture { } } -#[cfg(feature = "unstable-futures")] -impl futures2::Future for ConnectFuture { - type Item = TcpStream; - type Error = io::Error; - - 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 PollEvented) -> Poll @@ -714,17 +552,6 @@ impl Future for ConnectFutureState { } } -#[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(unix)] mod sys { use std::os::unix::prelude::*; diff --git a/tokio-threadpool/src/builder.rs b/tokio-threadpool/src/builder.rs index 11df6c49e..5d5a7e5af 100644 --- a/tokio-threadpool/src/builder.rs +++ b/tokio-threadpool/src/builder.rs @@ -16,9 +16,6 @@ use num_cpus; use tokio_executor::Enter; use tokio_executor::park::Park; -#[cfg(feature = "unstable-futures")] -use futures2; - /// Builds a thread pool with custom configuration values. /// /// Methods can be chained in order to set the configuration values. The thread diff --git a/tokio-threadpool/src/futures2_wake.rs b/tokio-threadpool/src/futures2_wake.rs deleted file mode 100644 index ed9d4552c..000000000 --- a/tokio-threadpool/src/futures2_wake.rs +++ /dev/null @@ -1,60 +0,0 @@ -use inner::Pool; -use notifier::Notifier; - -use std::marker::PhantomData; -use std::mem; -use std::sync::Arc; - -use futures::executor::Notify; -use futures2; - -pub(crate) struct Futures2Wake { - notifier: Arc, - id: usize, -} - -impl Futures2Wake { - pub(crate) fn new(id: usize, inner: &Arc) -> Futures2Wake { - let notifier = Arc::new(Notifier { - inner: Arc::downgrade(inner), - }); - Futures2Wake { id, notifier } - } -} - -impl Drop for Futures2Wake { - fn drop(&mut self) { - self.notifier.drop_id(self.id) - } -} - -struct ArcWrapped(PhantomData); - -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) - } -} - -pub(crate) 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/lib.rs b/tokio-threadpool/src/lib.rs index d43bae0bc..dd44c0514 100644 --- a/tokio-threadpool/src/lib.rs +++ b/tokio-threadpool/src/lib.rs @@ -89,9 +89,6 @@ extern crate rand; #[macro_use] extern crate log; -#[cfg(feature = "unstable-futures")] -extern crate futures2; - // ## Crate layout // // The primary type, `Pool`, holds the majority of a thread pool's state, @@ -148,8 +145,6 @@ mod blocking; mod builder; mod callback; mod config; -#[cfg(feature = "unstable-futures")] -mod futures2_wake; mod notifier; mod pool; mod sender; diff --git a/tokio-threadpool/src/pool/mod.rs b/tokio-threadpool/src/pool/mod.rs index 3364aa7cd..7a029aaec 100644 --- a/tokio-threadpool/src/pool/mod.rs +++ b/tokio-threadpool/src/pool/mod.rs @@ -116,9 +116,7 @@ impl Pool { backup_stack, blocking, shutdown_task: ShutdownTask { - task1: AtomicTask::new(), - #[cfg(feature = "unstable-futures")] - task2: futures2::task::AtomicWaker::new(), + task: AtomicTask::new(), }, config, }; diff --git a/tokio-threadpool/src/sender.rs b/tokio-threadpool/src/sender.rs index d540e5589..e2e6584c0 100644 --- a/tokio-threadpool/src/sender.rs +++ b/tokio-threadpool/src/sender.rs @@ -6,10 +6,6 @@ use std::sync::atomic::Ordering::{AcqRel, Acquire}; use tokio_executor::{self, SpawnError}; use futures::{future, Future}; -#[cfg(feature = "unstable-futures")] -use futures2; -#[cfg(feature = "unstable-futures")] -use futures2_wake::{into_waker, Futures2Wake}; /// Submit futures to the associated thread pool for execution. /// @@ -135,11 +131,6 @@ 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 { @@ -174,11 +165,6 @@ 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 @@ -200,47 +186,6 @@ 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 { diff --git a/tokio-threadpool/src/shutdown.rs b/tokio-threadpool/src/shutdown.rs index 29f2a342f..eb58ff084 100644 --- a/tokio-threadpool/src/shutdown.rs +++ b/tokio-threadpool/src/shutdown.rs @@ -2,8 +2,6 @@ use pool::Pool; use sender::Sender; use futures::{Future, Poll, Async}; -#[cfg(feature = "unstable-futures")] -use futures2; /// Future that resolves when the thread pool is shutdown. /// @@ -34,7 +32,7 @@ impl Future for Shutdown { fn poll(&mut self) -> Poll<(), ()> { use futures::task; - self.inner().shutdown_task.task1.register_task(task::current()); + self.inner().shutdown_task.task.register_task(task::current()); if !self.inner().is_shutdown() { return Ok(Async::NotReady); @@ -43,21 +41,3 @@ impl Future for Shutdown { Ok(().into()) } } - -#[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()) - } -} diff --git a/tokio-threadpool/src/shutdown_task.rs b/tokio-threadpool/src/shutdown_task.rs index 2d6e87d21..3dcbb4284 100644 --- a/tokio-threadpool/src/shutdown_task.rs +++ b/tokio-threadpool/src/shutdown_task.rs @@ -1,24 +1,12 @@ use futures::task::AtomicTask; -#[cfg(feature = "unstable-futures")] -use futures2; #[derive(Debug)] pub(crate) struct ShutdownTask { - pub task1: AtomicTask, - - #[cfg(feature = "unstable-futures")] - pub task2: futures2::task::AtomicWaker, + pub task: AtomicTask, } impl ShutdownTask { - #[cfg(not(feature = "unstable-futures"))] pub fn notify(&self) { - self.task1.notify(); - } - - #[cfg(feature = "unstable-futures")] - pub fn notify(&self) { - self.task1.notify(); - self.task2.wake(); + self.task.notify(); } } diff --git a/tokio-threadpool/src/task/mod.rs b/tokio-threadpool/src/task/mod.rs index 84309ffd6..1632c23ae 100644 --- a/tokio-threadpool/src/task/mod.rs +++ b/tokio-threadpool/src/task/mod.rs @@ -10,7 +10,6 @@ use self::state::State; use notifier::Notifier; use pool::Pool; -use sender::Sender; use futures::{self, Future, Async}; use futures::executor::{self, Spawn}; @@ -21,9 +20,6 @@ use std::sync::Arc; use std::sync::atomic::{AtomicUsize, AtomicPtr}; use std::sync::atomic::Ordering::{AcqRel, Release, Relaxed}; -#[cfg(feature = "unstable-futures")] -use futures2; - /// Harness around a future. /// /// This also behaves as a node in the inbound work queue and the blocking @@ -44,7 +40,7 @@ pub(crate) struct Task { /// Store the future at the head of the struct /// /// The future is dropped immediately when it transitions to Complete - future: UnsafeCell>, + future: UnsafeCell>>, } #[derive(Debug)] @@ -56,27 +52,13 @@ pub(crate) enum Run { 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, - } -} - // ===== impl Task ===== impl Task { /// Create a new `Task` as a harness for `future`. pub fn new(future: BoxFuture) -> Task { // Wrap the future with an execution context. - let task_fut = TaskFuture::Futures1(executor::spawn(future)); + let task_fut = executor::spawn(future); Task { state: AtomicUsize::new(State::new().into()), @@ -87,31 +69,11 @@ impl Task { } } - /// Create a new `Task` as a harness 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(Task { - state: AtomicUsize::new(State::new().into()), - blocking: AtomicUsize::new(BlockingState::new().into()), - next: AtomicPtr::new(ptr::null_mut()), - next_blocking: AtomicPtr::new(ptr::null_mut()), - 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) } - } - /// Create a fake `Task` to be used as part of the intrusive mpsc channel /// algorithm. fn stub() -> Task { - let future = Box::new(futures::empty()); - let task_fut = TaskFuture::Futures1(executor::spawn(future)); + let future = Box::new(futures::empty()) as BoxFuture; + let task_fut = executor::spawn(future); Task { state: AtomicUsize::new(State::stub().into()), @@ -124,7 +86,7 @@ impl Task { /// Execute the task returning `Run::Schedule` if the task needs to be /// scheduled again. - pub fn run(&self, unpark: &Arc, exec: &mut Sender) -> Run { + pub fn run(&self, unpark: &Arc) -> Run { use self::State::*; // Transition task to running state. At this point, the task must be @@ -149,7 +111,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) { @@ -163,8 +125,7 @@ impl Task { let mut g = Guard(fut, true); let ret = g.0.as_mut().unwrap() - .poll(unpark, self as *const _ as usize, exec); - + .poll_future_notify(unpark, self as *const _ as usize); g.1 = false; @@ -282,23 +243,3 @@ impl fmt::Debug for Task { .finish() } } - -// ===== 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/src/worker/mod.rs b/tokio-threadpool/src/worker/mod.rs index 037441f39..b71dbd267 100644 --- a/tokio-threadpool/src/worker/mod.rs +++ b/tokio-threadpool/src/worker/mod.rs @@ -224,7 +224,6 @@ 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; @@ -238,7 +237,7 @@ impl Worker { let consistent = self.drain_inbound(); // Run the next available task - if self.try_run_task(¬ify, &mut sender) { + if self.try_run_task(¬ify) { if self.is_blocking.get() { // Exit out of the run state return; @@ -296,12 +295,12 @@ impl Worker { /// /// Returns `true` if work was found. #[inline] - fn try_run_task(&self, notify: &Arc, sender: &mut Sender) -> bool { - if self.try_run_owned_task(notify, sender) { + fn try_run_task(&self, notify: &Arc) -> bool { + if self.try_run_owned_task(notify) { return true; } - self.try_steal_task(notify, sender) + self.try_steal_task(notify) } /// Checks the worker's current state, updating it as needed. @@ -383,13 +382,13 @@ impl Worker { /// Runs the next task on this worker's queue. /// /// Returns `true` if work was found. - fn try_run_owned_task(&self, notify: &Arc, sender: &mut Sender) -> bool { + fn try_run_owned_task(&self, notify: &Arc) -> bool { use deque::Pop; // Poll the internal queue for a task to run match self.entry().pop_task() { Pop::Data(task) => { - self.run_task(task, notify, sender); + self.run_task(task, notify); true } Pop::Empty => false, @@ -400,7 +399,7 @@ impl Worker { /// Tries to steal a task from another worker. /// /// Returns `true` if work was found - fn try_steal_task(&self, notify: &Arc, sender: &mut Sender) -> bool { + fn try_steal_task(&self, notify: &Arc) -> bool { use deque::Steal; debug_assert!(!self.is_blocking.get()); @@ -416,7 +415,7 @@ impl Worker { Steal::Data(task) => { trace!("stole task"); - self.run_task(task, notify, sender); + self.run_task(task, notify); trace!("try_steal_task -- signal_work; self={}; from={}", self.id.0, idx); @@ -446,10 +445,10 @@ impl Worker { found_work } - fn run_task(&self, task: Arc, notify: &Arc, sender: &mut Sender) { + fn run_task(&self, task: Arc, notify: &Arc) { use task::Run::*; - let run = self.run_task2(&task, notify, sender); + let run = self.run_task2(&task, notify); // TODO: Try to claim back the worker state in case the backup thread // did not start up fast enough. This is a performance optimization. @@ -512,8 +511,7 @@ impl Worker { /// function. fn run_task2(&self, task: &Arc, - notify: &Arc, - sender: &mut Sender) + notify: &Arc) -> task::Run { struct Guard<'a> { @@ -549,7 +547,7 @@ impl Worker { allocated_at_run: can_block == CanBlock::Allocated }; - task.run(notify, sender) + task.run(notify) } /// Drains all tasks on the extern queue and pushes them onto the internal diff --git a/tokio-threadpool/tests/threadpool.rs b/tokio-threadpool/tests/threadpool.rs index 9dcf2b71c..30dfea29a 100644 --- a/tokio-threadpool/tests/threadpool.rs +++ b/tokio-threadpool/tests/threadpool.rs @@ -3,27 +3,10 @@ extern crate tokio_executor; extern crate futures; extern crate env_logger; -#[cfg(feature = "unstable-futures")] -extern crate futures2; - use tokio_threadpool::*; - -#[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")] -fn lazy(f: F) -> Box + Send> where - F: Send + 'static + FnOnce() -> R, - R: Send + 'static + IntoFuture, - R::Future: Send, -{ - Box::new(::futures2::future::lazy(|_| f())) -} - use std::cell::Cell; use std::sync::{mpsc, Arc}; use std::sync::atomic::*; @@ -32,57 +15,10 @@ 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(); @@ -107,29 +43,29 @@ fn natural_shutdown_simple_futures() { .build() }; - let mut tx = pool.sender().clone(); + let tx = pool.sender().clone(); let a = { let (t, rx) = mpsc::channel(); - spawn_pool(&mut tx, lazy(move || { + tx.spawn(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(); - spawn_pool(&mut tx, lazy(move || { + tx.spawn(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 }; @@ -139,7 +75,7 @@ fn natural_shutdown_simple_futures() { assert_eq!("two", b.recv().unwrap()); // Wait for the pool to shutdown - await_shutdown(pool.shutdown()); + pool.shutdown().wait().unwrap(); // Assert that at least one thread started let num_inc = num_inc.load(Relaxed); @@ -163,7 +99,6 @@ fn force_shutdown_drops_futures() { struct Never(Arc); - #[cfg(not(feature = "unstable-futures"))] impl Future for Never { type Item = (); type Error = (); @@ -173,16 +108,6 @@ 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); @@ -201,10 +126,10 @@ fn force_shutdown_drops_futures() { .build(); let mut tx = pool.sender().clone(); - spawn_pool(&mut tx, Never(num_drop.clone())); + tx.spawn(Never(num_drop.clone())).unwrap(); // Wait for the pool to shutdown - await_shutdown(pool.shutdown_now()); + pool.shutdown_now().wait().unwrap(); // Assert that only a single thread was spawned. let a = num_inc.load(Relaxed); @@ -231,7 +156,6 @@ fn drop_threadpool_drops_futures() { struct Never(Arc); - #[cfg(not(feature = "unstable-futures"))] impl Future for Never { type Item = (); type Error = (); @@ -241,16 +165,6 @@ 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); @@ -271,7 +185,7 @@ fn drop_threadpool_drops_futures() { .build(); let mut tx = pool.sender().clone(); - spawn_pool(&mut tx, Never(num_drop.clone())); + tx.spawn(Never(num_drop.clone())).unwrap(); // Wait for the pool to shutdown drop(pool); @@ -309,13 +223,13 @@ fn thread_shutdown_timeout() { let _ = t.lock().unwrap().send(()); }) .build(); - let mut tx = pool.sender().clone(); + let tx = pool.sender().clone(); let t = complete_tx.clone(); - spawn_pool(&mut tx, lazy(move || { + tx.spawn(lazy(move || { t.send(()).unwrap(); Ok(()) - })); + })).unwrap(); // The future completes complete_rx.recv().unwrap(); @@ -324,14 +238,14 @@ fn thread_shutdown_timeout() { shutdown_rx.recv().unwrap(); // Futures can still be run - spawn_pool(&mut tx, lazy(move || { + tx.spawn(lazy(move || { complete_tx.send(()).unwrap(); Ok(()) - })); + })).unwrap(); complete_rx.recv().unwrap(); - await_shutdown(pool.shutdown()); + pool.shutdown().wait().unwrap(); } #[test] @@ -347,14 +261,14 @@ fn many_oneshot_futures() { for _ in 0..NUM { let cnt = cnt.clone(); - spawn_pool(&mut tx, lazy(move || { + tx.spawn(lazy(move || { cnt.fetch_add(1, Relaxed); Ok(()) - })); + })).unwrap(); } // Wait for the pool to shutdown - await_shutdown(pool.shutdown()); + pool.shutdown().wait().unwrap(); let num = cnt.load(Relaxed); assert_eq!(num, NUM); @@ -363,12 +277,8 @@ 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; @@ -392,11 +302,11 @@ fn many_multishot_futures() { .map_err(|e| panic!("{:?}", e)); // Forward all the messages - spawn_pool(&mut pool_tx, next_tx + pool_tx.spawn(next_tx .send_all(rx) .map(|_| ()) .map_err(|e| panic!("{:?}", e)) - ); + ).unwrap(); chain_rx = next_rx; } @@ -419,84 +329,73 @@ fn many_multishot_futures() { Ok(()) }) }); - spawn_pool(&mut pool_tx, ignore_results(task)); + pool_tx.spawn(ignore_results(task)).unwrap(); start_txs.push(start_tx); final_rxs.push(final_rx); } for start_tx in start_txs { - block_on(start_tx.send("ping")).unwrap(); + start_tx.send("ping").wait().unwrap(); } for final_rx in final_rxs { - {#![cfg(feature = "unstable-futures")] - block_on(final_rx.next()).unwrap(); - } - - {#![cfg(not(feature = "unstable-futures"))] - block_on(final_rx.into_future()).unwrap(); - } + final_rx.wait().next().unwrap().unwrap(); } // Shutdown the pool - await_shutdown(pool.shutdown()); + pool.shutdown().wait().unwrap(); } } #[test] fn global_executor_is_configured() { let pool = ThreadPool::new(); - let mut tx = pool.sender().clone(); + let tx = pool.sender().clone(); let (signal_tx, signal_rx) = mpsc::channel(); - spawn_pool(&mut tx, lazy(move || { - spawn_default(lazy(move || { + tx.spawn(lazy(move || { + tokio_executor::spawn(lazy(move || { signal_tx.send(()).unwrap(); Ok(()) })); Ok(()) - })); + })).unwrap(); signal_rx.recv().unwrap(); - await_shutdown(pool.shutdown()); + pool.shutdown().wait().unwrap(); } #[test] fn new_threadpool_is_idle() { let pool = ThreadPool::new(); - await_shutdown(pool.shutdown_on_idle()); + pool.shutdown_on_idle().wait().unwrap(); } #[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 pool = Builder::new() .pool_size(4) .max_blocking(2) .build(); - let mut tx = pool.sender().clone(); + let tx = pool.sender().clone(); let (term_tx, term_rx) = oneshot::channel(); - spawn_pool(&mut tx, term_rx.then(|_| { + tx.spawn(term_rx.then(|_| { Ok(()) - })); + })).unwrap(); let mut idle = pool.shutdown_on_idle(); struct IdleFut<'a>(&'a mut Shutdown); - #[cfg(not(feature = "unstable-futures"))] impl<'a> Future for IdleFut<'a> { type Item = (); type Error = (); @@ -506,31 +405,20 @@ fn busy_threadpool_is_not_idle() { } } - #[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(); + IdleFut(&mut idle).wait().unwrap(); term_tx.send(()).unwrap(); - await_shutdown(idle); + idle.wait().unwrap(); } #[test] fn panic_in_task() { let pool = ThreadPool::new(); - let mut tx = pool.sender().clone(); + let tx = pool.sender().clone(); struct Boom; - #[cfg(not(feature = "unstable-futures"))] impl Future for Boom { type Item = (); type Error = (); @@ -540,25 +428,15 @@ 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()); } } - spawn_pool(&mut tx, Boom); + tx.spawn(Boom).unwrap(); - await_shutdown(pool.shutdown_on_idle()); + pool.shutdown_on_idle().wait().unwrap(); } #[test] diff --git a/tokio-udp/src/lib.rs b/tokio-udp/src/lib.rs index 4a37b697e..ebb6727d6 100644 --- a/tokio-udp/src/lib.rs +++ b/tokio-udp/src/lib.rs @@ -29,9 +29,6 @@ extern crate tokio_codec; extern crate tokio_io; extern crate tokio_reactor; -#[cfg(feature = "unstable-futures")] -extern crate futures2; - mod frame; mod socket; mod send_dgram;