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