1
Fork 0

auto merge of #7916 : olsonjeffery/rust/newrt_timer, r=brson

My first bit of newsched IO work. Pretty simple and limited in scope.

the RtioTimer trait only has a `sleep(msecs: u64)` method, for now. Taking requests on what else ought to be here.

oh yeah: this resolves #6435
This commit is contained in:
bors 2013-07-22 17:28:35 -07:00
commit ff34064aa3
4 changed files with 137 additions and 0 deletions

View file

@ -252,6 +252,7 @@ pub use self::stdio::print;
pub use self::stdio::println;
pub use self::file::FileStream;
pub use self::timer::Timer;
pub use self::net::ip::IpAddr;
pub use self::net::tcp::TcpListener;
pub use self::net::tcp::TcpStream;
@ -296,6 +297,9 @@ mod extensions;
/// Non-I/O things needed by the I/O module
mod support;
/// Basic Timer
pub mod timer;
/// Thread-blocking implementations
pub mod native {
/// Posix file I/O

64
src/libstd/rt/io/timer.rs Normal file
View file

@ -0,0 +1,64 @@
// 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 option::{Option, Some, None};
use result::{Ok, Err};
use rt::io::{io_error};
use rt::rtio::{IoFactory, IoFactoryObject,
RtioTimer, RtioTimerObject};
use rt::local::Local;
pub struct Timer(~RtioTimerObject);
impl Timer {
fn new_on_rt(i: ~RtioTimerObject) -> Timer {
Timer(i)
}
pub fn new() -> Option<Timer> {
let timer = unsafe {
rtdebug!("Timer::init: borrowing io to init timer");
let io = Local::unsafe_borrow::<IoFactoryObject>();
rtdebug!("about to init timer");
(*io).timer_init()
};
match timer {
Ok(t) => Some(Timer::new_on_rt(t)),
Err(ioerr) => {
rtdebug!("Timer::init: failed to init: %?", ioerr);
io_error::cond.raise(ioerr);
None
}
}
}
}
impl RtioTimer for Timer {
fn sleep(&self, msecs: u64) {
(**self).sleep(msecs);
}
}
#[cfg(test)]
mod test {
use super::*;
use rt::test::*;
use option::{Some, None};
#[test]
fn test_io_timer_sleep_simple() {
do run_in_newsched_task {
let timer = Timer::new();
match timer {
Some(t) => t.sleep(1),
None => assert!(false)
}
}
}
}

View file

@ -23,6 +23,7 @@ pub type IoFactoryObject = uvio::UvIoFactory;
pub type RtioTcpStreamObject = uvio::UvTcpStream;
pub type RtioTcpListenerObject = uvio::UvTcpListener;
pub type RtioUdpSocketObject = uvio::UvUdpSocket;
pub type RtioTimerObject = uvio::UvTimer;
pub trait EventLoop {
fn run(&mut self);
@ -46,6 +47,7 @@ pub trait IoFactory {
fn tcp_connect(&mut self, addr: IpAddr) -> Result<~RtioTcpStreamObject, IoError>;
fn tcp_bind(&mut self, addr: IpAddr) -> Result<~RtioTcpListenerObject, IoError>;
fn udp_bind(&mut self, addr: IpAddr) -> Result<~RtioUdpSocketObject, IoError>;
fn timer_init(&mut self) -> Result<~RtioTimerObject, IoError>;
}
pub trait RtioTcpListener : RtioSocket {
@ -84,3 +86,7 @@ pub trait RtioUdpSocket : RtioSocket {
fn hear_broadcasts(&mut self);
fn ignore_broadcasts(&mut self);
}
pub trait RtioTimer {
fn sleep(&self, msecs: u64);
}

View file

@ -280,6 +280,10 @@ impl IoFactory for UvIoFactory {
}
}
}
fn timer_init(&mut self) -> Result<~RtioTimerObject, IoError> {
Ok(~UvTimer(TimerWatcher::new(self.uv_loop())))
}
}
// FIXME #6090: Prefer newtype structs but Drop doesn't work
@ -562,6 +566,48 @@ impl RtioUdpSocket for UvUdpSocket {
fn ignore_broadcasts(&mut self) { fail!(); }
}
pub struct UvTimer(timer::TimerWatcher);
impl UvTimer {
fn new(w: timer::TimerWatcher) -> UvTimer {
UvTimer(w)
}
}
impl Drop for UvTimer {
fn drop(&self) {
rtdebug!("closing UvTimer");
let scheduler = Local::take::<Scheduler>();
do scheduler.deschedule_running_task_and_then |_, task| {
let task_cell = Cell::new(task);
do self.close {
let scheduler = Local::take::<Scheduler>();
scheduler.resume_blocked_task_immediately(task_cell.take());
}
}
}
}
impl RtioTimer for UvTimer {
fn sleep(&self, msecs: u64) {
let scheduler = Local::take::<Scheduler>();
assert!(scheduler.in_task_context());
do scheduler.deschedule_running_task_and_then |sched, task| {
rtdebug!("sleep: entered scheduler context");
assert!(!sched.in_task_context());
let task_cell = Cell::new(task);
let mut watcher = **self;
do watcher.start(msecs, 0) |_, status| {
assert!(status.is_none());
let scheduler = Local::take::<Scheduler>();
scheduler.resume_blocked_task_immediately(task_cell.take());
}
}
let mut w = **self;
w.stop();
}
}
#[test]
fn test_simple_io_no_connect() {
do run_in_newsched_task {
@ -832,3 +878,20 @@ fn test_udp_many_read() {
}
}
}
fn test_timer_sleep_simple_impl() {
unsafe {
let io = Local::unsafe_borrow::<IoFactoryObject>();
let timer = (*io).timer_init();
match timer {
Ok(t) => t.sleep(1),
Err(_) => assert!(false)
}
}
}
#[test]
fn test_timer_sleep_simple() {
do run_in_newsched_task {
test_timer_sleep_simple_impl();
}
}