Another pass at executors

Author: alexcrichton

futures-cpupool/src/lib.rs

@@ -49,7 +49,7 @@ use std::thread;
 
 use crossbeam::sync::MsQueue;
 use futures::{Future, oneshot, Oneshot, Complete, Poll};
-use futures::task::{Task, Run, Executor};
+use futures::task::{self, Run, Executor};
 
 /// A thread pool intended to run CPU intensive work.
 ///
@@ -162,7 +162,7 @@ impl CpuPool {
             fut: AssertUnwindSafe(f).catch_unwind(),
             tx: Some(tx),
         };
-        Task::new(self.inner.clone(), sender.boxed()).unpark();
+        task::spawn(sender.boxed()).execute(self.inner.clone());
         CpuFuture { inner: rx }
     }
 }

src/lib.rs

@@ -150,7 +150,7 @@
 //! [README]: https://github.com/alexcrichton/futures-rs#futures-rs
 
 #![no_std]
-#![deny(missing_docs)]
+// #![deny(missing_docs)]
 
 #[macro_use]
 #[cfg(feature = "use_std")]
@@ -373,19 +373,10 @@ pub trait Future {
     /// This function does not attempt to catch panics. If the `poll` function
     /// panics, panics will be propagated to the caller.
     #[cfg(feature = "use_std")]
-    fn wait(mut self) -> Result<Self::Item, Self::Error>
+    fn wait(self) -> Result<Self::Item, Self::Error>
         where Self: Sized
     {
-        use std::thread;
-
-        let task = task::ThreadTask::new();
-        loop {
-            match task.enter(|| self.poll()) {
-                Poll::Ok(e) => return Ok(e),
-                Poll::Err(e) => return Err(e),
-                Poll::NotReady => thread::park(),
-            }
-        }
+        task::spawn(self).wait_future()
     }
 
     /// Convenience function for turning this future into a trait object.

src/stream/wait.rs

@@ -1,8 +1,5 @@
-use std::thread;
-
-use Poll;
 use stream::Stream;
-use task::ThreadTask;
+use task;
 
 /// A stream combinator which converts an asynchronous stream to a **blocking
 /// iterator**.
@@ -11,29 +8,19 @@ use task::ThreadTask;
 /// into a standard iterator. This is implemented by blocking the current thread
 /// while items on the underlying stream aren't ready yet.
 pub struct Wait<S> {
-    task: ThreadTask,
-    stream: S,
+    stream: task::Spawn<S>,
 }
 
 pub fn new<S: Stream>(s: S) -> Wait<S> {
     Wait {
-        task: ThreadTask::new(),
-        stream: s,
+        stream: task::spawn(s),
     }
 }
 
 impl<S: Stream> Iterator for Wait<S> {
     type Item = Result<S::Item, S::Error>;
 
     fn next(&mut self) -> Option<Self::Item> {
-        let stream = &mut self.stream;
-        loop {
-            match self.task.enter(|| stream.poll()) {
-                Poll::Ok(Some(e)) => return Some(Ok(e)),
-                Poll::Ok(None) => return None,
-                Poll::Err(e) => return Some(Err(e)),
-                Poll::NotReady => thread::park(),
-            }
-        }
+        self.stream.wait_stream()
     }
 }