diff --git a/src/event_loop.rs b/src/event_loop.rs index 522e96b8f..9e6c750be 100644 --- a/src/event_loop.rs +++ b/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 { + 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 { + 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. /// diff --git a/src/timeout.rs b/src/timeout.rs index d71a51e05..582a71fc6 100644 --- a/src/timeout.rs +++ b/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 { 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 } diff --git a/src/timer_wheel.rs b/src/timer_wheel.rs index a1fe66951..21250f334 100644 --- a/src/timer_wheel.rs +++ b/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 TimerWheel { /// /// 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 TimerWheel { 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 TimerWheel { // 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 TimerWheel { .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 TimerWheel { } } +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::::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::::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::::new(); - let now = Instant::now(); + let now = timer.start; timer.insert(now + ms(200), 2); assert_eq!(timer.poll(now + ms(100)), None);