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

Remove dead futures2 code. (#538)

Browse Source 
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.
This commit is contained in:
Carl Lerche
2018-08-09 21:56:53 -07:00
committed by GitHub
parent 96b556fbff
commit d91c775f36
31 changed files with 67 additions and 1380 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tokio-threadpool/src/builder.rs
-3
View File
@@ -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
tokio-threadpool/src/futures2_wake.rs
-60
View File
@@ -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<Notifier>,
id: usize,
}
impl Futures2Wake {
pub(crate) fn new(id: usize, inner: &Arc<Pool>) -> 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<Futures2Wake>);
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<Futures2Wake>)).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<Futures2Wake>;
(*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<Futures2Wake>;
(*me).notifier.notify((*me).id)
}
}
pub(crate) fn into_waker(rc: Arc<Futures2Wake>) -> futures2::task::Waker {
unsafe {
let ptr = mem::transmute::<Arc<Futures2Wake>, *mut ArcWrapped>(rc);
futures2::task::Waker::new(ptr)
}
}
tokio-threadpool/src/lib.rs
-5
View File
@@ -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;
tokio-threadpool/src/pool/mod.rs
+1 -3
View File
@@ -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,
};
tokio-threadpool/src/sender.rs
-55
View File
@@ -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<T> future::Executor<T> for Sender
@@ -200,47 +186,6 @@ where T: Future<Item = (), Error = ()> + Send + 'static,
}
}
#[cfg(feature = "unstable-futures")]
type Task2 = Box<futures2::Future<Item = (), Error = futures2::Never> + 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 {
tokio-threadpool/src/shutdown.rs
+1 -21
View File
@@ -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())
}
}
tokio-threadpool/src/shutdown_task.rs
+2 -14
View File
@@ -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();
}
}
tokio-threadpool/src/task/mod.rs
+7 -66
View File
@@ -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<Option<TaskFuture>>,
future: UnsafeCell<Option<Spawn<BoxFuture>>>,
}
#[derive(Debug)]
@@ -56,27 +52,13 @@ pub(crate) enum Run {
type BoxFuture = Box<Future<Item = (), Error = ()> + Send + 'static>;
#[cfg(feature = "unstable-futures")]
type BoxFuture2 = Box<futures2::Future<Item = (), Error = futures2::Never> + Send>;
enum TaskFuture {
Futures1(Spawn<BoxFuture>),
#[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<F>(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<Notifier>, exec: &mut Sender) -> Run {
pub fn run(&self, unpark: &Arc<Notifier>) -> 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<TaskFuture>, bool);
struct Guard<'a>(&'a mut Option<Spawn<BoxFuture>>, 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<Notifier>, 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)),
}
}
}
}
}
tokio-threadpool/src/worker/mod.rs
+12 -14
View File
@@ -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(&notify, &mut sender) {
if self.try_run_task(&notify) {
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<Notifier>, sender: &mut Sender) -> bool {
if self.try_run_owned_task(notify, sender) {
fn try_run_task(&self, notify: &Arc<Notifier>) -> 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<Notifier>, sender: &mut Sender) -> bool {
fn try_run_owned_task(&self, notify: &Arc<Notifier>) -> 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<Notifier>, sender: &mut Sender) -> bool {
fn try_steal_task(&self, notify: &Arc<Notifier>) -> 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<Task>, notify: &Arc<Notifier>, sender: &mut Sender) {
fn run_task(&self, task: Arc<Task>, notify: &Arc<Notifier>) {
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<Task>,
notify: &Arc<Notifier>,
sender: &mut Sender)
notify: &Arc<Notifier>)
-> 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
tokio-threadpool/tests/threadpool.rs
+38 -160
View File
@@ -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<R, F>(f: F) -> Box<Future<Item = R::Item, Error = R::Error> + 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<u32> = Cell::new(0));
#[cfg(not(feature = "unstable-futures"))]
fn spawn_pool<F>(pool: &mut Sender, f: F)
where F: Future<Item = (), Error = ()> + Send + 'static
{
pool.spawn(f).unwrap()
}
#[cfg(feature = "unstable-futures")]
fn spawn_pool<F>(pool: &mut Sender, f: F)
where F: Future<Item = (), Error = ()> + Send + 'static
{
futures2::executor::Executor::spawn(
pool,
Box::new(f.map_err(|_| panic!()))
).unwrap()
}
#[cfg(not(feature = "unstable-futures"))]
fn spawn_default<F>(f: F)
where F: Future<Item = (), Error = ()> + Send + 'static
{
tokio_executor::spawn(f)
}
#[cfg(feature = "unstable-futures")]
fn spawn_default<F>(f: F)
where F: Future<Item = (), Error = ()> + Send + 'static
{
tokio_executor::spawn2(Box::new(f.map_err(|_| panic!())))
}
fn ignore_results<F: Future + Send + 'static>(f: F) -> Box<Future<Item = (), Error = ()> + 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: Future>(f: F) -> Result<F::Item, F::Error> {
f.wait()
}
#[cfg(feature = "unstable-futures")]
fn block_on<F: Future>(f: F) -> Result<F::Item, F::Error> {
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<AtomicUsize>);
#[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<AtomicUsize>);
#[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]