src/libstd/sys/unix/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 //! Timers for non-Linux/non-Windows OSes
  12 //!
  13 //! This module implements timers with a worker thread, select(), and a lot of
  14 //! witchcraft that turns out to be horribly inaccurate timers. The unfortunate
  15 //! part is that I'm at a loss of what else to do one these OSes. This is also
  16 //! why Linux has a specialized timerfd implementation and windows has its own
  17 //! implementation (they're more accurate than this one).
  18 //!
  19 //! The basic idea is that there is a worker thread that's communicated to via a
  20 //! channel and a pipe, the pipe is used by the worker thread in a select()
  21 //! syscall with a timeout. The timeout is the "next timer timeout" while the
  22 //! channel is used to send data over to the worker thread.
  23 //!
  24 //! Whenever the call to select() times out, then a channel receives a message.
  25 //! Whenever the call returns that the file descriptor has information, then the
  26 //! channel from timers is drained, enqueuing all incoming requests.
  27 //!
  28 //! The actual implementation of the helper thread is a sorted array of
  29 //! timers in terms of target firing date. The target is the absolute time at
  30 //! which the timer should fire. Timers are then re-enqueued after a firing if
  31 //! the repeat boolean is set.
  32 //!
  33 //! Naturally, all this logic of adding times and keeping track of
  34 //! relative/absolute time is a little lossy and not quite exact. I've done the
  35 //! best I could to reduce the amount of calls to 'now()', but there's likely
  36 //! still inaccuracies trickling in here and there.
  37 //!
  38 //! One of the tricky parts of this implementation is that whenever a timer is
  39 //! acted upon, it must cancel whatever the previous action was (if one is
  40 //! active) in order to act like the other implementations of this timer. In
  41 //! order to do this, the timer's inner pointer is transferred to the worker
  42 //! thread. Whenever the timer is modified, it first takes ownership back from
  43 //! the worker thread in order to modify the same data structure. This has the
  44 //! side effect of "cancelling" the previous requests while allowing a
  45 //! re-enqueuing later on.
  46 //!
  47 //! Note that all time units in this file are in *milliseconds*.
  48
  49 #![allow(deprecated)]
  50
  51 use prelude::v1::*;
  52 use self::Req::*;
  53
  54 use old_io::IoResult;
  55 use libc;
  56 use mem;
  57 use os;
  58 use ptr;
  59 use sync::atomic::{self, Ordering};
  60 use sync::mpsc::{channel, Sender, Receiver, TryRecvError};
  61 use sys::c;
  62 use sys::fs::FileDesc;
  63 use sys_common::helper_thread::Helper;
  64
  65 helper_init! { static HELPER: Helper<Req> }
  66
  67 pub trait Callback {
  68     fn call(&mut self);
  69 }
  70
  71 pub struct Timer {
  72     id: uint,
  73     inner: Option<Box<Inner>>,
  74 }
  75
  76 pub struct Inner {
  77     cb: Option<Box<Callback + Send>>,
  78     interval: u64,
  79     repeat: bool,
  80     target: u64,
  81     id: uint,
  82 }
  83
  84 pub enum Req {
  85     // Add a new timer to the helper thread.
  86     NewTimer(Box<Inner>),
  87
  88     // Remove a timer based on its id and then send it back on the channel
  89     // provided
  90     RemoveTimer(uint, Sender<Box<Inner>>),
  91 }
  92
  93 // returns the current time (in milliseconds)
  94 pub fn now() -> u64 {
  95     unsafe {
  96         let mut now: libc::timeval = mem::zeroed();
  97         assert_eq!(c::gettimeofday(&mut now, ptr::null_mut()), 0);
  98         return (now.tv_sec as u64) * 1000 + (now.tv_usec as u64) / 1000;
  99     }
 100 }
 101
 102 fn helper(input: libc::c_int, messages: Receiver<Req>, _: ()) {
 103     let mut set: c::fd_set = unsafe { mem::zeroed() };
 104
 105     let fd = FileDesc::new(input, true);
 106     let mut timeout: libc::timeval = unsafe { mem::zeroed() };
 107
 108     // active timers are those which are able to be selected upon (and it's a
 109     // sorted list, and dead timers are those which have expired, but ownership
 110     // hasn't yet been transferred back to the timer itself.
 111     let mut active: Vec<Box<Inner>> = vec![];
 112     let mut dead = vec![];
 113
 114     // inserts a timer into an array of timers (sorted by firing time)
 115     fn insert(t: Box<Inner>, active: &mut Vec<Box<Inner>>) {
 116         match active.iter().position(|tm| tm.target > t.target) {
 117             Some(pos) => { active.insert(pos, t); }
 118             None => { active.push(t); }
 119         }
 120     }
 121
 122     // signals the first requests in the queue, possible re-enqueueing it.
 123     fn signal(active: &mut Vec<Box<Inner>>,
 124               dead: &mut Vec<(uint, Box<Inner>)>) {
 125         if active.is_empty() { return }
 126
 127         let mut timer = active.remove(0);
 128         let mut cb = timer.cb.take().unwrap();
 129         cb.call();
 130         if timer.repeat {
 131             timer.cb = Some(cb);
 132             timer.target += timer.interval;
 133             insert(timer, active);
 134         } else {
 135             dead.push((timer.id, timer));
 136         }
 137     }
 138
 139     'outer: loop {
 140         let timeout = if active.len() == 0 {
 141             // Empty array? no timeout (wait forever for the next request)
 142             ptr::null_mut()
 143         } else {
 144             let now = now();
 145             // If this request has already expired, then signal it and go
 146             // through another iteration
 147             if active[0].target <= now {
 148                 signal(&mut active, &mut dead);
 149                 continue;
 150             }
 151
 152             // The actual timeout listed in the requests array is an
 153             // absolute date, so here we translate the absolute time to a
 154             // relative time.
 155             let tm = active[0].target - now;
 156             timeout.tv_sec = (tm / 1000) as libc::time_t;
 157             timeout.tv_usec = ((tm % 1000) * 1000) as libc::suseconds_t;
 158             &mut timeout as *mut libc::timeval
 159         };
 160
 161         c::fd_set(&mut set, input);
 162         match unsafe {
 163             c::select(input + 1, &mut set, ptr::null_mut(),
 164                       ptr::null_mut(), timeout)
 165         } {
 166             // timed out
 167             0 => signal(&mut active, &mut dead),
 168
 169             // file descriptor write woke us up, we've got some new requests
 170             1 => {
 171                 loop {
 172                     match messages.try_recv() {
 173                         // Once we've been disconnected it means the main thread
 174                         // is exiting (at_exit has run). We could still have
 175                         // active timers for other threads, so we're just going
 176                         // to drop them all on the floor. This is all we can
 177                         // really do, however, to prevent resource leakage. The
 178                         // remaining timers will likely start panicking quickly
 179                         // as they attempt to re-use this thread but are
 180                         // disallowed to do so.
 181                         Err(TryRecvError::Disconnected) => {
 182                             break 'outer;
 183                         }
 184
 185                         Ok(NewTimer(timer)) => insert(timer, &mut active),
 186
 187                         Ok(RemoveTimer(id, ack)) => {
 188                             match dead.iter().position(|&(i, _)| id == i) {
 189                                 Some(i) => {
 190                                     let (_, i) = dead.remove(i);
 191                                     ack.send(i).unwrap();
 192                                     continue
 193                                 }
 194                                 None => {}
 195                             }
 196                             let i = active.iter().position(|i| i.id == id);
 197                             let i = i.expect("no timer found");
 198                             let t = active.remove(i);
 199                             ack.send(t).unwrap();
 200                         }
 201                         Err(..) => break
 202                     }
 203                 }
 204
 205                 // drain the file descriptor
 206                 let mut buf = [0];
 207                 assert_eq!(fd.read(&mut buf).unwrap(), 1);
 208             }
 209
 210             -1 if os::errno() == libc::EINTR as i32 => {}
 211             n => panic!("helper thread failed in select() with error: {} ({})",
 212                        n, os::last_os_error())
 213         }
 214     }
 215 }
 216
 217 impl Timer {
 218     pub fn new() -> IoResult<Timer> {
 219         // See notes above regarding using int return value
 220         // instead of ()
 221         HELPER.boot(|| {}, helper);
 222
 223         static ID: atomic::AtomicUsize = atomic::ATOMIC_USIZE_INIT;
 224         let id = ID.fetch_add(1, Ordering::Relaxed);
 225         Ok(Timer {
 226             id: id,
 227             inner: Some(box Inner {
 228                 cb: None,
 229                 interval: 0,
 230                 target: 0,
 231                 repeat: false,
 232                 id: id,
 233             })
 234         })
 235     }
 236
 237     pub fn sleep(&mut self, ms: u64) {
 238         let mut inner = self.inner();
 239         inner.cb = None; // cancel any previous request
 240         self.inner = Some(inner);
 241
 242         let mut to_sleep = libc::timespec {
 243             tv_sec: (ms / 1000) as libc::time_t,
 244             tv_nsec: ((ms % 1000) * 1000000) as libc::c_long,
 245         };
 246         while unsafe { libc::nanosleep(&to_sleep, &mut to_sleep) } != 0 {
 247             if os::errno() as int != libc::EINTR as int {
 248                 panic!("failed to sleep, but not because of EINTR?");
 249             }
 250         }
 251     }
 252
 253     pub fn oneshot(&mut self, msecs: u64, cb: Box<Callback + Send>) {
 254         let now = now();
 255         let mut inner = self.inner();
 256
 257         inner.repeat = false;
 258         inner.cb = Some(cb);
 259         inner.interval = msecs;
 260         inner.target = now + msecs;
 261
 262         HELPER.send(NewTimer(inner));
 263     }
 264
 265     pub fn period(&mut self, msecs: u64, cb: Box<Callback + Send>) {
 266         let now = now();
 267         let mut inner = self.inner();
 268
 269         inner.repeat = true;
 270         inner.cb = Some(cb);
 271         inner.interval = msecs;
 272         inner.target = now + msecs;
 273
 274         HELPER.send(NewTimer(inner));
 275     }
 276
 277     fn inner(&mut self) -> Box<Inner> {
 278         match self.inner.take() {
 279             Some(i) => i,
 280             None => {
 281                 let (tx, rx) = channel();
 282                 HELPER.send(RemoveTimer(self.id, tx));
 283                 rx.recv().unwrap()
 284             }
 285         }
 286     }
 287 }
 288
 289 impl Drop for Timer {
 290     fn drop(&mut self) {
 291         self.inner = Some(self.inner());
 292     }
 293 }