Update futures-curl

Also involved yet another round of bug fixes to the timer wheel as well as an
unfortunately serious rejiggering of the level-translation into libcurl. Eew.

src/event_loop.rs

@@ -237,6 +237,7 @@ impl Loop {
                 }
             }
             debug!("loop poll - {:?}", start.elapsed());
+            debug!("loop time - {:?}", Instant::now());
 
             // First up, process all timeouts that may have just occurred.
             let start = Instant::now();
@@ -314,10 +315,12 @@ impl Loop {
     /// Note that this should be used instead fo `handle.unpark()` to ensure
     /// that the `CURRENT_LOOP` variable is set appropriately.
     fn notify_handle(&self, handle: TaskHandle) {
+        debug!("notifying a task handle");
         CURRENT_LOOP.set(&self, || handle.unpark());
     }
 
     fn add_source(&self, source: IoSource) -> io::Result<usize> {
+        debug!("adding a new I/O source");
         let sched = Scheduled {
             source: source,
             reader: None,
@@ -334,11 +337,13 @@ impl Loop {
     }
 
     fn drop_source(&self, token: usize) {
+        debug!("dropping I/O source: {}", token);
         let sched = self.dispatch.borrow_mut().remove(token).unwrap();
         deregister(&self.io, &sched);
     }
 
     fn schedule(&self, token: usize, wake: TaskHandle, dir: Direction) {
+        debug!("scheduling direction for: {}", token);
         let to_call = {
             let mut dispatch = self.dispatch.borrow_mut();
             let sched = dispatch.get_mut(token).unwrap();
@@ -370,11 +375,17 @@ impl Loop {
         }
         let entry = timeouts.vacant_entry().unwrap();
         let timeout = self.timer_wheel.borrow_mut().insert(at, entry.index());
+        let when = *timeout.when();
         let entry = entry.insert((timeout, TimeoutState::NotFired));
-        Ok(TimeoutToken { token: entry.index() })
+        debug!("added a timeout: {}", entry.index());
+        Ok(TimeoutToken {
+            token: entry.index(),
+            when: when,
+        })
     }
 
     fn update_timeout(&self, token: &TimeoutToken, handle: TaskHandle) {
+        debug!("updating a timeout: {}", token.token);
         let to_wake = self.timeouts.borrow_mut()[token.token].1.block(handle);
         if let Some(to_wake) = to_wake {
             self.notify_handle(to_wake);
@@ -382,6 +393,7 @@ impl Loop {
     }
 
     fn cancel_timeout(&self, token: &TimeoutToken) {
+        debug!("cancel a timeout: {}", token.token);
         let pair = self.timeouts.borrow_mut().remove(token.token);
         if let Some((timeout, _state)) = pair {
             self.timer_wheel.borrow_mut().cancel(&timeout);
@@ -412,8 +424,14 @@ impl Loop {
             }
             Message::UpdateTimeout(t, handle) => self.update_timeout(&t, handle),
             Message::CancelTimeout(t) => self.cancel_timeout(&t),
-            Message::Run(f) => f.call(),
-            Message::Drop(data) => drop(data),
+            Message::Run(f) => {
+                debug!("running a closure");
+                f.call()
+            }
+            Message::Drop(data) => {
+                debug!("dropping some data");
+                drop(data);
+            }
         }
     }
 }
@@ -573,7 +591,7 @@ impl LoopHandle {
     /// This method will panic if the timeout specified was not created by this
     /// loop handle's `add_timeout` method.
     pub fn update_timeout(&self, timeout: &TimeoutToken) {
-        let timeout = TimeoutToken { token: timeout.token };
+        let timeout = TimeoutToken { token: timeout.token, when: timeout.when };
         self.send(Message::UpdateTimeout(timeout, task::park()))
     }
 
@@ -584,7 +602,7 @@ impl LoopHandle {
     /// This method will panic if the timeout specified was not created by this
     /// loop handle's `add_timeout` method.
     pub fn cancel_timeout(&self, timeout: &TimeoutToken) {
-        let timeout = TimeoutToken { token: timeout.token };
+        let timeout = TimeoutToken { token: timeout.token, when: timeout.when };
         self.send(Message::CancelTimeout(timeout))
     }
 
@@ -656,6 +674,11 @@ impl LoopPin {
     pub fn handle(&self) -> &LoopHandle {
         &self.handle
     }
+
+    /// TODO: dox
+    pub fn executor(&self) -> Arc<Executor> {
+        self.handle.tx.clone()
+    }
 }
 
 /// A future which will resolve a unique `tok` token for an I/O object.
@@ -684,6 +707,7 @@ pub struct AddTimeout {
 /// A token that identifies an active timeout.
 pub struct TimeoutToken {
     token: usize,
+    when: Instant,
 }
 
 impl Future for AddTimeout {
@@ -695,6 +719,18 @@ impl Future for AddTimeout {
     }
 }
 
+impl TimeoutToken {
+    /// Returns the instant in time when this timeout token will "fire".
+    ///
+    /// Note that this instant may *not* be the instant that was passed in when
+    /// the timeout was created. The event loop does not support high resolution
+    /// timers, so the exact resolution of when a timeout may fire may be
+    /// slightly fudged.
+    pub fn when(&self) -> &Instant {
+        &self.when
+    }
+}
+
 /// A handle to data that is owned by an event loop thread, and is only
 /// accessible on that thread itself.
 ///

src/timeout.rs

@@ -15,7 +15,6 @@ use event_loop::TimeoutToken;
 /// they will likely fire some granularity after the exact instant that they're
 /// otherwise indicated to fire at.
 pub struct Timeout {
-    at: Instant,
     token: TimeoutToken,
     handle: LoopHandle,
 }
@@ -38,7 +37,6 @@ impl LoopHandle {
     pub fn timeout_at(self, at: Instant) -> IoFuture<Timeout> {
         self.add_timeout(at).map(move |token| {
             Timeout {
-                at: at,
                 token: token,
                 handle: self,
             }
@@ -52,9 +50,12 @@ impl Future for Timeout {
 
     fn poll(&mut self) -> Poll<(), io::Error> {
         // TODO: is this fast enough?
-        if self.at <= Instant::now() {
+        let now = Instant::now();
+        if *self.token.when() <= now {
             Poll::Ok(())
         } else {
+            trace!("waiting for a timeout at {:?}", self.token.when());
+            trace!("current time is          {:?}", now);
             self.handle.update_timeout(&self.token);
             Poll::NotReady
         }

src/timer_wheel.rs

@@ -1,6 +1,5 @@
 //! A timer wheel implementation
 
-use std::cmp;
 use std::mem;
 use std::time::{Instant, Duration};
 
@@ -110,7 +109,7 @@ impl<T> TimerWheel<T> {
     ///
     /// This method will panic if `at` is before the time that this timer wheel
     /// was created.
-    pub fn insert(&mut self, at: Instant, data: T) -> Timeout {
+    pub fn insert(&mut self, mut at: Instant, data: T) -> Timeout {
         // First up, figure out where we're gonna go in the wheel. Note that if
         // we're being scheduled on or before the current wheel tick we just
         // make sure to defer ourselves to the next tick.
@@ -122,6 +121,12 @@ impl<T> TimerWheel<T> {
         let wheel_idx = self.ticks_to_wheel_idx(tick);
         trace!("inserting timeout at {} for {}", wheel_idx, tick);
 
+        let actual_tick = self.start +
+                          Duration::from_millis(TICK_MS) * (tick as u32);
+        trace!("actual_tick: {:?}", actual_tick);
+        trace!("at:          {:?}", at);
+        at = actual_tick;
+
         // Next, make sure there's enough space in the slab for the timeout.
         if self.slab.vacant_entry().is_none() {
             let amt = self.slab.count();
@@ -150,12 +155,8 @@ impl<T> TimerWheel<T> {
 
         // Update the wheel slot's next timeout field.
         if at <= slot.next_timeout.unwrap_or(at) {
-            let tick = tick as u32;
-            let actual_tick = self.start + Duration::from_millis(TICK_MS) * tick;
-            trace!("actual_tick: {:?}", actual_tick);
-            trace!("at:          {:?}", at);
-            let at = cmp::max(actual_tick, at);
             debug!("updating[{}] next timeout: {:?}", wheel_idx, at);
+            debug!("                    start: {:?}", self.start);
             slot.next_timeout = Some(at);
         }
 
@@ -300,7 +301,7 @@ impl<T> TimerWheel<T> {
                     .checked_mul(1_000)
                     .and_then(|m| m.checked_add(ms))
                     .expect("overflow scheduling timeout");
-        (ms + TICK_MS / 2) / TICK_MS
+        ms / TICK_MS
     }
 
     fn ticks_to_wheel_idx(&self, ticks: u64) -> usize {
@@ -308,6 +309,12 @@ impl<T> TimerWheel<T> {
     }
 }
 
+impl Timeout {
+    pub fn when(&self) -> &Instant {
+        &self.when
+    }
+}
+
 #[cfg(test)]
 mod tests {
     extern crate env_logger;
@@ -401,7 +408,7 @@ mod tests {
     fn next_timeout() {
         drop(env_logger::init());
         let mut timer = TimerWheel::<i32>::new();
-        let now = Instant::now();
+        let now = timer.start;
 
         assert!(timer.next_timeout().is_none());
         timer.insert(now + ms(400), 3);
@@ -423,15 +430,23 @@ mod tests {
         timer.insert(now + ms(200), 4);
         timer.insert(now + ms(201), 5);
         timer.insert(now + ms(251), 6);
+        timer.insert(now + ms(299), 7);
+        timer.insert(now + ms(300), 8);
+        timer.insert(now + ms(301), 9);
 
         let mut found = Vec::new();
         while let Some(i) = timer.poll(now + ms(200)) {
             found.push(i);
         }
         found.sort();
-        assert_eq!(found, [3, 4, 5]);
+        assert_eq!(found, [3, 4, 5, 6, 7]);
 
-        assert_eq!(timer.poll(now + ms(300)), Some(6));
+        let mut found = Vec::new();
+        while let Some(i) = timer.poll(now + ms(300)) {
+            found.push(i);
+        }
+        found.sort();
+        assert_eq!(found, [8, 9]);
         assert_eq!(timer.poll(now + ms(300)), None);
     }
 
@@ -439,10 +454,10 @@ mod tests {
     fn remove_clears_timeout() {
         drop(env_logger::init());
         let mut timer = TimerWheel::<i32>::new();
-        let now = Instant::now();
+        let now = timer.start;
 
         timer.insert(now + ms(100), 3);
-        assert_eq!(timer.next_timeout(), Some(now + ms(100)));
+        assert_eq!(timer.next_timeout(), Some(timer.start + ms(100)));
         assert_eq!(timer.poll(now + ms(200)), Some(3));
         assert_eq!(timer.next_timeout(), None);
     }
@@ -475,7 +490,7 @@ mod tests {
     fn poll_then_next_timeout() {
         drop(env_logger::init());
         let mut timer = TimerWheel::<i32>::new();
-        let now = Instant::now();
+        let now = timer.start;
 
         timer.insert(now + ms(200), 2);
         assert_eq!(timer.poll(now + ms(100)), None);