src/librustuv/timer.rs

   1 // Copyright 2013 The Rust Project Developers. See the COPYRIGHT
   2 // file at the top-level directory of this distribution and at
   3 // http://rust-lang.org/COPYRIGHT.
   4 //
   5 // Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
   6 // http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
   7 // <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
   8 // option. This file may not be copied, modified, or distributed
   9 // except according to those terms.
  10
  11 use std::mem;
  12 use std::rt::rtio::RtioTimer;
  13 use std::rt::task::BlockedTask;
  14
  15 use homing::{HomeHandle, HomingIO};
  16 use super::{UvHandle, ForbidUnwind, ForbidSwitch, wait_until_woken_after, Loop};
  17 use uvio::UvIoFactory;
  18 use uvll;
  19
  20 pub struct TimerWatcher {
  21     handle: *uvll::uv_timer_t,
  22     home: HomeHandle,
  23     action: Option<NextAction>,
  24     blocker: Option<BlockedTask>,
  25     id: uint, // see comments in timer_cb
  26 }
  27
  28 pub enum NextAction {
  29     WakeTask,
  30     SendOnce(Sender<()>),
  31     SendMany(Sender<()>, uint),
  32 }
  33
  34 impl TimerWatcher {
  35     pub fn new(io: &mut UvIoFactory) -> ~TimerWatcher {
  36         let handle = io.make_handle();
  37         let me = ~TimerWatcher::new_home(&io.loop_, handle);
  38         me.install()
  39     }
  40
  41     pub fn new_home(loop_: &Loop, home: HomeHandle) -> TimerWatcher {
  42         let handle = UvHandle::alloc(None::<TimerWatcher>, uvll::UV_TIMER);
  43         assert_eq!(unsafe { uvll::uv_timer_init(loop_.handle, handle) }, 0);
  44         TimerWatcher {
  45             handle: handle,
  46             action: None,
  47             blocker: None,
  48             home: home,
  49             id: 0,
  50         }
  51     }
  52
  53     pub fn start(&mut self, f: uvll::uv_timer_cb, msecs: u64, period: u64) {
  54         assert_eq!(unsafe {
  55             uvll::uv_timer_start(self.handle, f, msecs, period)
  56         }, 0)
  57     }
  58
  59     pub fn stop(&mut self) {
  60         assert_eq!(unsafe { uvll::uv_timer_stop(self.handle) }, 0)
  61     }
  62
  63     pub unsafe fn set_data<T>(&mut self, data: *T) {
  64         uvll::set_data_for_uv_handle(self.handle, data);
  65     }
  66 }
  67
  68 impl HomingIO for TimerWatcher {
  69     fn home<'r>(&'r mut self) -> &'r mut HomeHandle { &mut self.home }
  70 }
  71
  72 impl UvHandle<uvll::uv_timer_t> for TimerWatcher {
  73     fn uv_handle(&self) -> *uvll::uv_timer_t { self.handle }
  74 }
  75
  76 impl RtioTimer for TimerWatcher {
  77     fn sleep(&mut self, msecs: u64) {
  78         // As with all of the below functions, we must be extra careful when
  79         // destroying the previous action. If the previous action was a channel,
  80         // destroying it could invoke a context switch. For these situtations,
  81         // we must temporarily un-home ourselves, then destroy the action, and
  82         // then re-home again.
  83         let missile = self.fire_homing_missile();
  84         self.id += 1;
  85         self.stop();
  86         let _missile = match mem::replace(&mut self.action, None) {
  87             None => missile, // no need to do a homing dance
  88             Some(action) => {
  89                 drop(missile);      // un-home ourself
  90                 drop(action);       // destroy the previous action
  91                 self.fire_homing_missile()  // re-home ourself
  92             }
  93         };
  94
  95         // If the descheduling operation unwinds after the timer has been
  96         // started, then we need to call stop on the timer.
  97         let _f = ForbidUnwind::new("timer");
  98
  99         self.action = Some(WakeTask);
 100         wait_until_woken_after(&mut self.blocker, &self.uv_loop(), || {
 101             self.start(timer_cb, msecs, 0);
 102         });
 103         self.stop();
 104     }
 105
 106     fn oneshot(&mut self, msecs: u64) -> Receiver<()> {
 107         let (tx, rx) = channel();
 108
 109         // similarly to the destructor, we must drop the previous action outside
 110         // of the homing missile
 111         let _prev_action = {
 112             let _m = self.fire_homing_missile();
 113             self.id += 1;
 114             self.stop();
 115             self.start(timer_cb, msecs, 0);
 116             mem::replace(&mut self.action, Some(SendOnce(tx)))
 117         };
 118
 119         return rx;
 120     }
 121
 122     fn period(&mut self, msecs: u64) -> Receiver<()> {
 123         let (tx, rx) = channel();
 124
 125         // similarly to the destructor, we must drop the previous action outside
 126         // of the homing missile
 127         let _prev_action = {
 128             let _m = self.fire_homing_missile();
 129             self.id += 1;
 130             self.stop();
 131             self.start(timer_cb, msecs, msecs);
 132             mem::replace(&mut self.action, Some(SendMany(tx, self.id)))
 133         };
 134
 135         return rx;
 136     }
 137 }
 138
 139 extern fn timer_cb(handle: *uvll::uv_timer_t) {
 140     let _f = ForbidSwitch::new("timer callback can't switch");
 141     let timer: &mut TimerWatcher = unsafe { UvHandle::from_uv_handle(&handle) };
 142
 143     match timer.action.take_unwrap() {
 144         WakeTask => {
 145             let task = timer.blocker.take_unwrap();
 146             let _ = task.wake().map(|t| t.reawaken());
 147         }
 148         SendOnce(chan) => { let _ = chan.send_opt(()); }
 149         SendMany(chan, id) => {
 150             let _ = chan.send_opt(());
 151
 152             // Note that the above operation could have performed some form of
 153             // scheduling. This means that the timer may have decided to insert
 154             // some other action to happen. This 'id' keeps track of the updates
 155             // to the timer, so we only reset the action back to sending on this
 156             // channel if the id has remained the same. This is essentially a
 157             // bug in that we have mutably aliasable memory, but that's libuv
 158             // for you. We're guaranteed to all be running on the same thread,
 159             // so there's no need for any synchronization here.
 160             if timer.id == id {
 161                 timer.action = Some(SendMany(chan, id));
 162             }
 163         }
 164     }
 165 }
 166
 167 impl Drop for TimerWatcher {
 168     fn drop(&mut self) {
 169         // note that this drop is a little subtle. Dropping a channel which is
 170         // held internally may invoke some scheduling operations. We can't take
 171         // the channel unless we're on the home scheduler, but once we're on the
 172         // home scheduler we should never move. Hence, we take the timer's
 173         // action item and then move it outside of the homing block.
 174         let _action = {
 175             let _m = self.fire_homing_missile();
 176             self.stop();
 177             self.close();
 178             self.action.take()
 179         };
 180     }
 181 }
 182
 183 #[cfg(test)]
 184 mod test {
 185     use std::rt::rtio::RtioTimer;
 186     use super::super::local_loop;
 187     use super::TimerWatcher;
 188
 189     #[test]
 190     fn oneshot() {
 191         let mut timer = TimerWatcher::new(local_loop());
 192         let port = timer.oneshot(1);
 193         port.recv();
 194         let port = timer.oneshot(1);
 195         port.recv();
 196     }
 197
 198     #[test]
 199     fn override() {
 200         let mut timer = TimerWatcher::new(local_loop());
 201         let oport = timer.oneshot(1);
 202         let pport = timer.period(1);
 203         timer.sleep(1);
 204         assert_eq!(oport.recv_opt(), Err(()));
 205         assert_eq!(pport.recv_opt(), Err(()));
 206         timer.oneshot(1).recv();
 207     }
 208
 209     #[test]
 210     fn period() {
 211         let mut timer = TimerWatcher::new(local_loop());
 212         let port = timer.period(1);
 213         port.recv();
 214         port.recv();
 215         let port2 = timer.period(1);
 216         port2.recv();
 217         port2.recv();
 218     }
 219
 220     #[test]
 221     fn sleep() {
 222         let mut timer = TimerWatcher::new(local_loop());
 223         timer.sleep(1);
 224         timer.sleep(1);
 225     }
 226
 227     #[test] #[should_fail]
 228     fn oneshot_fail() {
 229         let mut timer = TimerWatcher::new(local_loop());
 230         let _port = timer.oneshot(1);
 231         fail!();
 232     }
 233
 234     #[test] #[should_fail]
 235     fn period_fail() {
 236         let mut timer = TimerWatcher::new(local_loop());
 237         let _port = timer.period(1);
 238         fail!();
 239     }
 240
 241     #[test] #[should_fail]
 242     fn normal_fail() {
 243         let _timer = TimerWatcher::new(local_loop());
 244         fail!();
 245     }
 246
 247     #[test]
 248     fn closing_channel_during_drop_doesnt_kill_everything() {
 249         // see issue #10375
 250         let mut timer = TimerWatcher::new(local_loop());
 251         let timer_port = timer.period(1000);
 252
 253         spawn(proc() {
 254             let _ = timer_port.recv_opt();
 255         });
 256
 257         // when we drop the TimerWatcher we're going to destroy the channel,
 258         // which must wake up the task on the other end
 259     }
 260
 261     #[test]
 262     fn reset_doesnt_switch_tasks() {
 263         // similar test to the one above.
 264         let mut timer = TimerWatcher::new(local_loop());
 265         let timer_port = timer.period(1000);
 266
 267         spawn(proc() {
 268             let _ = timer_port.recv_opt();
 269         });
 270
 271         drop(timer.oneshot(1));
 272     }
 273     #[test]
 274     fn reset_doesnt_switch_tasks2() {
 275         // similar test to the one above.
 276         let mut timer = TimerWatcher::new(local_loop());
 277         let timer_port = timer.period(1000);
 278
 279         spawn(proc() {
 280             let _ = timer_port.recv_opt();
 281         });
 282
 283         timer.sleep(1);
 284     }
 285
 286     #[test]
 287     fn sender_goes_away_oneshot() {
 288         let port = {
 289             let mut timer = TimerWatcher::new(local_loop());
 290             timer.oneshot(1000)
 291         };
 292         assert_eq!(port.recv_opt(), Err(()));
 293     }
 294
 295     #[test]
 296     fn sender_goes_away_period() {
 297         let port = {
 298             let mut timer = TimerWatcher::new(local_loop());
 299             timer.period(1000)
 300         };
 301         assert_eq!(port.recv_opt(), Err(()));
 302     }
 303
 304     #[test]
 305     fn receiver_goes_away_oneshot() {
 306         let mut timer1 = TimerWatcher::new(local_loop());
 307         drop(timer1.oneshot(1));
 308         let mut timer2 = TimerWatcher::new(local_loop());
 309         // while sleeping, the prevous timer should fire and not have its
 310         // callback do something terrible.
 311         timer2.sleep(2);
 312     }
 313
 314     #[test]
 315     fn receiver_goes_away_period() {
 316         let mut timer1 = TimerWatcher::new(local_loop());
 317         drop(timer1.period(1));
 318         let mut timer2 = TimerWatcher::new(local_loop());
 319         // while sleeping, the prevous timer should fire and not have its
 320         // callback do something terrible.
 321         timer2.sleep(2);
 322     }
 323 }