diff --git a/src/event_loop.rs b/src/event_loop.rs index f6fa66bc3..b41ebc47e 100644 --- a/src/event_loop.rs +++ b/src/event_loop.rs @@ -1,17 +1,19 @@ use std::cell::{Cell, RefCell}; use std::io::{self, ErrorKind}; +use std::mem; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, ATOMIC_USIZE_INIT, Ordering}; use std::sync::mpsc; -use std::time::Instant; +use std::time::{Instant, Duration}; -use mio; -use mio::channel::SendError; -use slab::Slab; use futures::{Future, Task, TaskHandle, Poll}; use futures_io::Ready; +use mio::channel::SendError; +use mio; +use slab::Slab; use slot::{self, Slot}; +use timer_wheel::{TimerWheel, Timeout}; static NEXT_LOOP_ID: AtomicUsize = ATOMIC_USIZE_INIT; scoped_thread_local!(static CURRENT_LOOP: Loop); @@ -32,6 +34,15 @@ pub struct Loop { tx: mio::channel::Sender, rx: mio::channel::Receiver, dispatch: RefCell>, + + // Timer wheel keeping track of all timeouts. The `usize` stored in the + // timer wheel is an index into the slab below. + // + // The slab below keeps track of the timeouts themselves as well as the + // state of the timeout itself. The `TimeoutToken` type is an index into the + // `timeouts` slab. + timer_wheel: RefCell>, + timeouts: RefCell>, } /// Handle to an event loop, used to construct I/O objects, send messages, and @@ -50,11 +61,20 @@ struct Scheduled { waiter: Option, } +enum TimeoutState { + NotFired, + Fired, + Waiting(TaskHandle), +} + enum Message { AddSource(IoSource, Arc>>), DropSource(usize), Schedule(usize, TaskHandle), Deschedule(usize), + AddTimeout(Instant, Arc>>), + UpdateTimeout(TimeoutToken, TaskHandle), + CancelTimeout(TimeoutToken), Shutdown, } @@ -96,6 +116,8 @@ impl Loop { tx: tx, rx: rx, dispatch: RefCell::new(Slab::new_starting_at(1, SLAB_CAPACITY)), + timeouts: RefCell::new(Slab::new_starting_at(0, SLAB_CAPACITY)), + timer_wheel: RefCell::new(TimerWheel::new()), }) } @@ -138,7 +160,14 @@ impl Loop { // attaching strace, or similar. let start = Instant::now(); loop { - match self.io.poll(&mut events, None) { + let timeout = self.timer_wheel.borrow().next_timeout().map(|t| { + if t < start { + Duration::new(0, 0) + } else { + t - start + } + }); + match self.io.poll(&mut events, timeout) { Ok(a) => { amt = a; break; @@ -151,21 +180,29 @@ impl Loop { } debug!("loop poll - {:?}", start.elapsed()); - // TODO: coalesce token sets for a given Wake? + // First up, process all timeouts that may have just occurred. let start = Instant::now(); + self.consume_timeouts(start); + + // Next, process all the events that came in. for i in 0..events.len() { let event = events.get(i).unwrap(); let token = usize::from(event.token()); + // Token 0 == our incoming message queue, so this means we + // process the whole queue of messages. if token == 0 { debug!("consuming notification queue"); self.consume_queue(); continue } + // For any other token we look at `dispatch` to see what we're + // supposed to do. If there's a waiter we get ready to notify + // it, and we also or-in atomically any events that have + // happened (currently read/write events). let mut waiter = None; - - if let Some(sched) = self.dispatch.borrow_mut().get_mut(token) { + if let Some(sched) = self.dispatch.get_mut().get_mut(token) { waiter = sched.waiter.take(); if event.kind().is_readable() { sched.source.readiness.fetch_or(1, Ordering::Relaxed); @@ -176,14 +213,11 @@ impl Loop { } else { debug!("notified on {} which no longer exists", token); } - debug!("dispatching {:?} {:?}", event.token(), event.kind()); - CURRENT_LOOP.set(&self, move || { - match waiter { - Some(waiter) => waiter.notify(), - None => debug!("no waiter"), - } - }); + // If we actually got a waiter, then notify! + if let Some(waiter) = waiter { + self.notify_handle(waiter); + } } debug!("loop process - {} events, {:?}", amt, start.elapsed()); @@ -192,6 +226,24 @@ impl Loop { debug!("loop is done!"); } + fn consume_timeouts(&mut self, now: Instant) { + while let Some(idx) = self.timer_wheel.get_mut().poll(now) { + trace!("firing timeout: {}", idx); + let handle = self.timeouts.get_mut()[idx].1.fire(); + if let Some(handle) = handle { + self.notify_handle(handle); + } + } + } + + /// Method used to notify a task handle. + /// + /// Note that this should be used instead fo `handle.notify()` to ensure + /// that the `CURRENT_LOOP` variable is set appropriately. + fn notify_handle(&self, handle: TaskHandle) { + CURRENT_LOOP.set(&self, || handle.notify()); + } + fn add_source(&self, source: IoSource) -> io::Result { let sched = Scheduled { source: source, @@ -225,7 +277,7 @@ impl Loop { } }; if let Some(to_call) = to_call { - to_call.notify(); + self.notify_handle(to_call); } } @@ -235,6 +287,32 @@ impl Loop { sched.waiter = None; } + fn add_timeout(&self, at: Instant) -> io::Result { + let mut timeouts = self.timeouts.borrow_mut(); + if timeouts.vacant_entry().is_none() { + let len = timeouts.count(); + timeouts.grow(len); + } + let entry = timeouts.vacant_entry().unwrap(); + let timeout = self.timer_wheel.borrow_mut().insert(at, entry.index()); + let entry = entry.insert((timeout, TimeoutState::NotFired)); + Ok(TimeoutToken { token: entry.index() }) + } + + fn update_timeout(&self, token: &TimeoutToken, handle: TaskHandle) { + let to_wake = self.timeouts.borrow_mut()[token.token].1.block(handle); + if let Some(to_wake) = to_wake { + self.notify_handle(to_wake); + } + } + + fn cancel_timeout(&self, token: &TimeoutToken) { + let pair = self.timeouts.borrow_mut().remove(token.token); + if let Some((timeout, _state)) = pair { + self.timer_wheel.borrow_mut().cancel(&timeout); + } + } + fn consume_queue(&self) { while let Ok(msg) = self.rx.try_recv() { self.notify(msg); @@ -252,6 +330,13 @@ impl Loop { Message::Schedule(tok, wake) => self.schedule(tok, wake), Message::Deschedule(tok) => self.deschedule(tok), Message::Shutdown => self.active.set(false), + + Message::AddTimeout(at, slot) => { + slot.try_produce(self.add_timeout(at)) + .ok().expect("interference with try_produce on timeout"); + } + Message::UpdateTimeout(t, handle) => self.update_timeout(&t, handle), + Message::CancelTimeout(t) => self.cancel_timeout(&t), } } } @@ -323,16 +408,14 @@ impl LoopHandle { /// with the event loop. pub fn add_source(&self, source: IoSource) -> AddSource { AddSource { - loop_handle: self.clone(), - source: Some(source), - result: None, + inner: LoopFuture { + loop_handle: self.clone(), + data: Some(source), + result: None, + } } } - fn add_source_(&self, source: IoSource, slot: Arc>>) { - self.send(Message::AddSource(source, slot)); - } - /// Begin listening for events on an event loop. /// /// Once an I/O object has been registered with the event loop through the @@ -394,6 +477,40 @@ impl LoopHandle { self.send(Message::DropSource(tok)); } + /// Adds a new timeout to get fired at the specified instant, notifying the + /// specified task. + pub fn add_timeout(&self, at: Instant) -> AddTimeout { + AddTimeout { + inner: LoopFuture { + loop_handle: self.clone(), + data: Some(at), + result: None, + }, + } + } + + /// Updates a previously added timeout to notify a new task instead. + /// + /// # Panics + /// + /// 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, task: &mut Task) { + let timeout = TimeoutToken { token: timeout.token }; + self.send(Message::UpdateTimeout(timeout, task.handle().clone())) + } + + /// Cancel a previously added timeout. + /// + /// # Panics + /// + /// 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 }; + self.send(Message::CancelTimeout(timeout)) + } + /// Send a message to the associated event loop that it should shut down, or /// otherwise break out of its current loop of iteration. /// @@ -417,9 +534,7 @@ impl LoopHandle { /// Created through the `LoopHandle::add_source` method, this future can also /// resolve to an error if there's an issue communicating with the event loop. pub struct AddSource { - loop_handle: LoopHandle, - source: Option, - result: Option<(Arc>>, slot::Token)>, + inner: LoopFuture, } impl Future for AddSource { @@ -427,6 +542,50 @@ impl Future for AddSource { type Error = io::Error; fn poll(&mut self, _task: &mut Task) -> Poll { + self.inner.poll(Loop::add_source) + } + + fn schedule(&mut self, task: &mut Task) { + self.inner.schedule(task, Message::AddSource) + } +} + +/// Return value from the `LoopHandle::add_timeout` method, a future that will +/// resolve to a `TimeoutToken` to configure the behavior of that timeout. +pub struct AddTimeout { + inner: LoopFuture, +} + +/// A token that identifies an active timeout. +pub struct TimeoutToken { + token: usize, +} + +impl Future for AddTimeout { + type Item = TimeoutToken; + type Error = io::Error; + + fn poll(&mut self, _task: &mut Task) -> Poll { + self.inner.poll(Loop::add_timeout) + } + + fn schedule(&mut self, task: &mut Task) { + self.inner.schedule(task, Message::AddTimeout) + } +} + +struct LoopFuture { + loop_handle: LoopHandle, + data: Option, + result: Option<(Arc>>, slot::Token)>, +} + +impl LoopFuture + where T: Send + 'static, +{ + fn poll(&mut self, f: F) -> Poll + where F: FnOnce(&Loop, U) -> io::Result, + { match self.result { Some((ref result, ref token)) => { result.cancel(*token); @@ -436,10 +595,10 @@ impl Future for AddSource { } } None => { - let source = &mut self.source; + let data = &mut self.data; self.loop_handle.with_loop(|lp| { match lp { - Some(lp) => lp.add_source(source.take().unwrap()).into(), + Some(lp) => f(lp, data.take().unwrap()).into(), None => Poll::NotReady, } }) @@ -447,7 +606,9 @@ impl Future for AddSource { } } - fn schedule(&mut self, task: &mut Task) { + fn schedule(&mut self, task: &mut Task, f: F) + where F: FnOnce(U, Arc>>) -> Message, + { if let Some((ref result, ref mut token)) = self.result { result.cancel(*token); let handle = task.handle().clone(); @@ -463,7 +624,26 @@ impl Future for AddSource { handle.notify(); }); self.result = Some((result.clone(), token)); - self.loop_handle.add_source_(self.source.take().unwrap(), result); + self.loop_handle.send(f(self.data.take().unwrap(), result)) + } +} + +impl TimeoutState { + fn block(&mut self, handle: TaskHandle) -> Option { + match *self { + TimeoutState::Fired => return Some(handle), + _ => {} + } + *self = TimeoutState::Waiting(handle); + None + } + + fn fire(&mut self) -> Option { + match mem::replace(self, TimeoutState::Fired) { + TimeoutState::NotFired => None, + TimeoutState::Fired => panic!("fired twice?"), + TimeoutState::Waiting(handle) => Some(handle), + } } } diff --git a/src/lib.rs b/src/lib.rs index aa50ddc0f..d08889633 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -23,6 +23,8 @@ mod readiness_stream; mod event_loop; mod tcp; mod udp; +mod timeout; +pub mod timer_wheel; #[path = "../../src/slot.rs"] mod slot; #[path = "../../src/lock.rs"] @@ -31,4 +33,5 @@ mod lock; pub use event_loop::{Loop, LoopHandle}; pub use readiness_stream::ReadinessStream; pub use tcp::{TcpListener, TcpStream}; +pub use timeout::Timeout; pub use udp::UdpSocket; diff --git a/src/timeout.rs b/src/timeout.rs new file mode 100644 index 000000000..e8ef7a225 --- /dev/null +++ b/src/timeout.rs @@ -0,0 +1,71 @@ +use std::io; +use std::time::{Duration, Instant}; + +use futures::{Future, Task, Poll}; +use futures_io::IoFuture; + +use LoopHandle; +use event_loop::TimeoutToken; + +/// A future representing the notification that a timeout has occurred. +/// +/// Timeouts are created through the `LoopHandle::timeout` or +/// `LoopHandle::timeout_at` methods indicating when a timeout should fire at. +/// Note that timeouts are not intended for high resolution timers, but rather +/// 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, +} + +impl LoopHandle { + /// Creates a new timeout which will fire at `dur` time into the future. + /// + /// This function will return a future that will resolve to the actual + /// timeout object. The timeout object itself is then a future which will be + /// set to fire at the specified point in the future. + pub fn timeout(self, dur: Duration) -> Box> { + self.timeout_at(Instant::now() + dur) + } + + /// Creates a new timeout which will fire at the time specified by `at`. + /// + /// This function will return a future that will resolve to the actual + /// timeout object. The timeout object itself is then a future which will be + /// set to fire at the specified point in the future. + pub fn timeout_at(self, at: Instant) -> Box> { + self.add_timeout(at).map(move |token| { + Timeout { + at: at, + token: token, + handle: self, + } + }).boxed() + } +} + +impl Future for Timeout { + type Item = (); + type Error = io::Error; + + fn poll(&mut self, _task: &mut Task) -> Poll<(), io::Error> { + // TODO: is this fast enough? + if self.at <= Instant::now() { + Poll::Ok(()) + } else { + Poll::NotReady + } + } + + fn schedule(&mut self, task: &mut Task) { + self.handle.update_timeout(&self.token, task); + } +} + +impl Drop for Timeout { + fn drop(&mut self) { + self.handle.cancel_timeout(&self.token); + } +} diff --git a/src/timer_wheel.rs b/src/timer_wheel.rs new file mode 100644 index 000000000..37f49f811 --- /dev/null +++ b/src/timer_wheel.rs @@ -0,0 +1,294 @@ +//! A timer wheel implementation + +use std::cmp; +use std::mem; +use std::time::{Instant, Duration}; + +use slab::Slab; + +/// An implementation of a timer wheel where data can be associated with each +/// timer firing. +/// +/// This structure implements a timer wheel data structure where each timeout +/// has a piece of associated data, `T`. A timer wheel supports O(1) insertion +/// and removal of timers, as well as quickly figuring out what needs to get +/// fired. +/// +/// Note, though, that the resolution of a timer wheel means that timeouts will +/// not arrive promptly when they expire, but rather in certain increments of +/// each time. The time delta between each slot of a time wheel is of a fixed +/// length, meaning that if a timeout is scheduled between two slots it'll end +/// up getting scheduled into the later slot. +pub struct TimerWheel { + // Actual timer wheel itself. + // + // Each slot represents a fixed duration of time, and this wheel also + // behaves like a ring buffer. All timeouts scheduled will correspond to one + // slot and therefore each slot has a linked list of timeouts scheduled in + // it. Right now linked lists are done through indices into the `slab` + // below. + // + // Each slot also contains the next timeout associated with it (the minimum + // of the entire linked list). + wheel: Vec, + + // A slab containing all the timeout entries themselves. This is the memory + // backing the "linked lists" in the wheel above. Each entry has a prev/next + // pointer (indices in this array) along with the data associated with the + // timeout and the time the timeout will fire. + slab: Slab, usize>, + + // The instant that this timer was created, through which all other timeout + // computations are relative to. + start: Instant, + + // State used during `poll`. The `cur_wheel_tick` field is the current tick + // we've poll'd to. That is, all events from `cur_wheel_tick` to the + // actual current tick in time still need to be processed. + // + // The `cur_slab_idx` variable is basically just an iterator over the linked + // list associated with a wheel slot. This will get incremented as we move + // forward in `poll` + cur_wheel_tick: u64, + cur_slab_idx: usize, +} + +#[derive(Clone)] +struct Slot { + head: usize, + next_timeout: Option, +} + +struct Entry { + data: T, + when: Instant, + prev: usize, + next: usize, +} + +/// A timeout which has been scheduled with a timer wheel. +/// +/// This can be used to later cancel a timeout, if necessary. +pub struct Timeout { + when: Instant, + slab_idx: usize, +} + +const EMPTY: usize = 0; +const LEN: usize = 256; +const MASK: usize = LEN - 1; +const TICK_MS: u64 = 100; + +impl TimerWheel { + /// Creates a new timer wheel configured with no timeouts and with the + /// default parameters. + /// + /// Currently this is a timer wheel of length 256 with a 100ms time + /// resolution. + pub fn new() -> TimerWheel { + TimerWheel { + wheel: vec![Slot { head: EMPTY, next_timeout: None }; LEN], + slab: Slab::new_starting_at(1, 256), + start: Instant::now(), + cur_wheel_tick: 0, + cur_slab_idx: EMPTY, + } + } + + /// Creates a new timeout to get fired at a particular point in the future. + /// + /// The timeout will be associated with the specified `data`, and this data + /// will be returned from `poll` when it's ready. + /// + /// The returned `Timeout` can later get passesd to `cancel` to retrieve the + /// data and ensure the timeout doesn't fire. + /// + /// This method completes in O(1) time. + /// + /// # Panics + /// + /// 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 { + // 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. + let mut tick = self.time_to_ticks(at); + if tick <= self.cur_wheel_tick { + debug!("moving {} to {}", tick, self.cur_wheel_tick + 1); + tick = self.cur_wheel_tick + 1; + } + let wheel_idx = self.ticks_to_wheel_idx(tick); + trace!("inserting timeout at {} for {}", wheel_idx, 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(); + self.slab.grow(amt); + } + + // Insert ourselves at the head of the linked list in the wheel. + let slot = &mut self.wheel[wheel_idx]; + let prev_head; + { + let entry = self.slab.vacant_entry().unwrap(); + prev_head = mem::replace(&mut slot.head, entry.index()); + + entry.insert(Entry { + data: data, + when: at, + prev: EMPTY, + next: prev_head, + }); + } + if prev_head != EMPTY { + self.slab[prev_head].prev = slot.head; + } + + // 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; + let at = cmp::max(actual_tick, at); + slot.next_timeout = Some(at); + } + + Timeout { + when: at, + slab_idx: slot.head, + } + } + + /// Queries this timer to see if any timeouts are ready to fire. + /// + /// This function will advance the internal wheel to the time specified by + /// `at`, returning any timeout which has happened up to that point. This + /// method should be called in a loop until it returns `None` to ensure that + /// all timeouts are processed. + /// + /// # Panics + /// + /// This method will panic if `at` is before the instant that this timer + /// wheel was created. + pub fn poll(&mut self, at: Instant) -> Option { + let wheel_tick = self.time_to_ticks(at); + + trace!("polling {} => {}", self.cur_wheel_tick, wheel_tick); + + // Advance forward in time to the `wheel_tick` specified. + // + // TODO: don't visit slots in the wheel more than once + while self.cur_wheel_tick <= wheel_tick { + let head = self.cur_slab_idx; + trace!("next head[{} => {}]: {}", + self.cur_wheel_tick, wheel_tick, head); + + // If the current slot has no entries or we're done iterating go to + // the next tick. + if head == EMPTY { + self.cur_wheel_tick += 1; + let idx = self.ticks_to_wheel_idx(self.cur_wheel_tick); + self.cur_slab_idx = self.wheel[idx].head; + continue + } + + // If we're starting to iterate over a slot, clear its timeout as + // we're probably going to remove entries. As we skip over each + // element of this slot we'll restore the `next_timeout` field if + // necessary. + let idx = self.ticks_to_wheel_idx(self.cur_wheel_tick); + if head == self.wheel[idx].head { + self.wheel[idx].next_timeout = None; + } + + // Otherwise, continue iterating over the linked list in the wheel + // slot we're on and remove anything which has expired. + self.cur_slab_idx = self.slab[head].next; + let head_timeout = self.slab[head].when; + if self.time_to_ticks(head_timeout) <= self.time_to_ticks(at) { + return self.remove_slab(head).map(|e| e.data) + } else { + let next = self.wheel[idx].next_timeout.unwrap_or(head_timeout); + if head_timeout <= next { + self.wheel[idx].next_timeout = Some(head_timeout); + } + } + } + + None + } + + /// Returns the instant in time that corresponds to the next timeout + /// scheduled in this wheel. + pub fn next_timeout(&self) -> Option { + // TODO: can this be optimized to not look at the whole array? + let timeouts = self.wheel.iter().map(|slot| slot.next_timeout); + let min = timeouts.fold(None, |prev, cur| { + match (prev, cur) { + (None, cur) => cur, + (Some(time), None) => Some(time), + (Some(a), Some(b)) => Some(cmp::min(a, b)), + } + }); + let time = min.map(|min| min + Duration::from_millis(TICK_MS / 2)); + if let Some(time) = time { + debug!("next timeout {:?}", time); + debug!("now {:?}", Instant::now()); + } + return time + } + + /// Cancels the specified timeout. + /// + /// For timeouts previously registered via `insert` they can be passed back + /// to this method to cancel the associated timeout, retrieving the value + /// inserted if the timeout has not already fired. + /// + /// This method completes in O(1) time. + /// + /// # Panics + /// + /// This method may panic if `timeout` wasn't created by this timer wheel. + pub fn cancel(&mut self, timeout: &Timeout) -> Option { + match self.slab.get(timeout.slab_idx) { + Some(e) if e.when == timeout.when => {} + _ => return None, + } + + self.remove_slab(timeout.slab_idx).map(|e| e.data) + } + + fn remove_slab(&mut self, slab_idx: usize) -> Option> { + let entry = match self.slab.remove(slab_idx) { + Some(e) => e, + None => return None, + }; + + // Remove the node from the linked list + if entry.prev == EMPTY { + let idx = self.ticks_to_wheel_idx(self.time_to_ticks(entry.when)); + self.wheel[idx].head = entry.next; + } else { + self.slab[entry.prev].next = entry.next; + } + if entry.next != EMPTY { + self.slab[entry.next].prev = entry.prev; + } + + return Some(entry) + } + + fn time_to_ticks(&self, time: Instant) -> u64 { + let dur = time - self.start; + let ms = dur.subsec_nanos() as u64 / 1_000_000; + let ms = dur.as_secs() + .checked_mul(1_000) + .and_then(|m| m.checked_add(ms)) + .expect("overflow scheduling timeout"); + (ms + TICK_MS / 2) / TICK_MS + } + + fn ticks_to_wheel_idx(&self, ticks: u64) -> usize { + (ticks as usize) & MASK + } +} diff --git a/tests/timeout.rs b/tests/timeout.rs new file mode 100644 index 000000000..7676a50c3 --- /dev/null +++ b/tests/timeout.rs @@ -0,0 +1,25 @@ +extern crate env_logger; +extern crate futures; +extern crate futures_mio; + +use std::time::{Instant, Duration}; + +use futures::Future; + +macro_rules! t { + ($e:expr) => (match $e { + Ok(e) => e, + Err(e) => panic!("{} failed with {:?}", stringify!($e), e), + }) +} + +#[test] +fn smoke() { + drop(env_logger::init()); + let mut l = t!(futures_mio::Loop::new()); + let dur = Duration::from_millis(10); + let timeout = l.handle().timeout(dur).and_then(|t| t); + let start = Instant::now(); + t!(l.run(timeout)); + assert!(start.elapsed() >= dur); +} diff --git a/tests/timer-wheel.rs b/tests/timer-wheel.rs new file mode 100644 index 000000000..6b3078ef1 --- /dev/null +++ b/tests/timer-wheel.rs @@ -0,0 +1,137 @@ +extern crate env_logger; +extern crate futures_mio; + +use std::time::{Instant, Duration}; + +use futures_mio::timer_wheel::TimerWheel; + +fn ms(amt: u64) -> Duration { + Duration::from_millis(amt) +} + +#[test] +fn smoke() { + drop(env_logger::init()); + let mut timer = TimerWheel::::new(); + let now = Instant::now(); + + assert!(timer.poll(now).is_none()); + assert!(timer.poll(now).is_none()); + + timer.insert(now + ms(200), 3); + + assert!(timer.poll(now).is_none()); + assert!(timer.poll(now + ms(100)).is_none()); + let res = timer.poll(now + ms(200)); + assert!(res.is_some()); + assert_eq!(res.unwrap(), 3); +} + +#[test] +fn poll_past_done() { + drop(env_logger::init()); + let mut timer = TimerWheel::::new(); + let now = Instant::now(); + + timer.insert(now + ms(200), 3); + let res = timer.poll(now + ms(300)); + assert!(res.is_some()); + assert_eq!(res.unwrap(), 3); +} + +#[test] +fn multiple_ready() { + drop(env_logger::init()); + let mut timer = TimerWheel::::new(); + let now = Instant::now(); + + timer.insert(now + ms(200), 3); + timer.insert(now + ms(201), 4); + timer.insert(now + ms(202), 5); + timer.insert(now + ms(300), 6); + timer.insert(now + ms(301), 7); + + let mut found = Vec::new(); + while let Some(i) = timer.poll(now + ms(400)) { + found.push(i); + } + found.sort(); + assert_eq!(found, [3, 4, 5, 6, 7]); +} + +#[test] +fn poll_now() { + drop(env_logger::init()); + let mut timer = TimerWheel::::new(); + let now = Instant::now(); + + timer.insert(now, 3); + let res = timer.poll(now + ms(100)); + assert!(res.is_some()); + assert_eq!(res.unwrap(), 3); +} + +#[test] +fn cancel() { + drop(env_logger::init()); + let mut timer = TimerWheel::::new(); + let now = Instant::now(); + + let timeout = timer.insert(now + ms(800), 3); + assert!(timer.poll(now + ms(200)).is_none()); + assert!(timer.poll(now + ms(400)).is_none()); + assert_eq!(timer.cancel(&timeout), Some(3)); + assert!(timer.poll(now + ms(600)).is_none()); + assert!(timer.poll(now + ms(800)).is_none()); + assert!(timer.poll(now + ms(1000)).is_none()); +} + +#[test] +fn next_timeout() { + drop(env_logger::init()); + let mut timer = TimerWheel::::new(); + let now = Instant::now(); + + assert!(timer.next_timeout().is_none()); + timer.insert(now + ms(400), 3); + let timeout = timer.next_timeout().expect("wanted a next_timeout"); + assert_eq!(timeout, now + ms(400)); + + timer.insert(now + ms(1000), 3); + let timeout = timer.next_timeout().expect("wanted a next_timeout"); + assert_eq!(timeout, now + ms(400)); +} + +#[test] +fn around_the_boundary() { + drop(env_logger::init()); + let mut timer = TimerWheel::::new(); + let now = Instant::now(); + + timer.insert(now + ms(199), 3); + timer.insert(now + ms(200), 4); + timer.insert(now + ms(201), 5); + timer.insert(now + ms(251), 6); + + 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!(timer.poll(now + ms(300)), Some(6)); + assert_eq!(timer.poll(now + ms(300)), None); +} + +#[test] +fn remove_clears_timeout() { + drop(env_logger::init()); + let mut timer = TimerWheel::::new(); + let now = Instant::now(); + + timer.insert(now + ms(100), 3); + assert_eq!(timer.next_timeout(), Some(now + ms(100))); + assert_eq!(timer.poll(now + ms(200)), Some(3)); + assert_eq!(timer.next_timeout(), None); +}