Merge pull request #5 from srijs/futures-0.2

convert to futures 0.2 (alpha)

Author: alexcrichton

Cargo.toml

@@ -12,4 +12,4 @@ Timeouts and intervals for futures.
 """
 
 [dependencies]
-futures = "0.1.15"
+futures = "0.2.0-alpha"

src/delay.rs

@@ -10,7 +10,7 @@ use std::sync::atomic::Ordering::SeqCst;
 use std::time::{Duration, Instant};
 
 use futures::{Future, Poll, Async};
-use futures::task::AtomicTask;
+use futures::task::{self, AtomicWaker};
 
 use arc_list::Node;
 use {TimerHandle, ScheduledTimer};
@@ -57,7 +57,7 @@ impl Delay {
         let state = Arc::new(Node::new(ScheduledTimer {
             at: Mutex::new(Some(at)),
             state: AtomicUsize::new(0),
-            task: AtomicTask::new(),
+            waker: AtomicWaker::new(),
             inner: handle.inner,
             slot: Mutex::new(None),
         }));
@@ -69,7 +69,7 @@ impl Delay {
             return Delay { state: None, when: at }
         }
 
-        inner.task.notify();
+        inner.waker.wake();
         Delay {
             state: Some(state),
             when: at,
@@ -126,7 +126,7 @@ impl Delay {
             // If we fail to push our node then we've become an inert timer, so
             // we'll want to clear our `state` field accordingly
             timeouts.list.push(state)?;
-            timeouts.task.notify();
+            timeouts.waker.wake();
         }
 
         Ok(())
@@ -141,7 +141,7 @@ impl Future for Delay {
     type Item = ();
     type Error = io::Error;
 
-    fn poll(&mut self) -> Poll<(), io::Error> {
+    fn poll(&mut self, cx: &mut task::Context) -> Poll<(), io::Error> {
         let state = match self.state {
             Some(ref state) => state,
             None => return Err(io::Error::new(io::ErrorKind::Other,
@@ -151,7 +151,7 @@ impl Future for Delay {
             return Ok(Async::Ready(()))
         }
 
-        state.task.register();
+        state.waker.register(cx.waker());
 
         // Now that we've registered, do the full check of our own internal
         // state. If we've fired the first bit is set, and if we've been
@@ -160,7 +160,7 @@ impl Future for Delay {
             n if n & 0b01 != 0 => Ok(Async::Ready(())),
             n if n & 0b10 != 0 => Err(io::Error::new(io::ErrorKind::Other,
                                                      "timer has gone away")),
-            _ => Ok(Async::NotReady),
+            _ => Ok(Async::Pending),
         }
     }
 }
@@ -174,7 +174,7 @@ impl Drop for Delay {
         if let Some(timeouts) = state.inner.upgrade() {
             *state.at.lock().unwrap() = None;
             if timeouts.list.push(state).is_ok() {
-                timeouts.task.notify();
+                timeouts.waker.wake();
             }
         }
     }

src/ext.rs

@@ -30,6 +30,7 @@ pub trait FutureExt: Future + Sized {
     ///
     /// use std::time::Duration;
     /// use futures::prelude::*;
+    /// use futures::executor::block_on;
     /// use futures_timer::FutureExt;
     ///
     /// # fn long_future() -> futures::future::FutureResult<(), std::io::Error> {
@@ -40,7 +41,7 @@ pub trait FutureExt: Future + Sized {
     ///     let future = long_future();
     ///     let timed_out = future.timeout(Duration::from_secs(1));
     ///
-    ///     match timed_out.wait() {
+    ///     match block_on(timed_out) {
     ///         Ok(item) => println!("got {:?} within enough time!", item),
     ///         Err(_) => println!("took too long to produce the item"),
     ///     }
@@ -86,16 +87,16 @@ impl<F> Future for Timeout<F>
     type Item = F::Item;
     type Error = F::Error;
 
-    fn poll(&mut self) -> Poll<F::Item, F::Error> {
-        match self.future.poll()? {
-            Async::NotReady => {}
+    fn poll(&mut self, cx: &mut task::Context) -> Poll<F::Item, F::Error> {
+        match self.future.poll(cx)? {
+            Async::Pending => {}
             other => return Ok(other)
         }
 
-        if self.timeout.poll()?.is_ready() {
+        if self.timeout.poll(cx)?.is_ready() {
             Err(io::Error::new(io::ErrorKind::TimedOut, "future timed out").into())
         } else {
-            Ok(Async::NotReady)
+            Ok(Async::Pending)
         }
     }
 }
@@ -142,20 +143,20 @@ impl<S> Stream for TimeoutStream<S>
     type Item = S::Item;
     type Error = S::Error;
 
-    fn poll(&mut self) -> Poll<Option<S::Item>, S::Error> {
-        match self.stream.poll() {
-            Ok(Async::NotReady) => {}
+    fn poll_next(&mut self, cx: &mut task::Context) -> Poll<Option<S::Item>, S::Error> {
+        match self.stream.poll_next(cx) {
+            Ok(Async::Pending) => {}
             other => {
                 self.timeout.reset(self.dur);
                 return other
             }
         }
 
-        if self.timeout.poll()?.is_ready() {
+        if self.timeout.poll(cx)?.is_ready() {
             self.timeout.reset(self.dur);
             Err(io::Error::new(io::ErrorKind::TimedOut, "stream item timed out").into())
         } else {
-            Ok(Async::NotReady)
+            Ok(Async::Pending)
         }
     }
 }

src/global.rs

@@ -4,7 +4,8 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::thread;
 use std::time::Instant;
 
-use futures::executor::{spawn, Notify};
+use futures::prelude::*;
+use futures::executor::{SpawnError, Executor};
 
 use {TimerHandle, Timer};
 
@@ -50,15 +51,18 @@ impl Drop for HelperThread {
     }
 }
 
-fn run(timer: Timer, done: Arc<AtomicBool>) {
-    let mut timer = spawn(timer);
+fn run(mut timer: Timer, done: Arc<AtomicBool>) {
 	let me = Arc::new(ThreadUnpark {
 		thread: thread::current(),
 	});
+    let mut local_map = task::LocalMap::new();
+    let waker = task::Waker::from(me);
+    let mut exec = NonFunctionalExecutor;
+    let mut cx = task::Context::new(&mut local_map, &waker, &mut exec);
     while !done.load(Ordering::SeqCst) {
-        drop(timer.poll_future_notify(&me, 0));
-        timer.get_mut().advance();
-        match timer.get_mut().next_event() {
+        drop(timer.poll(&mut cx));
+        timer.advance();
+        match timer.next_event() {
             // Ok, block for the specified time
             Some(when) => {
                 let now = Instant::now();
@@ -79,8 +83,16 @@ struct ThreadUnpark {
     thread: thread::Thread,
 }
 
-impl Notify for ThreadUnpark {
-    fn notify(&self, _unpark_id: usize) {
-        self.thread.unpark()
+impl task::Wake for ThreadUnpark {
+    fn wake(arc_self: &Arc<Self>) {
+        arc_self.thread.unpark()
+    }
+}
+
+struct NonFunctionalExecutor;
+
+impl Executor for NonFunctionalExecutor {
+    fn spawn(&mut self, _: Box<Future<Item = (), Error = Never> + 'static + Send>) -> Result<(), SpawnError> {
+        Err(SpawnError::shutdown())
     }
 }

src/interval.rs

@@ -58,9 +58,9 @@ impl Stream for Interval {
     type Item = ();
     type Error = io::Error;
 
-    fn poll(&mut self) -> Poll<Option<()>, io::Error> {
-        if self.delay.poll()?.is_not_ready() {
-            return Ok(Async::NotReady)
+    fn poll_next(&mut self, cx: &mut task::Context) -> Poll<Option<()>, io::Error> {
+        if self.delay.poll(cx)?.is_pending() {
+            return Ok(Async::Pending)
         }
         let next = next_interval(delay::fires_at(&self.delay),
                                  Instant::now(),

src/lib.rs

@@ -75,8 +75,8 @@ use std::sync::atomic::{AtomicUsize, ATOMIC_USIZE_INIT};
 use std::sync::{Arc, Weak, Mutex};
 use std::time::Instant;
 
-use futures::task::AtomicTask;
-use futures::{Future, Async, Poll};
+use futures::task::AtomicWaker;
+use futures::{Future, Async, Poll, task};
 
 use arc_list::{ArcList, Node};
 use heap::{Heap, Slot};
@@ -129,12 +129,12 @@ struct Inner {
     list: ArcList<ScheduledTimer>,
 
     /// The blocked `Timer` task to receive notifications to the `list` above.
-    task: AtomicTask,
+    waker: AtomicWaker,
 }
 
 /// Shared state between the `Timer` and a `Delay`.
 struct ScheduledTimer {
-    task: AtomicTask,
+    waker: AtomicWaker,
 
     // The lowest bit here is whether the timer has fired or not, the second
     // lowest bit is whether the timer has been invalidated, and all the other
@@ -164,7 +164,7 @@ impl Timer {
         Timer {
             inner: Arc::new(Inner {
                 list: ArcList::new(),
-                task: AtomicTask::new(),
+                waker: AtomicWaker::new(),
             }),
             timer_heap: Heap::new(),
         }
@@ -210,7 +210,7 @@ impl Timer {
             *heap_timer.node.slot.lock().unwrap() = None;
             let bits = heap_timer.gen << 2;
             match heap_timer.node.state.compare_exchange(bits, bits | 0b01, SeqCst, SeqCst) {
-                Ok(_) => heap_timer.node.task.notify(),
+                Ok(_) => heap_timer.node.waker.wake(),
                 Err(_b) => {}
             }
         }
@@ -249,16 +249,16 @@ impl Timer {
 
     fn invalidate(&mut self, node: Arc<Node<ScheduledTimer>>) {
         node.state.fetch_or(0b10, SeqCst);
-        node.task.notify();
+        node.waker.wake();
     }
 }
 
 impl Future for Timer {
     type Item = ();
     type Error = ();
 
-    fn poll(&mut self) -> Poll<(), ()> {
-        self.inner.task.register();
+    fn poll(&mut self, cx: &mut task::Context) -> Poll<(), ()> {
+        self.inner.waker.register(cx.waker());
         let mut list = self.inner.list.take();
         while let Some(node) = list.pop() {
             let at = *node.at.lock().unwrap();
@@ -267,7 +267,7 @@ impl Future for Timer {
                 None => self.remove(node),
             }
         }
-        Ok(Async::NotReady)
+        Ok(Async::Pending)
     }
 }
 

tests/interval.rs

@@ -4,14 +4,15 @@ extern crate futures_timer;
 use std::time::{Instant, Duration};
 
 use futures::prelude::*;
+use futures::executor::block_on;
 use futures_timer::Interval;
 
 #[test]
 fn single() {
     let dur = Duration::from_millis(10);
     let start = Instant::now();
     let interval = Interval::new(dur);
-    interval.take(1).collect().wait().unwrap();
+    block_on(interval.take(1).collect()).unwrap();
     assert!(start.elapsed() >= dur);
 }
 
@@ -20,7 +21,7 @@ fn two_times() {
     let dur = Duration::from_millis(10);
     let start = Instant::now();
     let interval = Interval::new(dur);
-    let result = interval.take(2).collect().wait().unwrap();
+    let result = block_on(interval.take(2).collect()).unwrap();
     assert!(start.elapsed() >= dur*2);
     assert_eq!(result, vec![(), ()]);
 }

tests/smoke.rs

@@ -5,6 +5,7 @@ use std::time::{Instant, Duration};
 
 use futures::future;
 use futures::prelude::*;
+use futures::executor::block_on;
 use futures_timer::{Timer, Delay};
 
 fn far_future() -> Instant {
@@ -16,7 +17,7 @@ fn works() {
     let i = Instant::now();
     let dur = Duration::from_millis(100);
     let d = Delay::new(dur);
-    d.wait().unwrap();
+    block_on(d).unwrap();
     assert!(i.elapsed() > dur);
 }
 
@@ -25,7 +26,7 @@ fn error_after_inert() {
     let t = Timer::new();
     let handle = t.handle();
     drop(t);
-    assert!(Delay::new_handle(far_future(), handle).poll().is_err());
+    assert!(block_on(Delay::new_handle(far_future(), handle)).is_err());
 }
 
 #[test]
@@ -34,20 +35,20 @@ fn drop_makes_inert() {
     let handle = t.handle();
     let timeout = Delay::new_handle(far_future(), handle);
     drop(t);
-    assert!(timeout.wait().is_err());
+    assert!(block_on(timeout).is_err());
 }
 
 #[test]
 fn reset() {
     let i = Instant::now();
     let dur = Duration::from_millis(100);
     let mut d = Delay::new(dur);
-    future::poll_fn(|| d.poll()).wait().unwrap();
+    block_on(future::poll_fn(|cx| d.poll(cx))).unwrap();
     assert!(i.elapsed() > dur);
 
     let i = Instant::now();
     d.reset(dur);
-    future::poll_fn(|| d.poll()).wait().unwrap();
+    block_on(future::poll_fn(|cx| d.poll(cx))).unwrap();
     assert!(i.elapsed() > dur);
 }
 
@@ -57,12 +58,12 @@ fn drop_timer_wakes() {
     let handle = t.handle();
     let mut timeout = Delay::new_handle(far_future(), handle);
     let mut t = Some(t);
-    assert!(future::poll_fn(|| {
-        match timeout.poll() {
-            Ok(Async::NotReady) => {}
+    assert!(block_on(future::poll_fn(|cx| {
+        match timeout.poll(cx) {
+            Ok(Async::Pending) => {}
             other => return other,
         }
         drop(t.take());
-        Ok(Async::NotReady)
-    }).wait().is_err());
+        Ok(Async::Pending)
+    })).is_err());
 }