rust/src/librustuv/timer.rs

// 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);
    }
}