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

Simultaneous futures compat (#172)

Browse Source 
This patch adds opt-in support for futures 0.2.
This commit is contained in:
Aaron Turon
2018-03-13 13:57:35 -07:00
committed by Carl Lerche
parent 5846b3fc2a
commit d304791c0e
27 changed files with 1045 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tokio-threadpool/src/lib.rs
+171 -11
View File
@@ -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<Inner>,
}
#[cfg(feature = "unstable-futures")]
struct Futures2Wake {
notifier: Arc<Notifier>,
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<T> future::Executor<T> for Sender
@@ -827,6 +858,48 @@ 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 {
@@ -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(&notify) {
if self.try_run_task(&notify, &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(&notify) {
if self.try_steal_task(&notify, &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<Notifier>) -> bool {
fn try_run_task(&self, notify: &Arc<Notifier>, 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<Notifier>) -> bool {
fn try_steal_task(&self, notify: &Arc<Notifier>, 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<Notifier>) {
fn run_task(&self, task: Task, notify: &Arc<Notifier>, 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<Inner>) -> 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<Futures2Wake>);
#[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<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)
}
}
#[cfg(feature = "unstable-futures")]
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/task.rs
+67 -10
View File
@@ -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<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,
}
}
struct Inner {
// Next pointer in the queue that submits tasks to a worker.
next: AtomicPtr<Inner>,
@@ -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<Spawn<BoxFuture>>,
future: Option<TaskFuture>,
}
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
@@ -64,23 +83,41 @@ enum State {
Complete,
}
type BoxFuture = Box<Future<Item = (), Error = ()> + 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<F>(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<Notifier>) -> Run {
pub fn run(&self, unpark: &Arc<Notifier>, 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<Spawn<BoxFuture>>, bool);
struct Guard<'a>(&'a mut Option<TaskFuture>, 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<State> for usize {
}
}
}
// ===== 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)),
}
}
}
}
}