itsscb/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2025-10-01 12:20:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

Don't remove timeouts that have fired

Browse Source 
Closes #22
This commit is contained in:
Alex Crichton 2016-09-08 07:46:47 -07:00
parent cd331dde34
commit 83784fd983
2 changed files with 16 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/reactor/mod.rs
View File

@ -69,7 +69,7 @@ struct Inner {
// state of the timeout itself. The `TimeoutToken` type is an index into the // state of the timeout itself. The `TimeoutToken` type is an index into the
// `timeouts` slab. // `timeouts` slab.
timer_heap: Heap<(Instant, usize)>, timer_heap: Heap<(Instant, usize)>,
timeouts: Slab<(Slot, TimeoutState)>, timeouts: Slab<(Option<Slot>, TimeoutState)>,
} }
/// Handle to an event loop, used to construct I/O objects, send messages, and /// Handle to an event loop, used to construct I/O objects, send messages, and
@ -359,6 +359,7 @@ impl Core {
let (_, slab_idx) = inner.timer_heap.pop().unwrap(); let (_, slab_idx) = inner.timer_heap.pop().unwrap();
trace!("firing timeout: {}", slab_idx); trace!("firing timeout: {}", slab_idx);
inner.timeouts[slab_idx].0.take().unwrap();
let handle = inner.timeouts[slab_idx].1.fire(); let handle = inner.timeouts[slab_idx].1.fire();
drop(inner); drop(inner);
if let Some(handle) = handle { if let Some(handle) = handle {
@ -457,7 +458,7 @@ impl Inner {
} }
let entry = self.timeouts.vacant_entry().unwrap(); let entry = self.timeouts.vacant_entry().unwrap();
let slot = self.timer_heap.push((at, entry.index())); let slot = self.timer_heap.push((at, entry.index()));
let entry = entry.insert((slot, TimeoutState::NotFired)); let entry = entry.insert((Some(slot), TimeoutState::NotFired));
debug!("added a timeout: {}", entry.index()); debug!("added a timeout: {}", entry.index());
Ok((entry.index(), at)) Ok((entry.index(), at))
} }
@ -470,7 +471,7 @@ impl Inner {
fn cancel_timeout(&mut self, token: usize) { fn cancel_timeout(&mut self, token: usize) {
debug!("cancel a timeout: {}", token); debug!("cancel a timeout: {}", token);
let pair = self.timeouts.remove(token); let pair = self.timeouts.remove(token);
if let Some((slot, _state)) = pair { if let Some((Some(slot), _state)) = pair {
self.timer_heap.remove(slot); self.timer_heap.remove(slot);
} }
} }

12
tests/timeout.rs
View File

@ -23,3 +23,15 @@ fn smoke() {
t!(l.run(timeout)); t!(l.run(timeout));
assert!(start.elapsed() >= dur); assert!(start.elapsed() >= dur);
} }
#[test]
fn two() {
drop(env_logger::init());
let mut l = t!(Core::new());
let dur = Duration::from_millis(10);
let timeout = t!(Timeout::new(dur, &l.handle()));
t!(l.run(timeout));
let timeout = t!(Timeout::new(dur, &l.handle()));
t!(l.run(timeout));
}