b2c6d6fd3f
This commit implements the set{,_read,_write}_timeout() methods for the libuv-based networking I/O objects. The implementation details are commented thoroughly throughout the implementation.
324 lines
9.8 KiB
Rust
324 lines
9.8 KiB
Rust
// Copyright 2013 The Rust Project Developers. See the COPYRIGHT
|
|
// file at the top-level directory of this distribution and at
|
|
// http://rust-lang.org/COPYRIGHT.
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
|
|
// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
|
|
// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
|
|
// option. This file may not be copied, modified, or distributed
|
|
// except according to those terms.
|
|
|
|
use std::mem;
|
|
use std::rt::rtio::RtioTimer;
|
|
use std::rt::task::BlockedTask;
|
|
|
|
use homing::{HomeHandle, HomingIO};
|
|
use super::{UvHandle, ForbidUnwind, ForbidSwitch, wait_until_woken_after, Loop};
|
|
use uvio::UvIoFactory;
|
|
use uvll;
|
|
|
|
pub struct TimerWatcher {
|
|
pub handle: *uvll::uv_timer_t,
|
|
home: HomeHandle,
|
|
action: Option<NextAction>,
|
|
blocker: Option<BlockedTask>,
|
|
id: uint, // see comments in timer_cb
|
|
}
|
|
|
|
pub enum NextAction {
|
|
WakeTask,
|
|
SendOnce(Sender<()>),
|
|
SendMany(Sender<()>, uint),
|
|
}
|
|
|
|
impl TimerWatcher {
|
|
pub fn new(io: &mut UvIoFactory) -> Box<TimerWatcher> {
|
|
let handle = io.make_handle();
|
|
let me = box TimerWatcher::new_home(&io.loop_, handle);
|
|
me.install()
|
|
}
|
|
|
|
pub fn new_home(loop_: &Loop, home: HomeHandle) -> TimerWatcher {
|
|
let handle = UvHandle::alloc(None::<TimerWatcher>, uvll::UV_TIMER);
|
|
assert_eq!(unsafe { uvll::uv_timer_init(loop_.handle, handle) }, 0);
|
|
TimerWatcher {
|
|
handle: handle,
|
|
action: None,
|
|
blocker: None,
|
|
home: home,
|
|
id: 0,
|
|
}
|
|
}
|
|
|
|
pub fn start(&mut self, f: uvll::uv_timer_cb, msecs: u64, period: u64) {
|
|
assert_eq!(unsafe {
|
|
uvll::uv_timer_start(self.handle, f, msecs, period)
|
|
}, 0)
|
|
}
|
|
|
|
pub fn stop(&mut self) {
|
|
assert_eq!(unsafe { uvll::uv_timer_stop(self.handle) }, 0)
|
|
}
|
|
|
|
pub unsafe fn set_data<T>(&mut self, data: *T) {
|
|
uvll::set_data_for_uv_handle(self.handle, data);
|
|
}
|
|
}
|
|
|
|
impl HomingIO for TimerWatcher {
|
|
fn home<'r>(&'r mut self) -> &'r mut HomeHandle { &mut self.home }
|
|
}
|
|
|
|
impl UvHandle<uvll::uv_timer_t> for TimerWatcher {
|
|
fn uv_handle(&self) -> *uvll::uv_timer_t { self.handle }
|
|
}
|
|
|
|
impl RtioTimer for TimerWatcher {
|
|
fn sleep(&mut self, msecs: u64) {
|
|
// As with all of the below functions, we must be extra careful when
|
|
// destroying the previous action. If the previous action was a channel,
|
|
// destroying it could invoke a context switch. For these situtations,
|
|
// we must temporarily un-home ourselves, then destroy the action, and
|
|
// then re-home again.
|
|
let missile = self.fire_homing_missile();
|
|
self.id += 1;
|
|
self.stop();
|
|
let _missile = match mem::replace(&mut self.action, None) {
|
|
None => missile, // no need to do a homing dance
|
|
Some(action) => {
|
|
drop(missile); // un-home ourself
|
|
drop(action); // destroy the previous action
|
|
self.fire_homing_missile() // re-home ourself
|
|
}
|
|
};
|
|
|
|
// If the descheduling operation unwinds after the timer has been
|
|
// started, then we need to call stop on the timer.
|
|
let _f = ForbidUnwind::new("timer");
|
|
|
|
self.action = Some(WakeTask);
|
|
wait_until_woken_after(&mut self.blocker, &self.uv_loop(), || {
|
|
self.start(timer_cb, msecs, 0);
|
|
});
|
|
self.stop();
|
|
}
|
|
|
|
fn oneshot(&mut self, msecs: u64) -> Receiver<()> {
|
|
let (tx, rx) = channel();
|
|
|
|
// similarly to the destructor, we must drop the previous action outside
|
|
// of the homing missile
|
|
let _prev_action = {
|
|
let _m = self.fire_homing_missile();
|
|
self.id += 1;
|
|
self.stop();
|
|
self.start(timer_cb, msecs, 0);
|
|
mem::replace(&mut self.action, Some(SendOnce(tx)))
|
|
};
|
|
|
|
return rx;
|
|
}
|
|
|
|
fn period(&mut self, msecs: u64) -> Receiver<()> {
|
|
let (tx, rx) = channel();
|
|
|
|
// similarly to the destructor, we must drop the previous action outside
|
|
// of the homing missile
|
|
let _prev_action = {
|
|
let _m = self.fire_homing_missile();
|
|
self.id += 1;
|
|
self.stop();
|
|
self.start(timer_cb, msecs, msecs);
|
|
mem::replace(&mut self.action, Some(SendMany(tx, self.id)))
|
|
};
|
|
|
|
return rx;
|
|
}
|
|
}
|
|
|
|
extern fn timer_cb(handle: *uvll::uv_timer_t) {
|
|
let _f = ForbidSwitch::new("timer callback can't switch");
|
|
let timer: &mut TimerWatcher = unsafe { UvHandle::from_uv_handle(&handle) };
|
|
|
|
match timer.action.take_unwrap() {
|
|
WakeTask => {
|
|
let task = timer.blocker.take_unwrap();
|
|
let _ = task.wake().map(|t| t.reawaken());
|
|
}
|
|
SendOnce(chan) => { let _ = chan.send_opt(()); }
|
|
SendMany(chan, id) => {
|
|
let _ = chan.send_opt(());
|
|
|
|
// Note that the above operation could have performed some form of
|
|
// scheduling. This means that the timer may have decided to insert
|
|
// some other action to happen. This 'id' keeps track of the updates
|
|
// to the timer, so we only reset the action back to sending on this
|
|
// channel if the id has remained the same. This is essentially a
|
|
// bug in that we have mutably aliasable memory, but that's libuv
|
|
// for you. We're guaranteed to all be running on the same thread,
|
|
// so there's no need for any synchronization here.
|
|
if timer.id == id {
|
|
timer.action = Some(SendMany(chan, id));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Drop for TimerWatcher {
|
|
fn drop(&mut self) {
|
|
// note that this drop is a little subtle. Dropping a channel which is
|
|
// held internally may invoke some scheduling operations. We can't take
|
|
// the channel unless we're on the home scheduler, but once we're on the
|
|
// home scheduler we should never move. Hence, we take the timer's
|
|
// action item and then move it outside of the homing block.
|
|
let _action = {
|
|
let _m = self.fire_homing_missile();
|
|
self.stop();
|
|
self.close();
|
|
self.action.take()
|
|
};
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod test {
|
|
use std::rt::rtio::RtioTimer;
|
|
use super::super::local_loop;
|
|
use super::TimerWatcher;
|
|
|
|
#[test]
|
|
fn oneshot() {
|
|
let mut timer = TimerWatcher::new(local_loop());
|
|
let port = timer.oneshot(1);
|
|
port.recv();
|
|
let port = timer.oneshot(1);
|
|
port.recv();
|
|
}
|
|
|
|
#[test]
|
|
fn override() {
|
|
let mut timer = TimerWatcher::new(local_loop());
|
|
let oport = timer.oneshot(1);
|
|
let pport = timer.period(1);
|
|
timer.sleep(1);
|
|
assert_eq!(oport.recv_opt(), Err(()));
|
|
assert_eq!(pport.recv_opt(), Err(()));
|
|
timer.oneshot(1).recv();
|
|
}
|
|
|
|
#[test]
|
|
fn period() {
|
|
let mut timer = TimerWatcher::new(local_loop());
|
|
let port = timer.period(1);
|
|
port.recv();
|
|
port.recv();
|
|
let port2 = timer.period(1);
|
|
port2.recv();
|
|
port2.recv();
|
|
}
|
|
|
|
#[test]
|
|
fn sleep() {
|
|
let mut timer = TimerWatcher::new(local_loop());
|
|
timer.sleep(1);
|
|
timer.sleep(1);
|
|
}
|
|
|
|
#[test] #[should_fail]
|
|
fn oneshot_fail() {
|
|
let mut timer = TimerWatcher::new(local_loop());
|
|
let _port = timer.oneshot(1);
|
|
fail!();
|
|
}
|
|
|
|
#[test] #[should_fail]
|
|
fn period_fail() {
|
|
let mut timer = TimerWatcher::new(local_loop());
|
|
let _port = timer.period(1);
|
|
fail!();
|
|
}
|
|
|
|
#[test] #[should_fail]
|
|
fn normal_fail() {
|
|
let _timer = TimerWatcher::new(local_loop());
|
|
fail!();
|
|
}
|
|
|
|
#[test]
|
|
fn closing_channel_during_drop_doesnt_kill_everything() {
|
|
// see issue #10375
|
|
let mut timer = TimerWatcher::new(local_loop());
|
|
let timer_port = timer.period(1000);
|
|
|
|
spawn(proc() {
|
|
let _ = timer_port.recv_opt();
|
|
});
|
|
|
|
// when we drop the TimerWatcher we're going to destroy the channel,
|
|
// which must wake up the task on the other end
|
|
}
|
|
|
|
#[test]
|
|
fn reset_doesnt_switch_tasks() {
|
|
// similar test to the one above.
|
|
let mut timer = TimerWatcher::new(local_loop());
|
|
let timer_port = timer.period(1000);
|
|
|
|
spawn(proc() {
|
|
let _ = timer_port.recv_opt();
|
|
});
|
|
|
|
drop(timer.oneshot(1));
|
|
}
|
|
#[test]
|
|
fn reset_doesnt_switch_tasks2() {
|
|
// similar test to the one above.
|
|
let mut timer = TimerWatcher::new(local_loop());
|
|
let timer_port = timer.period(1000);
|
|
|
|
spawn(proc() {
|
|
let _ = timer_port.recv_opt();
|
|
});
|
|
|
|
timer.sleep(1);
|
|
}
|
|
|
|
#[test]
|
|
fn sender_goes_away_oneshot() {
|
|
let port = {
|
|
let mut timer = TimerWatcher::new(local_loop());
|
|
timer.oneshot(1000)
|
|
};
|
|
assert_eq!(port.recv_opt(), Err(()));
|
|
}
|
|
|
|
#[test]
|
|
fn sender_goes_away_period() {
|
|
let port = {
|
|
let mut timer = TimerWatcher::new(local_loop());
|
|
timer.period(1000)
|
|
};
|
|
assert_eq!(port.recv_opt(), Err(()));
|
|
}
|
|
|
|
#[test]
|
|
fn receiver_goes_away_oneshot() {
|
|
let mut timer1 = TimerWatcher::new(local_loop());
|
|
drop(timer1.oneshot(1));
|
|
let mut timer2 = TimerWatcher::new(local_loop());
|
|
// while sleeping, the prevous timer should fire and not have its
|
|
// callback do something terrible.
|
|
timer2.sleep(2);
|
|
}
|
|
|
|
#[test]
|
|
fn receiver_goes_away_period() {
|
|
let mut timer1 = TimerWatcher::new(local_loop());
|
|
drop(timer1.period(1));
|
|
let mut timer2 = TimerWatcher::new(local_loop());
|
|
// while sleeping, the prevous timer should fire and not have its
|
|
// callback do something terrible.
|
|
timer2.sleep(2);
|
|
}
|
|
}
|