nordic-dev.net/rust
nordic-dev.net/rust
2
0
Fork 0
mirror of https://github.com/rust-lang/rust.git synced 2025-04-16 22:16:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

auto merge of #10321 : alexcrichton/rust/uv-rewrite, r=brson

Browse Source 
The major impetus for this pull request was to remove all usage of `~fn()` in `librustuv`. This construct is going away as a language feature, and additionally it imposes the requirement that all I/O operations have at least one allocation. This allocation has been seen to have a fairly high performance impact in profiles of I/O benchmarks.

I've migrated `librustuv` away from all usage of `~fn()`, and at the same time it no longer allocates on every I/O operation anywhere. The scheduler is now much more tightly integrated with all of the libuv bindings and most of the uv callbacks are specialized functions for a certain procedure. This is a step backwards in terms of making `librustuv` usable anywhere else, but I think that the performance gains are a big win here.

In just a simple benchmark of reading/writing 4k of 0s at a time between a tcp client/server in separate processes on the same system, I have witnessed the throughput increase from ~750MB/s to ~1200MB/s with this change applied.

I'm still in the process of testing this change, although all the major bugs (to my knowledge) have been fleshed out and removed. There are still a few spurious segfaults, and that's what I'm currently investigating. In the meantime, I wanted to put this up for review to get some eyes on it other than mine. I'll update this once I've got all the tests passing reliably again.
This commit is contained in:
bors 2013-11-10 12:26:10 -08:00
commit b5e602ac56
35 changed files with 3699 additions and 5741 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
mk/rt.mk
View File

@ -207,7 +207,7 @@ LIBUV_MAKEFILE_$(1) := $$(CFG_BUILD_DIR)$$(RT_OUTPUT_DIR_$(1))/libuv/Makefile
$$(LIBUV_MAKEFILE_$(1)): $$(LIBUV_DEPS)
(cd $(S)src/libuv/ && \
$$(CFG_PYTHON) ./gyp_uv -f make -Dtarget_arch=$$(LIBUV_ARCH_$(1)) \
$$(CFG_PYTHON) ./gyp_uv.py -f make -Dtarget_arch=$$(LIBUV_ARCH_$(1)) \
-D ninja \
-DOS=$$(LIBUV_OSTYPE_$(1)) \
-Goutput_dir=$$(@D) --generator-output $$(@D))
@ -218,7 +218,7 @@ $$(LIBUV_MAKEFILE_$(1)): $$(LIBUV_DEPS)
ifdef CFG_WINDOWSY_$(1)
$$(LIBUV_LIB_$(1)): $$(LIBUV_DEPS)
$$(Q)$$(MAKE) -C $$(S)src/libuv -f Makefile.mingw \
CFLAGS="$$(CFG_GCCISH_CFLAGS) $$(LIBUV_FLAGS_$$(HOST_$(1))) $$(SNAP_DEFINES)" \
CC="$$(CC) $$(CFG_GCCISH_CFLAGS) $$(LIBUV_FLAGS_$$(HOST_$(1))) $$(SNAP_DEFINES)" \
AR="$$(AR_$(1))" \
V=$$(VERBOSE)
$$(Q)cp $$(S)src/libuv/libuv.a $$@

205
src/librustuv/addrinfo.rs
View File

@ -8,41 +8,34 @@
// option. This file may not be copied, modified, or distributed
// except according to those terms.
use std::cast::transmute;
use std::cell::Cell;
use std::libc::{c_int, c_void};
use std::ptr::null;
use ai = std::rt::io::net::addrinfo;
use std::libc::c_int;
use std::ptr::null;
use std::rt::BlockedTask;
use std::rt::local::Local;
use std::rt::sched::Scheduler;
use uvll;
use uvll::UV_GETADDRINFO;
use super::{Loop, UvError, NativeHandle, status_to_maybe_uv_error};
use net;
use super::{Loop, UvError, Request, wait_until_woken_after};
use uvll;
type GetAddrInfoCallback = ~fn(GetAddrInfoRequest, &net::UvAddrInfo, Option<UvError>);
pub struct GetAddrInfoRequest(*uvll::uv_getaddrinfo_t);
pub struct RequestData {
priv getaddrinfo_cb: Option<GetAddrInfoCallback>,
struct Addrinfo {
handle: *uvll::addrinfo,
}
struct Ctx {
slot: Option<BlockedTask>,
status: c_int,
addrinfo: Option<Addrinfo>,
}
pub struct GetAddrInfoRequest;
impl GetAddrInfoRequest {
pub fn new() -> GetAddrInfoRequest {
let req = unsafe { uvll::malloc_req(UV_GETADDRINFO) };
assert!(req.is_not_null());
let mut req: GetAddrInfoRequest = NativeHandle::from_native_handle(req);
req.install_req_data();
return req;
}
pub fn getaddrinfo(&mut self, loop_: &Loop, node: Option<&str>,
service: Option<&str>, hints: Option<ai::Hint>,
cb: GetAddrInfoCallback) {
pub fn run(loop_: &Loop, node: Option<&str>, service: Option<&str>,
hints: Option<ai::Hint>) -> Result<~[ai::Info], UvError> {
assert!(node.is_some() || service.is_some());
let (c_node, c_node_ptr) = match node {
let (_c_node, c_node_ptr) = match node {
Some(n) => {
let c_node = n.to_c_str();
let c_node_ptr = c_node.with_ref(|r| r);
@ -51,7 +44,7 @@ impl GetAddrInfoRequest {
None => (None, null())
};
let (c_service, c_service_ptr) = match service {
let (_c_service, c_service_ptr) = match service {
Some(s) => {
let c_service = s.to_c_str();
let c_service_ptr = c_service.with_ref(|r| r);
@ -60,17 +53,6 @@ impl GetAddrInfoRequest {
None => (None, null())
};
let cb = Cell::new(cb);
let wrapper_cb: GetAddrInfoCallback = |req, addrinfo, err| {
// Capture some heap values that need to stay alive for the
// getaddrinfo call
let _ = &c_node;
let _ = &c_service;
let cb = cb.take();
cb(req, addrinfo, err)
};
let hint = hints.map(|hint| {
let mut flags = 0;
do each_ai_flag |cval, aival| {
@ -78,19 +60,6 @@ impl GetAddrInfoRequest {
flags |= cval as i32;
}
}
/* XXX: do we really want to support these?
let socktype = match hint.socktype {
Some(ai::Stream) => uvll::rust_SOCK_STREAM(),
Some(ai::Datagram) => uvll::rust_SOCK_DGRAM(),
Some(ai::Raw) => uvll::rust_SOCK_RAW(),
None => 0,
};
let protocol = match hint.protocol {
Some(ai::UDP) => uvll::rust_IPPROTO_UDP(),
Some(ai::TCP) => uvll::rust_IPPROTO_TCP(),
_ => 0,
};
*/
let socktype = 0;
let protocol = 0;
@ -106,66 +75,48 @@ impl GetAddrInfoRequest {
}
});
let hint_ptr = hint.as_ref().map_default(null(), |x| x as *uvll::addrinfo);
let mut req = Request::new(uvll::UV_GETADDRINFO);
self.get_req_data().getaddrinfo_cb = Some(wrapper_cb);
return match unsafe {
uvll::uv_getaddrinfo(loop_.handle, req.handle,
getaddrinfo_cb, c_node_ptr, c_service_ptr,
hint_ptr)
} {
0 => {
req.defuse(); // uv callback now owns this request
let mut cx = Ctx { slot: None, status: 0, addrinfo: None };
unsafe {
assert!(0 == uvll::getaddrinfo(loop_.native_handle(),
self.native_handle(),
getaddrinfo_cb,
c_node_ptr,
c_service_ptr,
hint_ptr));
}
do wait_until_woken_after(&mut cx.slot) {
req.set_data(&cx);
}
extern "C" fn getaddrinfo_cb(req: *uvll::uv_getaddrinfo_t,
status: c_int,
res: *uvll::addrinfo) {
let mut req: GetAddrInfoRequest = NativeHandle::from_native_handle(req);
let err = status_to_maybe_uv_error(status);
let addrinfo = net::UvAddrInfo(res);
let data = req.get_req_data();
(*data.getaddrinfo_cb.get_ref())(req, &addrinfo, err);
unsafe {
uvll::freeaddrinfo(res);
match cx.status {
0 => Ok(accum_addrinfo(cx.addrinfo.get_ref())),
n => Err(UvError(n))
}
}
}
}
fn get_loop(&self) -> Loop {
unsafe {
Loop {
handle: uvll::get_loop_from_fs_req(self.native_handle())
}
}
}
fn install_req_data(&mut self) {
let req = self.native_handle() as *uvll::uv_getaddrinfo_t;
let data = ~RequestData {
getaddrinfo_cb: None
n => Err(UvError(n))
};
unsafe {
let data = transmute::<~RequestData, *c_void>(data);
uvll::set_data_for_req(req, data);
extern fn getaddrinfo_cb(req: *uvll::uv_getaddrinfo_t,
status: c_int,
res: *uvll::addrinfo) {
let req = Request::wrap(req);
assert!(status != uvll::ECANCELED);
let cx: &mut Ctx = unsafe { req.get_data() };
cx.status = status;
cx.addrinfo = Some(Addrinfo { handle: res });
let sched: ~Scheduler = Local::take();
sched.resume_blocked_task_immediately(cx.slot.take_unwrap());
}
}
}
fn get_req_data<'r>(&'r mut self) -> &'r mut RequestData {
unsafe {
let data = uvll::get_data_for_req(self.native_handle());
let data = transmute::<&*c_void, &mut ~RequestData>(&data);
return &mut **data;
}
}
fn delete(self) {
unsafe {
let data = uvll::get_data_for_req(self.native_handle());
let _data = transmute::<*c_void, ~RequestData>(data);
uvll::set_data_for_req(self.native_handle(), null::<()>());
uvll::free_req(self.native_handle());
}
impl Drop for Addrinfo {
fn drop(&mut self) {
unsafe { uvll::uv_freeaddrinfo(self.handle) }
}
}
@ -184,15 +135,13 @@ fn each_ai_flag(_f: &fn(c_int, ai::Flag)) {
}
// Traverse the addrinfo linked list, producing a vector of Rust socket addresses
pub fn accum_addrinfo(addr: &net::UvAddrInfo) -> ~[ai::Info] {
pub fn accum_addrinfo(addr: &Addrinfo) -> ~[ai::Info] {
unsafe {
let &net::UvAddrInfo(addr) = addr;
let mut addr = addr;
let mut addr = addr.handle;
let mut addrs = ~[];
loop {
let uvaddr = net::sockaddr_to_UvSocketAddr((*addr).ai_addr);
let rustaddr = net::uv_socket_addr_to_socket_addr(uvaddr);
let rustaddr = net::sockaddr_to_socket_addr((*addr).ai_addr);
let mut flags = 0;
do each_ai_flag |cval, aival| {
@ -235,39 +184,27 @@ pub fn accum_addrinfo(addr: &net::UvAddrInfo) -> ~[ai::Info] {
}
}
impl NativeHandle<*uvll::uv_getaddrinfo_t> for GetAddrInfoRequest {
fn from_native_handle(handle: *uvll::uv_getaddrinfo_t) -> GetAddrInfoRequest {
GetAddrInfoRequest(handle)
}
fn native_handle(&self) -> *uvll::uv_getaddrinfo_t {
match self { &GetAddrInfoRequest(ptr) => ptr }
}
}
#[cfg(test)]
mod test {
use Loop;
use std::rt::io::net::ip::{SocketAddr, Ipv4Addr};
use super::*;
use super::super::local_loop;
#[test]
fn getaddrinfo_test() {
let mut loop_ = Loop::new();
let mut req = GetAddrInfoRequest::new();
do req.getaddrinfo(&loop_, Some("localhost"), None, None) |_, addrinfo, _| {
let sockaddrs = accum_addrinfo(addrinfo);
let mut found_local = false;
let local_addr = &SocketAddr {
ip: Ipv4Addr(127, 0, 0, 1),
port: 0
};
for addr in sockaddrs.iter() {
found_local = found_local || addr.address == *local_addr;
match GetAddrInfoRequest::run(local_loop(), Some("localhost"), None, None) {
Ok(infos) => {
let mut found_local = false;
let local_addr = &SocketAddr {
ip: Ipv4Addr(127, 0, 0, 1),
port: 0
};
for addr in infos.iter() {
found_local = found_local || addr.address == *local_addr;
}
assert!(found_local);
}
assert!(found_local);
Err(e) => fail!("{:?}", e),
}
loop_.run();
loop_.close();
req.delete();
}
}

177
src/librustuv/async.rs
View File

@ -8,76 +8,155 @@
// option. This file may not be copied, modified, or distributed
// except according to those terms.
use std::cast;
use std::libc::c_int;
use std::rt::rtio::{Callback, RemoteCallback};
use std::unstable::sync::Exclusive;
use uvll;
use super::{Watcher, Loop, NativeHandle, AsyncCallback, WatcherInterop};
use super::status_to_maybe_uv_error;
use super::{Loop, UvHandle};
pub struct AsyncWatcher(*uvll::uv_async_t);
impl Watcher for AsyncWatcher { }
// The entire point of async is to call into a loop from other threads so it
// does not need to home.
pub struct AsyncWatcher {
handle: *uvll::uv_async_t,
// A flag to tell the callback to exit, set from the dtor. This is
// almost never contested - only in rare races with the dtor.
exit_flag: Exclusive<bool>
}
struct Payload {
callback: ~Callback,
exit_flag: Exclusive<bool>,
}
impl AsyncWatcher {
pub fn new(loop_: &mut Loop, cb: AsyncCallback) -> AsyncWatcher {
pub fn new(loop_: &mut Loop, cb: ~Callback) -> AsyncWatcher {
let handle = UvHandle::alloc(None::<AsyncWatcher>, uvll::UV_ASYNC);
assert_eq!(unsafe {
uvll::uv_async_init(loop_.handle, handle, async_cb)
}, 0);
let flag = Exclusive::new(false);
let payload = ~Payload { callback: cb, exit_flag: flag.clone() };
unsafe {
let handle = uvll::malloc_handle(uvll::UV_ASYNC);
assert!(handle.is_not_null());
let mut watcher: AsyncWatcher = NativeHandle::from_native_handle(handle);
watcher.install_watcher_data();
let data = watcher.get_watcher_data();
data.async_cb = Some(cb);
assert_eq!(0, uvll::async_init(loop_.native_handle(), handle, async_cb));
return watcher;
}
extern fn async_cb(handle: *uvll::uv_async_t, status: c_int) {
let mut watcher: AsyncWatcher = NativeHandle::from_native_handle(handle);
let status = status_to_maybe_uv_error(status);
let data = watcher.get_watcher_data();
let cb = data.async_cb.get_ref();
(*cb)(watcher, status);
}
}
pub fn send(&mut self) {
unsafe {
let handle = self.native_handle();
uvll::async_send(handle);
let payload: *u8 = cast::transmute(payload);
uvll::set_data_for_uv_handle(handle, payload);
}
return AsyncWatcher { handle: handle, exit_flag: flag, };
}
}
impl NativeHandle<*uvll::uv_async_t> for AsyncWatcher {
fn from_native_handle(handle: *uvll::uv_async_t) -> AsyncWatcher {
AsyncWatcher(handle)
impl UvHandle<uvll::uv_async_t> for AsyncWatcher {
fn uv_handle(&self) -> *uvll::uv_async_t { self.handle }
unsafe fn from_uv_handle<'a>(_: &'a *uvll::uv_async_t) -> &'a mut AsyncWatcher {
fail!("async watchers can't be built from their handles");
}
fn native_handle(&self) -> *uvll::uv_async_t {
match self { &AsyncWatcher(ptr) => ptr }
}
extern fn async_cb(handle: *uvll::uv_async_t, status: c_int) {
assert!(status == 0);
let payload: &mut Payload = unsafe {
cast::transmute(uvll::get_data_for_uv_handle(handle))
};
// The synchronization logic here is subtle. To review,
// the uv async handle type promises that, after it is
// triggered the remote callback is definitely called at
// least once. UvRemoteCallback needs to maintain those
// semantics while also shutting down cleanly from the
// dtor. In our case that means that, when the
// UvRemoteCallback dtor calls `async.send()`, here `f` is
// always called later.
// In the dtor both the exit flag is set and the async
// callback fired under a lock. Here, before calling `f`,
// we take the lock and check the flag. Because we are
// checking the flag before calling `f`, and the flag is
// set under the same lock as the send, then if the flag
// is set then we're guaranteed to call `f` after the
// final send.
// If the check was done after `f()` then there would be a
// period between that call and the check where the dtor
// could be called in the other thread, missing the final
// callback while still destroying the handle.
let should_exit = unsafe {
payload.exit_flag.with_imm(|&should_exit| should_exit)
};
payload.callback.call();
if should_exit {
unsafe { uvll::uv_close(handle, close_cb) }
}
}
extern fn close_cb(handle: *uvll::uv_handle_t) {
// drop the payload
let _payload: ~Payload = unsafe {
cast::transmute(uvll::get_data_for_uv_handle(handle))
};
// and then free the handle
unsafe { uvll::free_handle(handle) }
}
impl RemoteCallback for AsyncWatcher {
fn fire(&mut self) {
unsafe { uvll::uv_async_send(self.handle) }
}
}
impl Drop for AsyncWatcher {
fn drop(&mut self) {
unsafe {
do self.exit_flag.with |should_exit| {
// NB: These two things need to happen atomically. Otherwise
// the event handler could wake up due to a *previous*
// signal and see the exit flag, destroying the handle
// before the final send.
*should_exit = true;
uvll::uv_async_send(self.handle)
}
}
}
}
#[cfg(test)]
mod test {
mod test_remote {
use std::cell::Cell;
use std::rt::rtio::Callback;
use std::rt::thread::Thread;
use std::rt::tube::Tube;
use super::*;
use Loop;
use std::unstable::run_in_bare_thread;
use std::rt::thread::Thread;
use std::cell::Cell;
use super::super::local_loop;
// Make sure that we can fire watchers in remote threads and that they
// actually trigger what they say they will.
#[test]
fn smoke_test() {
do run_in_bare_thread {
let mut loop_ = Loop::new();
let watcher = AsyncWatcher::new(&mut loop_, |w, _| w.close(||()) );
let watcher_cell = Cell::new(watcher);
let thread = do Thread::start {
let mut watcher = watcher_cell.take();
watcher.send();
};
loop_.run();
loop_.close();
thread.join();
struct MyCallback(Option<Tube<int>>);
impl Callback for MyCallback {
fn call(&mut self) {
// this can get called more than once, but we only want to send
// once
if self.is_some() {
self.take_unwrap().send(1);
}
}
}
let mut tube = Tube::new();
let cb = ~MyCallback(Some(tube.clone()));
let watcher = Cell::new(AsyncWatcher::new(local_loop(), cb as ~Callback));
let thread = do Thread::start {
watcher.take().fire();
};
assert_eq!(tube.recv(), 1);
thread.join();
}
}

1041
src/librustuv/file.rs
View File

File diff suppressed because it is too large Load Diff

215
src/librustuv/idle.rs
View File

@ -8,130 +8,161 @@
// option. This file may not be copied, modified, or distributed
// except according to those terms.
use std::libc::c_int;
use std::cast;
use std::libc::{c_int, c_void};
use uvll;
use super::{Watcher, Loop, NativeHandle, IdleCallback, status_to_maybe_uv_error};
use super::{Loop, UvHandle};
use std::rt::rtio::{Callback, PausibleIdleCallback};
pub struct IdleWatcher(*uvll::uv_idle_t);
impl Watcher for IdleWatcher { }
pub struct IdleWatcher {
handle: *uvll::uv_idle_t,
idle_flag: bool,
closed: bool,
callback: ~Callback,
}
impl IdleWatcher {
pub fn new(loop_: &mut Loop) -> IdleWatcher {
unsafe {
let handle = uvll::malloc_handle(uvll::UV_IDLE);
assert!(handle.is_not_null());
assert_eq!(uvll::idle_init(loop_.native_handle(), handle), 0);
let mut watcher: IdleWatcher = NativeHandle::from_native_handle(handle);
watcher.install_watcher_data();
return watcher
}
pub fn new(loop_: &mut Loop, cb: ~Callback) -> ~IdleWatcher {
let handle = UvHandle::alloc(None::<IdleWatcher>, uvll::UV_IDLE);
assert_eq!(unsafe {
uvll::uv_idle_init(loop_.handle, handle)
}, 0);
let me = ~IdleWatcher {
handle: handle,
idle_flag: false,
closed: false,
callback: cb,
};
return me.install();
}
pub fn start(&mut self, cb: IdleCallback) {
{
let data = self.get_watcher_data();
data.idle_cb = Some(cb);
pub fn onetime(loop_: &mut Loop, f: proc()) {
let handle = UvHandle::alloc(None::<IdleWatcher>, uvll::UV_IDLE);
unsafe {
assert_eq!(uvll::uv_idle_init(loop_.handle, handle), 0);
let data: *c_void = cast::transmute(~f);
uvll::set_data_for_uv_handle(handle, data);
assert_eq!(uvll::uv_idle_start(handle, onetime_cb), 0)
}
unsafe {
assert_eq!(uvll::idle_start(self.native_handle(), idle_cb), 0)
extern fn onetime_cb(handle: *uvll::uv_idle_t, status: c_int) {
assert_eq!(status, 0);
unsafe {
let data = uvll::get_data_for_uv_handle(handle);
let f: ~proc() = cast::transmute(data);
(*f)();
uvll::uv_idle_stop(handle);
uvll::uv_close(handle, close_cb);
}
}
}
pub fn restart(&mut self) {
unsafe {
assert!(self.get_watcher_data().idle_cb.is_some());
assert_eq!(uvll::idle_start(self.native_handle(), idle_cb), 0)
}
}
pub fn stop(&mut self) {
// NB: Not resetting the Rust idle_cb to None here because `stop` is
// likely called from *within* the idle callback, causing a use after
// free
unsafe {
assert_eq!(uvll::idle_stop(self.native_handle()), 0);
extern fn close_cb(handle: *uvll::uv_handle_t) {
unsafe { uvll::free_handle(handle) }
}
}
}
impl NativeHandle<*uvll::uv_idle_t> for IdleWatcher {
fn from_native_handle(handle: *uvll::uv_idle_t) -> IdleWatcher {
IdleWatcher(handle)
impl PausibleIdleCallback for IdleWatcher {
fn pause(&mut self) {
if self.idle_flag == true {
assert_eq!(unsafe {uvll::uv_idle_stop(self.handle) }, 0);
self.idle_flag = false;
}
}
fn native_handle(&self) -> *uvll::uv_idle_t {
match self { &IdleWatcher(ptr) => ptr }
fn resume(&mut self) {
if self.idle_flag == false {
assert_eq!(unsafe { uvll::uv_idle_start(self.handle, idle_cb) }, 0)
self.idle_flag = true;
}
}
}
impl UvHandle<uvll::uv_idle_t> for IdleWatcher {
fn uv_handle(&self) -> *uvll::uv_idle_t { self.handle }
}
extern fn idle_cb(handle: *uvll::uv_idle_t, status: c_int) {
let mut idle_watcher: IdleWatcher = NativeHandle::from_native_handle(handle);
let data = idle_watcher.get_watcher_data();
let cb: &IdleCallback = data.idle_cb.get_ref();
let status = status_to_maybe_uv_error(status);
(*cb)(idle_watcher, status);
assert_eq!(status, 0);
let idle: &mut IdleWatcher = unsafe { UvHandle::from_uv_handle(&handle) };
idle.callback.call();
}
impl Drop for IdleWatcher {
fn drop(&mut self) {
self.pause();
self.close_async_();
}
}
#[cfg(test)]
mod test {
use Loop;
use super::*;
use std::unstable::run_in_bare_thread;
use std::rt::tube::Tube;
use std::rt::rtio::{Callback, PausibleIdleCallback};
use super::super::local_loop;
#[test]
#[ignore(reason = "valgrind - loop destroyed before watcher?")]
fn idle_new_then_close() {
do run_in_bare_thread {
let mut loop_ = Loop::new();
let idle_watcher = { IdleWatcher::new(&mut loop_) };
idle_watcher.close(||());
struct MyCallback(Tube<int>, int);
impl Callback for MyCallback {
fn call(&mut self) {
match *self {
MyCallback(ref mut tube, val) => tube.send(val)
}
}
}
#[test]
fn idle_smoke_test() {
do run_in_bare_thread {
let mut loop_ = Loop::new();
let mut idle_watcher = { IdleWatcher::new(&mut loop_) };
let mut count = 10;
let count_ptr: *mut int = &mut count;
do idle_watcher.start |idle_watcher, status| {
let mut idle_watcher = idle_watcher;
assert!(status.is_none());
if unsafe { *count_ptr == 10 } {
idle_watcher.stop();
idle_watcher.close(||());
} else {
unsafe { *count_ptr = *count_ptr + 1; }
}
}
loop_.run();
loop_.close();
assert_eq!(count, 10);
}
fn not_used() {
let cb = ~MyCallback(Tube::new(), 1);
let _idle = IdleWatcher::new(local_loop(), cb as ~Callback);
}
#[test]
fn idle_start_stop_start() {
do run_in_bare_thread {
let mut loop_ = Loop::new();
let mut idle_watcher = { IdleWatcher::new(&mut loop_) };
do idle_watcher.start |idle_watcher, status| {
let mut idle_watcher = idle_watcher;
assert!(status.is_none());
idle_watcher.stop();
do idle_watcher.start |idle_watcher, status| {
assert!(status.is_none());
let mut idle_watcher = idle_watcher;
idle_watcher.stop();
idle_watcher.close(||());
}
}
loop_.run();
loop_.close();
}
fn smoke_test() {
let mut tube = Tube::new();
let cb = ~MyCallback(tube.clone(), 1);
let mut idle = IdleWatcher::new(local_loop(), cb as ~Callback);
idle.resume();
tube.recv();
}
#[test] #[should_fail]
fn smoke_fail() {
let tube = Tube::new();
let cb = ~MyCallback(tube.clone(), 1);
let mut idle = IdleWatcher::new(local_loop(), cb as ~Callback);
idle.resume();
fail!();
}
#[test]
fn fun_combinations_of_methods() {
let mut tube = Tube::new();
let cb = ~MyCallback(tube.clone(), 1);
let mut idle = IdleWatcher::new(local_loop(), cb as ~Callback);
idle.resume();
tube.recv();
idle.pause();
idle.resume();
idle.resume();
tube.recv();
idle.pause();
idle.pause();
idle.resume();
tube.recv();
}
#[test]
fn pause_pauses() {
let mut tube = Tube::new();
let cb = ~MyCallback(tube.clone(), 1);
let mut idle1 = IdleWatcher::new(local_loop(), cb as ~Callback);
let cb = ~MyCallback(tube.clone(), 2);
let mut idle2 = IdleWatcher::new(local_loop(), cb as ~Callback);
idle2.resume();
assert_eq!(tube.recv(), 2);
idle2.pause();
idle1.resume();
assert_eq!(tube.recv(), 1);
}
}

463
src/librustuv/lib.rs
View File

@ -45,29 +45,31 @@ via `close` and `delete` methods.
#[feature(macro_rules, globs)];
use std::str::raw::from_c_str;
use std::vec;
use std::ptr;
use std::str;
use std::libc::{c_void, c_int, size_t, malloc, free};
use std::cast::transmute;
use std::cast;
use std::libc::{c_int, malloc};
use std::ptr::null;
use std::ptr;
use std::rt::BlockedTask;
use std::rt::local::Local;
use std::rt::sched::Scheduler;
use std::str::raw::from_c_str;
use std::str;
use std::task;
use std::unstable::finally::Finally;
use std::rt::io::net::ip::SocketAddr;
use std::rt::io::signal::Signum;
use std::vec;
use std::rt::io::IoError;
//#[cfg(test)] use unstable::run_in_bare_thread;
pub use self::file::{FsRequest};
pub use self::net::{StreamWatcher, TcpWatcher, UdpWatcher};
pub use self::idle::IdleWatcher;
pub use self::timer::TimerWatcher;
pub use self::async::AsyncWatcher;
pub use self::file::{FsRequest, FileWatcher};
pub use self::idle::IdleWatcher;
pub use self::net::{TcpWatcher, TcpListener, TcpAcceptor, UdpWatcher};
pub use self::pipe::{PipeWatcher, PipeListener, PipeAcceptor};
pub use self::process::Process;
pub use self::pipe::Pipe;
pub use self::signal::SignalWatcher;
pub use self::timer::TimerWatcher;
pub use self::tty::TtyWatcher;
mod macros;
@ -87,6 +89,171 @@ pub mod process;
pub mod pipe;
pub mod tty;
pub mod signal;
pub mod stream;
/// A type that wraps a uv handle
pub trait UvHandle<T> {
fn uv_handle(&self) -> *T;
// FIXME(#8888) dummy self
fn alloc(_: Option<Self>, ty: uvll::uv_handle_type) -> *T {
unsafe {
let handle = uvll::malloc_handle(ty);
assert!(!handle.is_null());
handle as *T
}
}
unsafe fn from_uv_handle<'a>(h: &'a *T) -> &'a mut Self {
cast::transmute(uvll::get_data_for_uv_handle(*h))
}
fn install(~self) -> ~Self {
unsafe {
let myptr = cast::transmute::<&~Self, &*u8>(&self);
uvll::set_data_for_uv_handle(self.uv_handle(), *myptr);
}
self
}
fn close_async_(&mut self) {
// we used malloc to allocate all handles, so we must always have at
// least a callback to free all the handles we allocated.
extern fn close_cb(handle: *uvll::uv_handle_t) {
unsafe { uvll::free_handle(handle) }
}
unsafe {
uvll::set_data_for_uv_handle(self.uv_handle(), null::<()>());
uvll::uv_close(self.uv_handle() as *uvll::uv_handle_t, close_cb)
}
}
fn close(&mut self) {
let mut slot = None;
unsafe {
uvll::uv_close(self.uv_handle() as *uvll::uv_handle_t, close_cb);
uvll::set_data_for_uv_handle(self.uv_handle(), ptr::null::<()>());
do wait_until_woken_after(&mut slot) {
uvll::set_data_for_uv_handle(self.uv_handle(), &slot);
}
}
extern fn close_cb(handle: *uvll::uv_handle_t) {
unsafe {
let data = uvll::get_data_for_uv_handle(handle);
uvll::free_handle(handle);
if data == ptr::null() { return }
let slot: &mut Option<BlockedTask> = cast::transmute(data);
let sched: ~Scheduler = Local::take();
sched.resume_blocked_task_immediately(slot.take_unwrap());
}
}
}
}
pub struct ForbidSwitch {
msg: &'static str,
sched: uint,
}
impl ForbidSwitch {
fn new(s: &'static str) -> ForbidSwitch {
ForbidSwitch {
msg: s, sched: Local::borrow(|s: &mut Scheduler| s.sched_id())
}
}
}
impl Drop for ForbidSwitch {
fn drop(&mut self) {
assert!(self.sched == Local::borrow(|s: &mut Scheduler| s.sched_id()),
"didnt want a scheduler switch: {}", self.msg);
}
}
pub struct ForbidUnwind {
msg: &'static str,
failing_before: bool,
}
impl ForbidUnwind {
fn new(s: &'static str) -> ForbidUnwind {
ForbidUnwind {
msg: s, failing_before: task::failing(),
}
}
}
impl Drop for ForbidUnwind {
fn drop(&mut self) {
assert!(self.failing_before == task::failing(),
"didnt want an unwind during: {}", self.msg);
}
}
fn wait_until_woken_after(slot: *mut Option<BlockedTask>, f: &fn()) {
let _f = ForbidUnwind::new("wait_until_woken_after");
unsafe {
assert!((*slot).is_none());
let sched: ~Scheduler = Local::take();
do sched.deschedule_running_task_and_then |_, task| {
f();
*slot = Some(task);
}
}
}
pub struct Request {
handle: *uvll::uv_req_t,
priv defused: bool,
}
impl Request {
pub fn new(ty: uvll::uv_req_type) -> Request {
unsafe {
let handle = uvll::malloc_req(ty);
uvll::set_data_for_req(handle, null::<()>());
Request::wrap(handle)
}
}
pub fn wrap(handle: *uvll::uv_req_t) -> Request {
Request { handle: handle, defused: false }
}
pub fn set_data<T>(&self, t: *T) {
unsafe { uvll::set_data_for_req(self.handle, t) }
}
pub unsafe fn get_data<T>(&self) -> &'static mut T {
let data = uvll::get_data_for_req(self.handle);
assert!(data != null());
cast::transmute(data)
}
// This function should be used when the request handle has been given to an
// underlying uv function, and the uv function has succeeded. This means
// that uv will at some point invoke the callback, and in the meantime we
// can't deallocate the handle because libuv could be using it.
//
// This is still a problem in blocking situations due to linked failure. In
// the connection callback the handle should be re-wrapped with the `wrap`
// function to ensure its destruction.
pub fn defuse(&mut self) {
self.defused = true;
}
}
impl Drop for Request {
fn drop(&mut self) {
if !self.defused {
unsafe { uvll::free_req(self.handle) }
}
}
}
/// XXX: Loop(*handle) is buggy with destructors. Normal structs
/// with dtors may not be destructured, but tuple structs can,
@ -95,169 +262,21 @@ pub struct Loop {
priv handle: *uvll::uv_loop_t
}
pub struct Handle(*uvll::uv_handle_t);
impl Watcher for Handle {}
impl NativeHandle<*uvll::uv_handle_t> for Handle {
fn from_native_handle(h: *uvll::uv_handle_t) -> Handle { Handle(h) }
fn native_handle(&self) -> *uvll::uv_handle_t { **self }
}
/// The trait implemented by uv 'watchers' (handles). Watchers are
/// non-owning wrappers around the uv handles and are not completely
/// safe - there may be multiple instances for a single underlying
/// handle. Watchers are generally created, then `start`ed, `stop`ed
/// and `close`ed, but due to their complex life cycle may not be
/// entirely memory safe if used in unanticipated patterns.
pub trait Watcher { }
pub trait Request { }
/// A type that wraps a native handle
pub trait NativeHandle<T> {
fn from_native_handle(T) -> Self;
fn native_handle(&self) -> T;
}
impl Loop {
pub fn new() -> Loop {
let handle = unsafe { uvll::loop_new() };
assert!(handle.is_not_null());
NativeHandle::from_native_handle(handle)
Loop::wrap(handle)
}
pub fn wrap(handle: *uvll::uv_loop_t) -> Loop { Loop { handle: handle } }
pub fn run(&mut self) {
unsafe { uvll::run(self.native_handle()) };
unsafe { uvll::uv_run(self.handle, uvll::RUN_DEFAULT) };
}
pub fn close(&mut self) {
unsafe { uvll::loop_delete(self.native_handle()) };
}
}
impl NativeHandle<*uvll::uv_loop_t> for Loop {
fn from_native_handle(handle: *uvll::uv_loop_t) -> Loop {
Loop { handle: handle }
}
fn native_handle(&self) -> *uvll::uv_loop_t {
self.handle
}
}
// XXX: The uv alloc callback also has a *uv_handle_t arg
pub type AllocCallback = ~fn(uint) -> Buf;
pub type ReadCallback = ~fn(StreamWatcher, int, Buf, Option<UvError>);
pub type NullCallback = ~fn();
pub type IdleCallback = ~fn(IdleWatcher, Option<UvError>);
pub type ConnectionCallback = ~fn(StreamWatcher, Option<UvError>);
pub type FsCallback = ~fn(&mut FsRequest, Option<UvError>);
// first int is exit_status, second is term_signal
pub type ExitCallback = ~fn(Process, int, int, Option<UvError>);
pub type TimerCallback = ~fn(TimerWatcher, Option<UvError>);
pub type AsyncCallback = ~fn(AsyncWatcher, Option<UvError>);
pub type UdpReceiveCallback = ~fn(UdpWatcher, int, Buf, SocketAddr, uint, Option<UvError>);
pub type UdpSendCallback = ~fn(UdpWatcher, Option<UvError>);
pub type SignalCallback = ~fn(SignalWatcher, Signum);
/// Callbacks used by StreamWatchers, set as custom data on the foreign handle.
/// XXX: Would be better not to have all watchers allocate room for all callback types.
struct WatcherData {
read_cb: Option<ReadCallback>,
write_cb: Option<ConnectionCallback>,
connect_cb: Option<ConnectionCallback>,
close_cb: Option<NullCallback>,
alloc_cb: Option<AllocCallback>,
idle_cb: Option<IdleCallback>,
timer_cb: Option<TimerCallback>,
async_cb: Option<AsyncCallback>,
udp_recv_cb: Option<UdpReceiveCallback>,
udp_send_cb: Option<UdpSendCallback>,
exit_cb: Option<ExitCallback>,
signal_cb: Option<SignalCallback>,
}
pub trait WatcherInterop {
fn event_loop(&self) -> Loop;
fn install_watcher_data(&mut self);
fn get_watcher_data<'r>(&'r mut self) -> &'r mut WatcherData;
fn drop_watcher_data(&mut self);
fn close(self, cb: NullCallback);
fn close_async(self);
}
impl<H, W: Watcher + NativeHandle<*H>> WatcherInterop for W {
/// Get the uv event loop from a Watcher
fn event_loop(&self) -> Loop {
unsafe {
let handle = self.native_handle();
let loop_ = uvll::get_loop_for_uv_handle(handle);
NativeHandle::from_native_handle(loop_)
}
}
fn install_watcher_data(&mut self) {
unsafe {
let data = ~WatcherData {
read_cb: None,
write_cb: None,
connect_cb: None,
close_cb: None,
alloc_cb: None,
idle_cb: None,
timer_cb: None,
async_cb: None,
udp_recv_cb: None,
udp_send_cb: None,
exit_cb: None,
signal_cb: None,
};
let data = transmute::<~WatcherData, *c_void>(data);
uvll::set_data_for_uv_handle(self.native_handle(), data);
}
}
fn get_watcher_data<'r>(&'r mut self) -> &'r mut WatcherData {
unsafe {
let data = uvll::get_data_for_uv_handle(self.native_handle());
let data = transmute::<&*c_void, &mut ~WatcherData>(&data);
return &mut **data;
}
}
fn drop_watcher_data(&mut self) {
unsafe {
let data = uvll::get_data_for_uv_handle(self.native_handle());
let _data = transmute::<*c_void, ~WatcherData>(data);
uvll::set_data_for_uv_handle(self.native_handle(), null::<()>());
}
}
fn close(mut self, cb: NullCallback) {
{
let data = self.get_watcher_data();
assert!(data.close_cb.is_none());
data.close_cb = Some(cb);
}
unsafe { uvll::close(self.native_handle(), close_cb); }
extern fn close_cb(handle: *uvll::uv_handle_t) {
let mut h: Handle = NativeHandle::from_native_handle(handle);
h.get_watcher_data().close_cb.take_unwrap()();
h.drop_watcher_data();
unsafe { uvll::free_handle(handle as *c_void) }
}
}
fn close_async(self) {
unsafe { uvll::close(self.native_handle(), close_cb); }
extern fn close_cb(handle: *uvll::uv_handle_t) {
let mut h: Handle = NativeHandle::from_native_handle(handle);
h.drop_watcher_data();
unsafe { uvll::free_handle(handle as *c_void) }
}
unsafe { uvll::uv_loop_delete(self.handle) };
}
}
@ -270,7 +289,7 @@ impl UvError {
pub fn name(&self) -> ~str {
unsafe {
let inner = match self { &UvError(a) => a };
let name_str = uvll::err_name(inner);
let name_str = uvll::uv_err_name(inner);
assert!(name_str.is_not_null());
from_c_str(name_str)
}
@ -279,7 +298,7 @@ impl UvError {
pub fn desc(&self) -> ~str {
unsafe {
let inner = match self { &UvError(a) => a };
let desc_str = uvll::strerror(inner);
let desc_str = uvll::uv_strerror(inner);
assert!(desc_str.is_not_null());
from_c_str(desc_str)
}
@ -309,7 +328,7 @@ pub fn uv_error_to_io_error(uverr: UvError) -> IoError {
use std::rt::io::*;
// uv error descriptions are static
let c_desc = uvll::strerror(*uverr);
let c_desc = uvll::uv_strerror(*uverr);
let desc = str::raw::c_str_to_static_slice(c_desc);
let kind = match *uverr {
@ -337,9 +356,8 @@ pub fn uv_error_to_io_error(uverr: UvError) -> IoError {
}
}
/// Given a uv handle, convert a callback status to a UvError
pub fn status_to_maybe_uv_error(status: c_int) -> Option<UvError>
{
/// Given a uv error code, convert a callback status to a UvError
pub fn status_to_maybe_uv_error(status: c_int) -> Option<UvError> {
if status >= 0 {
None
} else {
@ -347,6 +365,10 @@ pub fn status_to_maybe_uv_error(status: c_int) -> Option<UvError>
}
}
pub fn status_to_io_result(status: c_int) -> Result<(), IoError> {
if status >= 0 {Ok(())} else {Err(uv_error_to_io_error(UvError(status)))}
}
/// The uv buffer type
pub type Buf = uvll::uv_buf_t;
@ -360,65 +382,56 @@ pub fn empty_buf() -> Buf {
/// Borrow a slice to a Buf
pub fn slice_to_uv_buf(v: &[u8]) -> Buf {
let data = vec::raw::to_ptr(v);
unsafe { uvll::buf_init(data, v.len()) }
uvll::uv_buf_t { base: data, len: v.len() as uvll::uv_buf_len_t }
}
// XXX: Do these conversions without copying
/// Transmute an owned vector to a Buf
pub fn vec_to_uv_buf(v: ~[u8]) -> Buf {
#[fixed_stack_segment]; #[inline(never)];
#[cfg(test)]
fn local_loop() -> &'static mut Loop {
unsafe {
let data = malloc(v.len() as size_t) as *u8;
assert!(data.is_not_null());
do v.as_imm_buf |b, l| {
let data = data as *mut u8;
ptr::copy_memory(data, b, l)
cast::transmute(do Local::borrow |sched: &mut Scheduler| {
let mut io = None;
do sched.event_loop.io |i| {
let (_vtable, uvio): (uint, &'static mut uvio::UvIoFactory) =
cast::transmute(i);
io = Some(uvio);
}
io.unwrap()
}.uv_loop())
}
}
#[cfg(test)]
mod test {
use std::cast::transmute;
use std::ptr;
use std::unstable::run_in_bare_thread;
use super::{slice_to_uv_buf, Loop};
#[test]
fn test_slice_to_uv_buf() {
let slice = [0, .. 20];
let buf = slice_to_uv_buf(slice);
assert_eq!(buf.len, 20);
unsafe {
let base = transmute::<*u8, *mut u8>(buf.base);
(*base) = 1;
(*ptr::mut_offset(base, 1)) = 2;
}
uvll::buf_init(data, v.len())
}
}
/// Transmute a Buf that was once a ~[u8] back to ~[u8]
pub fn vec_from_uv_buf(buf: Buf) -> Option<~[u8]> {
#[fixed_stack_segment]; #[inline(never)];
if !(buf.len == 0 && buf.base.is_null()) {
let v = unsafe { vec::from_buf(buf.base, buf.len as uint) };
unsafe { free(buf.base as *c_void) };
return Some(v);
} else {
// No buffer
uvdebug!("No buffer!");
return None;
}
}
/*
#[test]
fn test_slice_to_uv_buf() {
let slice = [0, .. 20];
let buf = slice_to_uv_buf(slice);
assert!(buf.len == 20);
unsafe {
let base = transmute::<*u8, *mut u8>(buf.base);
(*base) = 1;
(*ptr::mut_offset(base, 1)) = 2;
assert!(slice[0] == 1);
assert!(slice[1] == 2);
}
assert!(slice[0] == 1);
assert!(slice[1] == 2);
}
#[test]
fn loop_smoke_test() {
do run_in_bare_thread {
let mut loop_ = Loop::new();
loop_.run();
loop_.close();
#[test]
fn loop_smoke_test() {
do run_in_bare_thread {
let mut loop_ = Loop::new();
loop_.run();
loop_.close();
}
}
}
*/

5
src/librustuv/macros.rs
View File

@ -27,6 +27,11 @@ macro_rules! uvdebug (
})
)
// get a handle for the current scheduler
macro_rules! get_handle_to_current_scheduler(
() => (do Local::borrow |sched: &mut Scheduler| { sched.make_handle() })
)
pub fn dumb_println(args: &fmt::Arguments) {
use std::rt::io::native::stdio::stderr;
use std::rt::io::Writer;

1806
src/librustuv/net.rs
View File

File diff suppressed because it is too large Load Diff

366
src/librustuv/pipe.rs
View File

@ -8,91 +8,327 @@
// option. This file may not be copied, modified, or distributed
// except according to those terms.
use std::libc;
use std::c_str::CString;
use std::libc;
use std::rt::BlockedTask;
use std::rt::io::IoError;
use std::rt::local::Local;
use std::rt::rtio::{RtioPipe, RtioUnixListener, RtioUnixAcceptor};
use std::rt::sched::{Scheduler, SchedHandle};
use std::rt::tube::Tube;
use std::task;
use super::{Loop, UvError, Watcher, NativeHandle, status_to_maybe_uv_error};
use super::ConnectionCallback;
use net;
use stream::StreamWatcher;
use super::{Loop, UvError, UvHandle, Request, uv_error_to_io_error,
wait_until_woken_after};
use uvio::HomingIO;
use uvll;
pub struct Pipe(*uvll::uv_pipe_t);
pub struct PipeWatcher {
stream: StreamWatcher,
home: SchedHandle,
priv defused: bool,
}
impl Watcher for Pipe {}
pub struct PipeListener {
home: SchedHandle,
pipe: *uvll::uv_pipe_t,
priv outgoing: Tube<Result<~RtioPipe, IoError>>,
}
impl Pipe {
pub fn new(loop_: &Loop, ipc: bool) -> Pipe {
unsafe {
pub struct PipeAcceptor {
listener: ~PipeListener,
priv incoming: Tube<Result<~RtioPipe, IoError>>,
}
// PipeWatcher implementation and traits
impl PipeWatcher {
// Creates an uninitialized pipe watcher. The underlying uv pipe is ready to
// get bound to some other source (this is normally a helper method paired
// with another call).
pub fn new(loop_: &Loop, ipc: bool) -> PipeWatcher {
let handle = unsafe {
let handle = uvll::malloc_handle(uvll::UV_NAMED_PIPE);
assert!(handle.is_not_null());
assert!(!handle.is_null());
let ipc = ipc as libc::c_int;
assert_eq!(uvll::pipe_init(loop_.native_handle(), handle, ipc), 0);
let mut ret: Pipe =
NativeHandle::from_native_handle(handle);
ret.install_watcher_data();
ret
assert_eq!(uvll::uv_pipe_init(loop_.handle, handle, ipc), 0);
handle
};
PipeWatcher {
stream: StreamWatcher::new(handle),
home: get_handle_to_current_scheduler!(),
defused: false,
}
}
pub fn as_stream(&self) -> net::StreamWatcher {
net::StreamWatcher(**self as *uvll::uv_stream_t)
}
#[fixed_stack_segment] #[inline(never)]
pub fn open(&mut self, file: libc::c_int) -> Result<(), UvError> {
match unsafe { uvll::pipe_open(self.native_handle(), file) } {
0 => Ok(()),
pub fn open(loop_: &Loop, file: libc::c_int) -> Result<PipeWatcher, UvError>
{
let pipe = PipeWatcher::new(loop_, false);
match unsafe { uvll::uv_pipe_open(pipe.handle(), file) } {
0 => Ok(pipe),
n => Err(UvError(n))
}
}
#[fixed_stack_segment] #[inline(never)]
pub fn bind(&mut self, name: &CString) -> Result<(), UvError> {
do name.with_ref |name| {
match unsafe { uvll::pipe_bind(self.native_handle(), name) } {
0 => Ok(()),
pub fn connect(loop_: &Loop, name: &CString) -> Result<PipeWatcher, UvError>
{
struct Ctx { task: Option<BlockedTask>, result: libc::c_int, }
return do task::unkillable {
let mut cx = Ctx { task: None, result: 0 };
let mut req = Request::new(uvll::UV_CONNECT);
let pipe = PipeWatcher::new(loop_, false);
do wait_until_woken_after(&mut cx.task) {
unsafe {
uvll::uv_pipe_connect(req.handle,
pipe.handle(),
name.with_ref(|p| p),
connect_cb)
}
req.set_data(&cx);
req.defuse(); // uv callback now owns this request
}
match cx.result {
0 => Ok(pipe),
n => Err(UvError(n))
}
};
extern fn connect_cb(req: *uvll::uv_connect_t, status: libc::c_int) {;
let req = Request::wrap(req);
assert!(status != uvll::ECANCELED);
let cx: &mut Ctx = unsafe { req.get_data() };
cx.result = status;
let sched: ~Scheduler = Local::take();
sched.resume_blocked_task_immediately(cx.task.take_unwrap());
}
}
pub fn handle(&self) -> *uvll::uv_pipe_t { self.stream.handle }
// Unwraps the underlying uv pipe. This cancels destruction of the pipe and
// allows the pipe to get moved elsewhere
fn unwrap(mut self) -> *uvll::uv_pipe_t {
self.defused = true;
return self.stream.handle;
}
}
impl RtioPipe for PipeWatcher {
fn read(&mut self, buf: &mut [u8]) -> Result<uint, IoError> {
let _m = self.fire_homing_missile();
self.stream.read(buf).map_err(uv_error_to_io_error)
}
fn write(&mut self, buf: &[u8]) -> Result<(), IoError> {
let _m = self.fire_homing_missile();
self.stream.write(buf).map_err(uv_error_to_io_error)
}
}
impl HomingIO for PipeWatcher {
fn home<'a>(&'a mut self) -> &'a mut SchedHandle { &mut self.home }
}
impl UvHandle<uvll::uv_pipe_t> for PipeWatcher {
fn uv_handle(&self) -> *uvll::uv_pipe_t { self.stream.handle }
}
impl Drop for PipeWatcher {
fn drop(&mut self) {
if !self.defused {
let _m = self.fire_homing_missile();
self.close();
}
}
}
extern fn pipe_close_cb(handle: *uvll::uv_handle_t) {
unsafe { uvll::free_handle(handle) }
}
// PipeListener implementation and traits
impl PipeListener {
pub fn bind(loop_: &Loop, name: &CString) -> Result<~PipeListener, UvError> {
do task::unkillable {
let pipe = PipeWatcher::new(loop_, false);
match unsafe {
uvll::uv_pipe_bind(pipe.handle(), name.with_ref(|p| p))
} {
0 => {
// If successful, unwrap the PipeWatcher because we control how
// we close the pipe differently. We can't rely on
// StreamWatcher's default close method.
let p = ~PipeListener {
home: get_handle_to_current_scheduler!(),
pipe: pipe.unwrap(),
outgoing: Tube::new(),
};
Ok(p.install())
}
n => Err(UvError(n))
}
}
}
#[fixed_stack_segment] #[inline(never)]
pub fn connect(&mut self, name: &CString, cb: ConnectionCallback) {
{
let data = self.get_watcher_data();
assert!(data.connect_cb.is_none());
data.connect_cb = Some(cb);
}
let connect = net::ConnectRequest::new();
let name = do name.with_ref |p| { p };
unsafe {
uvll::pipe_connect(connect.native_handle(),
self.native_handle(),
name,
connect_cb)
}
extern "C" fn connect_cb(req: *uvll::uv_connect_t, status: libc::c_int) {
let connect_request: net::ConnectRequest =
NativeHandle::from_native_handle(req);
let mut stream_watcher = connect_request.stream();
connect_request.delete();
let cb = stream_watcher.get_watcher_data().connect_cb.take_unwrap();
let status = status_to_maybe_uv_error(status);
cb(stream_watcher, status);
}
}
}
impl NativeHandle<*uvll::uv_pipe_t> for Pipe {
fn from_native_handle(handle: *uvll::uv_pipe_t) -> Pipe {
Pipe(handle)
}
fn native_handle(&self) -> *uvll::uv_pipe_t {
match self { &Pipe(ptr) => ptr }
impl RtioUnixListener for PipeListener {
fn listen(mut ~self) -> Result<~RtioUnixAcceptor, IoError> {
// create the acceptor object from ourselves
let incoming = self.outgoing.clone();
let mut acceptor = ~PipeAcceptor {
listener: self,
incoming: incoming,
};
let _m = acceptor.fire_homing_missile();
// XXX: the 128 backlog should be configurable
match unsafe { uvll::uv_listen(acceptor.listener.pipe, 128, listen_cb) } {
0 => Ok(acceptor as ~RtioUnixAcceptor),
n => Err(uv_error_to_io_error(UvError(n))),
}
}
}
impl HomingIO for PipeListener {
fn home<'r>(&'r mut self) -> &'r mut SchedHandle { &mut self.home }
}
impl UvHandle<uvll::uv_pipe_t> for PipeListener {
fn uv_handle(&self) -> *uvll::uv_pipe_t { self.pipe }
}
extern fn listen_cb(server: *uvll::uv_stream_t, status: libc::c_int) {
assert!(status != uvll::ECANCELED);
let msg = match status {
0 => {
let loop_ = Loop::wrap(unsafe {
uvll::get_loop_for_uv_handle(server)
});
let client = PipeWatcher::new(&loop_, false);
assert_eq!(unsafe { uvll::uv_accept(server, client.handle()) }, 0);
Ok(~client as ~RtioPipe)
}
n => Err(uv_error_to_io_error(UvError(n)))
};
let pipe: &mut PipeListener = unsafe { UvHandle::from_uv_handle(&server) };
pipe.outgoing.send(msg);
}
impl Drop for PipeListener {
fn drop(&mut self) {
let _m = self.fire_homing_missile();
self.close();
}
}
// PipeAcceptor implementation and traits
impl RtioUnixAcceptor for PipeAcceptor {
fn accept(&mut self) -> Result<~RtioPipe, IoError> {
let _m = self.fire_homing_missile();
self.incoming.recv()
}
}
impl HomingIO for PipeAcceptor {
fn home<'r>(&'r mut self) -> &'r mut SchedHandle { self.listener.home() }
}
#[cfg(test)]
mod tests {
use std::cell::Cell;
use std::comm::oneshot;
use std::rt::rtio::{RtioUnixListener, RtioUnixAcceptor, RtioPipe};
use std::rt::test::next_test_unix;
use std::task;
use super::*;
use super::super::local_loop;
#[test]
#[ignore(cfg(windows))] // FIXME(#10386): how windows pipes work
fn connect_err() {
match PipeWatcher::connect(local_loop(), &"path/to/nowhere".to_c_str()) {
Ok(*) => fail!(),
Err(*) => {}
}
}
#[test]
#[ignore(cfg(windows))] // FIXME(#10386): how windows pipes work
fn bind_err() {
match PipeListener::bind(local_loop(), &"path/to/nowhere".to_c_str()) {
Ok(*) => fail!(),
Err(e) => assert_eq!(e.name(), ~"EACCES"),
}
}
#[test]
#[ignore(cfg(windows))] // FIXME(#10386): how windows pipes work
fn bind() {
let p = next_test_unix().to_c_str();
match PipeListener::bind(local_loop(), &p) {
Ok(*) => {}
Err(*) => fail!(),
}
}
#[test] #[should_fail]
#[ignore(cfg(windows))] // FIXME(#10386): how windows pipes work
fn bind_fail() {
let p = next_test_unix().to_c_str();
let _w = PipeListener::bind(local_loop(), &p).unwrap();
fail!();
}
#[test]
#[ignore(cfg(windows))] // FIXME(#10386): how windows pipes work
fn connect() {
let path = next_test_unix();
let path2 = path.clone();
let (port, chan) = oneshot();
let chan = Cell::new(chan);
do spawn {
let p = PipeListener::bind(local_loop(), &path2.to_c_str()).unwrap();
let mut p = p.listen().unwrap();
chan.take().send(());
let mut client = p.accept().unwrap();
let mut buf = [0];
assert!(client.read(buf).unwrap() == 1);
assert_eq!(buf[0], 1);
assert!(client.write([2]).is_ok());
}
port.recv();
let mut c = PipeWatcher::connect(local_loop(), &path.to_c_str()).unwrap();
assert!(c.write([1]).is_ok());
let mut buf = [0];
assert!(c.read(buf).unwrap() == 1);
assert_eq!(buf[0], 2);
}
#[test] #[should_fail]
#[ignore(cfg(windows))] // FIXME(#10386): how windows pipes work
fn connect_fail() {
let path = next_test_unix();
let path2 = path.clone();
let (port, chan) = oneshot();
let chan = Cell::new(chan);
do task::spawn_unlinked { // plz no linked failure
let p = PipeListener::bind(local_loop(), &path2.to_c_str()).unwrap();
let mut p = p.listen().unwrap();
chan.take().send(());
p.accept();
}
port.recv();
let _c = PipeWatcher::connect(local_loop(), &path.to_c_str()).unwrap();
fail!()
}
}

177
src/librustuv/process.rs
View File

@ -8,59 +8,44 @@
// option. This file may not be copied, modified, or distributed
// except according to those terms.
use std::cell::Cell;
use std::libc::c_int;
use std::libc;
use std::ptr;
use std::vec;
use std::rt::BlockedTask;
use std::rt::io::IoError;
use std::rt::io::process::*;
use std::rt::local::Local;
use std::rt::rtio::RtioProcess;
use std::rt::sched::{Scheduler, SchedHandle};
use std::vec;
use super::{Watcher, Loop, NativeHandle, UvError};
use super::{status_to_maybe_uv_error, ExitCallback};
use uvio::{UvPipeStream, UvUnboundPipe};
use super::{Loop, UvHandle, UvError, uv_error_to_io_error,
wait_until_woken_after};
use uvio::HomingIO;
use uvll;
use pipe::PipeWatcher;
/// A process wraps the handle of the underlying uv_process_t.
pub struct Process(*uvll::uv_process_t);
pub struct Process {
handle: *uvll::uv_process_t,
home: SchedHandle,
impl Watcher for Process {}
/// Task to wake up (may be null) for when the process exits
to_wake: Option<BlockedTask>,
/// Collected from the exit_cb
exit_status: Option<int>,
term_signal: Option<int>,
}
impl Process {
/// Creates a new process, ready to spawn inside an event loop
pub fn new() -> Process {
let handle = unsafe { uvll::malloc_handle(uvll::UV_PROCESS) };
assert!(handle.is_not_null());
let mut ret: Process = NativeHandle::from_native_handle(handle);
ret.install_watcher_data();
return ret;
}
/// Spawn a new process inside the specified event loop.
///
/// The `config` variable will be passed down to libuv, and the `exit_cb`
/// will be run only once, when the process exits.
///
/// Returns either the corresponding process object or an error which
/// occurred.
pub fn spawn(&mut self, loop_: &Loop, config: ProcessConfig,
exit_cb: ExitCallback)
-> Result<~[Option<~UvPipeStream>], UvError>
pub fn spawn(loop_: &Loop, config: ProcessConfig)
-> Result<(~Process, ~[Option<PipeWatcher>]), UvError>
{
let cwd = config.cwd.map(|s| s.to_c_str());
extern fn on_exit(p: *uvll::uv_process_t,
exit_status: libc::c_int,
term_signal: libc::c_int) {
let mut p: Process = NativeHandle::from_native_handle(p);
let err = match exit_status {
0 => None,
_ => status_to_maybe_uv_error(-1)
};
p.get_watcher_data().exit_cb.take_unwrap()(p,
exit_status as int,
term_signal as int,
err);
}
let io = config.io;
let mut stdio = vec::with_capacity::<uvll::uv_stdio_container_t>(io.len());
let mut ret_io = vec::with_capacity(io.len());
@ -73,9 +58,7 @@ impl Process {
}
}
let exit_cb = Cell::new(exit_cb);
let ret_io = Cell::new(ret_io);
do with_argv(config.program, config.args) |argv| {
let ret = do with_argv(config.program, config.args) |argv| {
do with_env(config.env) |envp| {
let options = uvll::uv_process_options_t {
exit_cb: on_exit,
@ -93,40 +76,52 @@ impl Process {
gid: 0,
};
let handle = UvHandle::alloc(None::<Process>, uvll::UV_PROCESS);
let process = ~Process {
handle: handle,
home: get_handle_to_current_scheduler!(),
to_wake: None,
exit_status: None,
term_signal: None,
};
match unsafe {
uvll::spawn(loop_.native_handle(), **self, options)
uvll::uv_spawn(loop_.handle, handle, &options)
} {
0 => {
(*self).get_watcher_data().exit_cb = Some(exit_cb.take());
Ok(ret_io.take())
}
err => Err(UvError(err))
0 => Ok(process.install()),
err => Err(UvError(err)),
}
}
};
match ret {
Ok(p) => Ok((p, ret_io)),
Err(e) => Err(e),
}
}
}
/// Sends a signal to this process.
///
/// This is a wrapper around `uv_process_kill`
pub fn kill(&self, signum: int) -> Result<(), UvError> {
match unsafe {
uvll::process_kill(self.native_handle(), signum as libc::c_int)
} {
0 => Ok(()),
err => Err(UvError(err))
extern fn on_exit(handle: *uvll::uv_process_t,
exit_status: i64,
term_signal: libc::c_int) {
let p: &mut Process = unsafe { UvHandle::from_uv_handle(&handle) };
assert!(p.exit_status.is_none());
assert!(p.term_signal.is_none());
p.exit_status = Some(exit_status as int);
p.term_signal = Some(term_signal as int);
match p.to_wake.take() {
Some(task) => {
let scheduler: ~Scheduler = Local::take();
scheduler.resume_blocked_task_immediately(task);
}
}
/// Returns the process id of a spawned process
pub fn pid(&self) -> libc::pid_t {
unsafe { uvll::process_pid(**self) as libc::pid_t }
None => {}
}
}
unsafe fn set_stdio(dst: *uvll::uv_stdio_container_t,
io: &StdioContainer,
loop_: &Loop) -> Option<~UvPipeStream> {
loop_: &Loop) -> Option<PipeWatcher> {
match *io {
Ignored => {
uvll::set_stdio_container_flags(dst, uvll::STDIO_IGNORE);
@ -145,11 +140,10 @@ unsafe fn set_stdio(dst: *uvll::uv_stdio_container_t,
if writable {
flags |= uvll::STDIO_WRITABLE_PIPE as libc::c_int;
}
let pipe = UvUnboundPipe::new(loop_);
let handle = pipe.pipe.as_stream().native_handle();
let pipe = PipeWatcher::new(loop_, false);
uvll::set_stdio_container_flags(dst, flags);
uvll::set_stdio_container_stream(dst, handle);
Some(~UvPipeStream::new(pipe))
uvll::set_stdio_container_stream(dst, pipe.handle());
Some(pipe)
}
}
}
@ -192,11 +186,52 @@ fn with_env<T>(env: Option<&[(~str, ~str)]>, f: &fn(**libc::c_char) -> T) -> T {
c_envp.as_imm_buf(|buf, _| f(buf))
}
impl NativeHandle<*uvll::uv_process_t> for Process {
fn from_native_handle(handle: *uvll::uv_process_t) -> Process {
Process(handle)
impl HomingIO for Process {
fn home<'r>(&'r mut self) -> &'r mut SchedHandle { &mut self.home }
}
impl UvHandle<uvll::uv_process_t> for Process {
fn uv_handle(&self) -> *uvll::uv_process_t { self.handle }
}
impl RtioProcess for Process {
fn id(&self) -> libc::pid_t {
unsafe { uvll::process_pid(self.handle) as libc::pid_t }
}
fn native_handle(&self) -> *uvll::uv_process_t {
match self { &Process(ptr) => ptr }
fn kill(&mut self, signal: int) -> Result<(), IoError> {
let _m = self.fire_homing_missile();
match unsafe {
uvll::uv_process_kill(self.handle, signal as libc::c_int)
} {
0 => Ok(()),
err => Err(uv_error_to_io_error(UvError(err)))
}
}
fn wait(&mut self) -> int {
// Make sure (on the home scheduler) that we have an exit status listed
let _m = self.fire_homing_missile();
match self.exit_status {
Some(*) => {}
None => {
// If there's no exit code previously listed, then the
// process's exit callback has yet to be invoked. We just
// need to deschedule ourselves and wait to be reawoken.
wait_until_woken_after(&mut self.to_wake, || {});
assert!(self.exit_status.is_some());
}
}
// FIXME(#10109): this is wrong
self.exit_status.unwrap()
}
}
impl Drop for Process {
fn drop(&mut self) {
let _m = self.fire_homing_missile();
assert!(self.to_wake.is_none());
self.close();
}
}

116
src/librustuv/signal.rs
View File

@ -8,65 +8,93 @@
// option. This file may not be copied, modified, or distributed
// except according to those terms.
use std::cast;
use std::libc::c_int;
use std::rt::io::signal::Signum;
use std::rt::sched::{SchedHandle, Scheduler};
use std::comm::{SharedChan, SendDeferred};
use std::rt::local::Local;
use std::rt::rtio::RtioSignal;
use super::{Loop, NativeHandle, SignalCallback, UvError, Watcher};
use super::{Loop, UvError, UvHandle};
use uvll;
use uvio::HomingIO;
pub struct SignalWatcher(*uvll::uv_signal_t);
pub struct SignalWatcher {
handle: *uvll::uv_signal_t,
home: SchedHandle,
impl Watcher for SignalWatcher { }
channel: SharedChan<Signum>,
signal: Signum,
}
impl SignalWatcher {
pub fn new(loop_: &mut Loop) -> SignalWatcher {
unsafe {
let handle = uvll::malloc_handle(uvll::UV_SIGNAL);
assert!(handle.is_not_null());
assert!(0 == uvll::signal_init(loop_.native_handle(), handle));
let mut watcher: SignalWatcher = NativeHandle::from_native_handle(handle);
watcher.install_watcher_data();
return watcher;
}
}
pub fn start(&mut self, signum: Signum, callback: SignalCallback)
-> Result<(), UvError>
{
return unsafe {
match uvll::signal_start(self.native_handle(), signal_cb,
signum as c_int) {
0 => {
let data = self.get_watcher_data();
data.signal_cb = Some(callback);
Ok(())
}
n => Err(UvError(n)),
}
pub fn new(loop_: &mut Loop, signum: Signum,
channel: SharedChan<Signum>) -> Result<~SignalWatcher, UvError> {
let s = ~SignalWatcher {
handle: UvHandle::alloc(None::<SignalWatcher>, uvll::UV_SIGNAL),
home: get_handle_to_current_scheduler!(),
channel: channel,
signal: signum,
};
assert_eq!(unsafe {
uvll::uv_signal_init(loop_.handle, s.handle)
}, 0);
extern fn signal_cb(handle: *uvll::uv_signal_t, signum: c_int) {
let mut watcher: SignalWatcher = NativeHandle::from_native_handle(handle);
let data = watcher.get_watcher_data();
let cb = data.signal_cb.get_ref();
(*cb)(watcher, unsafe { cast::transmute(signum as int) });
match unsafe {
uvll::uv_signal_start(s.handle, signal_cb, signum as c_int)
} {
0 => Ok(s.install()),
n => Err(UvError(n)),
}
}
pub fn stop(&mut self) {
unsafe {
uvll::signal_stop(self.native_handle());
}
}
}
impl NativeHandle<*uvll::uv_signal_t> for SignalWatcher {
fn from_native_handle(handle: *uvll::uv_signal_t) -> SignalWatcher {
SignalWatcher(handle)
}
extern fn signal_cb(handle: *uvll::uv_signal_t, signum: c_int) {
let s: &mut SignalWatcher = unsafe { UvHandle::from_uv_handle(&handle) };
assert_eq!(signum as int, s.signal as int);
s.channel.send_deferred(s.signal);
}
fn native_handle(&self) -> *uvll::uv_signal_t {
match self { &SignalWatcher(ptr) => ptr }
impl HomingIO for SignalWatcher {
fn home<'r>(&'r mut self) -> &'r mut SchedHandle { &mut self.home }
}
impl UvHandle<uvll::uv_signal_t> for SignalWatcher {
fn uv_handle(&self) -> *uvll::uv_signal_t { self.handle }
}
impl RtioSignal for SignalWatcher {}
impl Drop for SignalWatcher {
fn drop(&mut self) {
let _m = self.fire_homing_missile();
self.close_async_();
}
}
#[cfg(test)]
mod test {
use super::*;
use std::cell::Cell;
use super::super::local_loop;
use std::rt::io::signal;
use std::comm::{SharedChan, stream};
#[test]
fn closing_channel_during_drop_doesnt_kill_everything() {
// see issue #10375, relates to timers as well.
let (port, chan) = stream();
let chan = SharedChan::new(chan);
let _signal = SignalWatcher::new(local_loop(), signal::Interrupt,
chan);
let port = Cell::new(port);
do spawn {
port.take().try_recv();
}
// when we drop the SignalWatcher we're going to destroy the channel,
// which must wake up the task on the other end
}
}

180
src/librustuv/stream.rs Normal file
View File

@ -0,0 +1,180 @@
// 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::cast;
use std::libc::{c_int, size_t, ssize_t};
use std::ptr;
use std::rt::BlockedTask;
use std::rt::local::Local;
use std::rt::sched::Scheduler;
use super::{UvError, Buf, slice_to_uv_buf, Request, wait_until_woken_after,
ForbidUnwind};
use uvll;
// This is a helper structure which is intended to get embedded into other
// Watcher structures. This structure will retain a handle to the underlying
// uv_stream_t instance, and all I/O operations assume that it's already located
// on the appropriate scheduler.
pub struct StreamWatcher {
handle: *uvll::uv_stream_t,
// Cache the last used uv_write_t so we don't have to allocate a new one on
// every call to uv_write(). Ideally this would be a stack-allocated
// structure, but currently we don't have mappings for all the structures
// defined in libuv, so we're foced to malloc this.
priv last_write_req: Option<Request>,
}
struct ReadContext {
buf: Option<Buf>,
result: ssize_t,
task: Option<BlockedTask>,
}
struct WriteContext {
result: c_int,
task: Option<BlockedTask>,
}
impl StreamWatcher {
// Creates a new helper structure which should be then embedded into another
// watcher. This provides the generic read/write methods on streams.
//
// This structure will *not* close the stream when it is dropped. It is up
// to the enclosure structure to be sure to call the close method (which
// will block the task). Note that this is also required to prevent memory
// leaks.
//
// It should also be noted that the `data` field of the underlying uv handle
// will be manipulated on each of the methods called on this watcher.
// Wrappers should ensure to always reset the field to an appropriate value
// if they rely on the field to perform an action.
pub fn new(stream: *uvll::uv_stream_t) -> StreamWatcher {
StreamWatcher {
handle: stream,
last_write_req: None,
}
}
pub fn read(&mut self, buf: &mut [u8]) -> Result<uint, UvError> {
// This read operation needs to get canceled on an unwind via libuv's
// uv_read_stop function
let _f = ForbidUnwind::new("stream read");
// Send off the read request, but don't block until we're sure that the
// read request is queued.
match unsafe {
uvll::uv_read_start(self.handle, alloc_cb, read_cb)
} {
0 => {
let mut rcx = ReadContext {
buf: Some(slice_to_uv_buf(buf)),
result: 0,
task: None,
};
do wait_until_woken_after(&mut rcx.task) {
unsafe {
uvll::set_data_for_uv_handle(self.handle, &rcx)
}
}
match rcx.result {
n if n < 0 => Err(UvError(n as c_int)),
n => Ok(n as uint),
}
}
n => Err(UvError(n))
}
}
pub fn write(&mut self, buf: &[u8]) -> Result<(), UvError> {
// The ownership of the write request is dubious if this function
// unwinds. I believe that if the write_cb fails to re-schedule the task
// then the write request will be leaked.
let _f = ForbidUnwind::new("stream write");
// Prepare the write request, either using a cached one or allocating a
// new one
let mut req = match self.last_write_req.take() {
Some(req) => req, None => Request::new(uvll::UV_WRITE),
};
req.set_data(ptr::null::<()>());
// Send off the request, but be careful to not block until we're sure
// that the write reqeust is queued. If the reqeust couldn't be queued,
// then we should return immediately with an error.
match unsafe {
uvll::uv_write(req.handle, self.handle, [slice_to_uv_buf(buf)],
write_cb)
} {
0 => {
let mut wcx = WriteContext { result: 0, task: None, };
req.defuse(); // uv callback now owns this request
do wait_until_woken_after(&mut wcx.task) {
req.set_data(&wcx);
}
self.last_write_req = Some(Request::wrap(req.handle));
match wcx.result {
0 => Ok(()),
n => Err(UvError(n)),
}
}
n => Err(UvError(n)),
}
}
}
// This allocation callback expects to be invoked once and only once. It will
// unwrap the buffer in the ReadContext stored in the stream and return it. This
// will fail if it is called more than once.
extern fn alloc_cb(stream: *uvll::uv_stream_t, _hint: size_t, buf: *mut Buf) {
uvdebug!("alloc_cb");
unsafe {
let rcx: &mut ReadContext =
cast::transmute(uvll::get_data_for_uv_handle(stream));
*buf = rcx.buf.take().expect("stream alloc_cb called more than once");
}
}
// When a stream has read some data, we will always forcibly stop reading and
// return all the data read (even if it didn't fill the whole buffer).
extern fn read_cb(handle: *uvll::uv_stream_t, nread: ssize_t, _buf: *Buf) {
uvdebug!("read_cb {}", nread);
assert!(nread != uvll::ECANCELED as ssize_t);
let rcx: &mut ReadContext = unsafe {
cast::transmute(uvll::get_data_for_uv_handle(handle))
};
// Stop reading so that no read callbacks are
// triggered before the user calls `read` again.
// XXX: Is there a performance impact to calling
// stop here?
unsafe { assert_eq!(uvll::uv_read_stop(handle), 0); }
rcx.result = nread;
let scheduler: ~Scheduler = Local::take();
scheduler.resume_blocked_task_immediately(rcx.task.take_unwrap());
}
// Unlike reading, the WriteContext is stored in the uv_write_t request. Like
// reading, however, all this does is wake up the blocked task after squirreling
// away the error code as a result.
extern fn write_cb(req: *uvll::uv_write_t, status: c_int) {
let mut req = Request::wrap(req);
assert!(status != uvll::ECANCELED);
// Remember to not free the request because it is re-used between writes on
// the same stream.
let wcx: &mut WriteContext = unsafe { req.get_data() };
wcx.result = status;
req.defuse();
let sched: ~Scheduler = Local::take();
sched.resume_blocked_task_immediately(wcx.task.take_unwrap());
}

379
src/librustuv/timer.rs
View File

@ -8,150 +8,297 @@
// option. This file may not be copied, modified, or distributed
// except according to those terms.
use std::comm::{oneshot, stream, PortOne, ChanOne, SendDeferred};
use std::libc::c_int;
use std::rt::BlockedTask;
use std::rt::local::Local;
use std::rt::rtio::RtioTimer;
use std::rt::sched::{Scheduler, SchedHandle};
use std::util;
use uvll;
use super::{Watcher, Loop, NativeHandle, TimerCallback, status_to_maybe_uv_error};
use super::{Loop, UvHandle, ForbidUnwind, ForbidSwitch};
use uvio::HomingIO;
pub struct TimerWatcher(*uvll::uv_timer_t);
impl Watcher for TimerWatcher { }
pub struct TimerWatcher {
handle: *uvll::uv_timer_t,
home: SchedHandle,
action: Option<NextAction>,
}
pub enum NextAction {
WakeTask(BlockedTask),
SendOnce(ChanOne<()>),
SendMany(Chan<()>),
}
impl TimerWatcher {
pub fn new(loop_: &mut Loop) -> TimerWatcher {
unsafe {
let handle = uvll::malloc_handle(uvll::UV_TIMER);
assert!(handle.is_not_null());
assert!(0 == uvll::timer_init(loop_.native_handle(), handle));
let mut watcher: TimerWatcher = NativeHandle::from_native_handle(handle);
watcher.install_watcher_data();
return watcher;
}
pub fn new(loop_: &mut Loop) -> ~TimerWatcher {
let handle = UvHandle::alloc(None::<TimerWatcher>, uvll::UV_TIMER);
assert_eq!(unsafe {
uvll::uv_timer_init(loop_.handle, handle)
}, 0);
let me = ~TimerWatcher {
handle: handle,
action: None,
home: get_handle_to_current_scheduler!(),
};
return me.install();
}
pub fn start(&mut self, timeout: u64, repeat: u64, cb: TimerCallback) {
{
let data = self.get_watcher_data();
data.timer_cb = Some(cb);
}
unsafe {
uvll::timer_start(self.native_handle(), timer_cb, timeout, repeat);
}
extern fn timer_cb(handle: *uvll::uv_timer_t, status: c_int) {
let mut watcher: TimerWatcher = NativeHandle::from_native_handle(handle);
let data = watcher.get_watcher_data();
let cb = data.timer_cb.get_ref();
let status = status_to_maybe_uv_error(status);
(*cb)(watcher, status);
}
fn start(&mut self, msecs: u64, period: u64) {
assert_eq!(unsafe {
uvll::uv_timer_start(self.handle, timer_cb, msecs, period)
}, 0)
}
pub fn stop(&mut self) {
unsafe {
uvll::timer_stop(self.native_handle());
fn stop(&mut self) {
assert_eq!(unsafe { uvll::uv_timer_stop(self.handle) }, 0)
}
}
impl HomingIO for TimerWatcher {
fn home<'r>(&'r mut self) -> &'r mut SchedHandle { &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.stop();
let _missile = match util::replace(&mut self.action, None) {
None => missile, // no need to do a homing dance
Some(action) => {
util::ignore(missile); // un-home ourself
util::ignore(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");
let sched: ~Scheduler = Local::take();
do sched.deschedule_running_task_and_then |_sched, task| {
self.action = Some(WakeTask(task));
self.start(msecs, 0);
}
self.stop();
}
fn oneshot(&mut self, msecs: u64) -> PortOne<()> {
let (port, chan) = oneshot();
// 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.stop();
self.start(msecs, 0);
util::replace(&mut self.action, Some(SendOnce(chan)))
};
return port;
}
fn period(&mut self, msecs: u64) -> Port<()> {
let (port, chan) = stream();
// 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.stop();
self.start(msecs, msecs);
util::replace(&mut self.action, Some(SendMany(chan)))
};
return port;
}
}
extern fn timer_cb(handle: *uvll::uv_timer_t, status: c_int) {
let _f = ForbidSwitch::new("timer callback can't switch");
assert_eq!(status, 0);
let timer: &mut TimerWatcher = unsafe { UvHandle::from_uv_handle(&handle) };
match timer.action.take_unwrap() {
WakeTask(task) => {
let sched: ~Scheduler = Local::take();
sched.resume_blocked_task_immediately(task);
}
SendOnce(chan) => chan.send_deferred(()),
SendMany(chan) => {
chan.send_deferred(());
timer.action = Some(SendMany(chan));
}
}
}
impl NativeHandle<*uvll::uv_timer_t> for TimerWatcher {
fn from_native_handle(handle: *uvll::uv_timer_t) -> TimerWatcher {
TimerWatcher(handle)
}
fn native_handle(&self) -> *uvll::uv_idle_t {
match self { &TimerWatcher(ptr) => ptr }
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_async_();
self.action.take()
};
}
}
#[cfg(test)]
mod test {
use super::*;
use Loop;
use std::unstable::run_in_bare_thread;
use std::cell::Cell;
use std::rt::rtio::RtioTimer;
use super::super::local_loop;
#[test]
fn smoke_test() {
do run_in_bare_thread {
let mut count = 0;
let count_ptr: *mut int = &mut count;
let mut loop_ = Loop::new();
let mut timer = TimerWatcher::new(&mut loop_);
do timer.start(10, 0) |timer, status| {
assert!(status.is_none());
unsafe { *count_ptr += 1 };
timer.close(||());
}
loop_.run();
loop_.close();
assert!(count == 1);
}
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 start_twice() {
do run_in_bare_thread {
let mut count = 0;
let count_ptr: *mut int = &mut count;
let mut loop_ = Loop::new();
let mut timer = TimerWatcher::new(&mut loop_);
do timer.start(10, 0) |timer, status| {
let mut timer = timer;
assert!(status.is_none());
unsafe { *count_ptr += 1 };
do timer.start(10, 0) |timer, status| {
assert!(status.is_none());
unsafe { *count_ptr += 1 };
timer.close(||());
}
}
loop_.run();
loop_.close();
assert!(count == 2);
}
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.try_recv(), None);
assert_eq!(pport.try_recv(), None);
timer.oneshot(1).recv();
}
#[test]
fn repeat_stop() {
do run_in_bare_thread {
let mut count = 0;
let count_ptr: *mut int = &mut count;
let mut loop_ = Loop::new();
let mut timer = TimerWatcher::new(&mut loop_);
do timer.start(1, 2) |timer, status| {
assert!(status.is_none());
unsafe {
*count_ptr += 1;
if *count_ptr == 10 {
// Stop the timer and do something else
let mut timer = timer;
timer.stop();
// Freeze timer so it can be captured
let timer = timer;
let mut loop_ = timer.event_loop();
let mut timer2 = TimerWatcher::new(&mut loop_);
do timer2.start(10, 0) |timer2, _| {
*count_ptr += 1;
timer2.close(||());
// Restart the original timer
let mut timer = timer;
do timer.start(1, 0) |timer, _| {
*count_ptr += 1;
timer.close(||());
}
}
}
};
}
loop_.run();
loop_.close();
assert!(count == 12);
}
fn period() {
let mut timer = TimerWatcher::new(local_loop());
let port = timer.period(1);
port.recv();
port.recv();
let port = timer.period(1);
port.recv();
port.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 = Cell::new(timer.period(1000));
do spawn {
timer_port.take().try_recv();
}
// 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 = Cell::new(timer.period(1000));
do spawn {
timer_port.take().try_recv();
}
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 = Cell::new(timer.period(1000));
do spawn {
timer_port.take().try_recv();
}
timer.sleep(1);
}
#[test]
fn sender_goes_away_oneshot() {
let port = {
let mut timer = TimerWatcher::new(local_loop());
timer.oneshot(1000)
};
assert_eq!(port.try_recv(), None);
}
#[test]
fn sender_goes_away_period() {
let port = {
let mut timer = TimerWatcher::new(local_loop());
timer.period(1000)
};
assert_eq!(port.try_recv(), None);
}
#[test]
fn receiver_goes_away_oneshot() {
let mut timer1 = TimerWatcher::new(local_loop());
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());
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);
}
}

113
src/librustuv/tty.rs
View File

@ -9,75 +9,110 @@
// except according to those terms.
use std::libc;
use std::rt::io::IoError;
use std::rt::local::Local;
use std::rt::rtio::RtioTTY;
use std::rt::sched::{Scheduler, SchedHandle};
use super::{Watcher, Loop, NativeHandle, UvError};
use net;
use stream::StreamWatcher;
use super::{Loop, UvError, UvHandle, uv_error_to_io_error};
use uvio::HomingIO;
use uvll;
/// A process wraps the handle of the underlying uv_process_t.
pub struct TTY(*uvll::uv_tty_t);
pub struct TtyWatcher{
tty: *uvll::uv_tty_t,
stream: StreamWatcher,
home: SchedHandle,
fd: libc::c_int,
}
impl Watcher for TTY {}
impl TTY {
#[fixed_stack_segment] #[inline(never)]
pub fn new(loop_: &Loop, fd: libc::c_int, readable: bool) ->
Result<TTY, UvError>
impl TtyWatcher {
pub fn new(loop_: &Loop, fd: libc::c_int, readable: bool)
-> Result<TtyWatcher, UvError>
{
let handle = unsafe { uvll::malloc_handle(uvll::UV_TTY) };
assert!(handle.is_not_null());
// libuv may succeed in giving us a handle (via uv_tty_init), but if the
// handle isn't actually connected to a terminal there are frequently
// many problems in using it with libuv. To get around this, always
// return a failure if the specified file descriptor isn't actually a
// TTY.
//
// Related:
// - https://github.com/joyent/libuv/issues/982
// - https://github.com/joyent/libuv/issues/988
if unsafe { uvll::guess_handle(fd) != uvll::UV_TTY as libc::c_int } {
return Err(UvError(uvll::EBADF));
}
let ret = unsafe {
uvll::tty_init(loop_.native_handle(), handle, fd as libc::c_int,
readable as libc::c_int)
};
match ret {
// If this file descriptor is indeed guessed to be a tty, then go ahead
// with attempting to open it as a tty.
let handle = UvHandle::alloc(None::<TtyWatcher>, uvll::UV_TTY);
match unsafe {
uvll::uv_tty_init(loop_.handle, handle, fd as libc::c_int,
readable as libc::c_int)
} {
0 => {
let mut ret: TTY = NativeHandle::from_native_handle(handle);
ret.install_watcher_data();
Ok(ret)
Ok(TtyWatcher {
tty: handle,
stream: StreamWatcher::new(handle),
home: get_handle_to_current_scheduler!(),
fd: fd,
})
}
n => {
unsafe { uvll::free_handle(handle); }
unsafe { uvll::free_handle(handle) }
Err(UvError(n))
}
}
}
}
pub fn as_stream(&self) -> net::StreamWatcher {
net::StreamWatcher(**self as *uvll::uv_stream_t)
impl RtioTTY for TtyWatcher {
fn read(&mut self, buf: &mut [u8]) -> Result<uint, IoError> {
let _m = self.fire_homing_missile();
self.stream.read(buf).map_err(uv_error_to_io_error)
}
#[fixed_stack_segment] #[inline(never)]
pub fn set_mode(&self, raw: bool) -> Result<(), UvError> {
fn write(&mut self, buf: &[u8]) -> Result<(), IoError> {
let _m = self.fire_homing_missile();
self.stream.write(buf).map_err(uv_error_to_io_error)
}
fn set_raw(&mut self, raw: bool) -> Result<(), IoError> {
let raw = raw as libc::c_int;
match unsafe { uvll::tty_set_mode(self.native_handle(), raw) } {
let _m = self.fire_homing_missile();
match unsafe { uvll::uv_tty_set_mode(self.tty, raw) } {
0 => Ok(()),
n => Err(UvError(n))
n => Err(uv_error_to_io_error(UvError(n)))
}
}
#[fixed_stack_segment] #[inline(never)] #[allow(unused_mut)]
pub fn get_winsize(&self) -> Result<(int, int), UvError> {
#[allow(unused_mut)]
fn get_winsize(&mut self) -> Result<(int, int), IoError> {
let mut width: libc::c_int = 0;
let mut height: libc::c_int = 0;
let widthptr: *libc::c_int = &width;
let heightptr: *libc::c_int = &width;
match unsafe { uvll::tty_get_winsize(self.native_handle(),
widthptr, heightptr) } {
let _m = self.fire_homing_missile();
match unsafe { uvll::uv_tty_get_winsize(self.tty,
widthptr, heightptr) } {
0 => Ok((width as int, height as int)),
n => Err(UvError(n))
n => Err(uv_error_to_io_error(UvError(n)))
}
}
}
impl NativeHandle<*uvll::uv_tty_t> for TTY {
fn from_native_handle(handle: *uvll::uv_tty_t) -> TTY {
TTY(handle)
}
fn native_handle(&self) -> *uvll::uv_tty_t {
match self { &TTY(ptr) => ptr }
}
impl UvHandle<uvll::uv_tty_t> for TtyWatcher {
fn uv_handle(&self) -> *uvll::uv_tty_t { self.tty }
}
impl HomingIO for TtyWatcher {
fn home<'a>(&'a mut self) -> &'a mut SchedHandle { &mut self.home }
}
impl Drop for TtyWatcher {
fn drop(&mut self) {
let _m = self.fire_homing_missile();
self.close_async_();
}
}

2492
src/librustuv/uvio.rs
View File

File diff suppressed because it is too large Load Diff

913
src/librustuv/uvll.rs
View File

File diff suppressed because it is too large Load Diff

16
src/libstd/logging.rs
View File

@ -107,14 +107,16 @@ pub fn log(_level: u32, args: &fmt::Arguments) {
let optional_task: Option<*mut Task> = Local::try_unsafe_borrow();
match optional_task {
Some(local) => {
// Use the available logger
(*local).logger.log(args);
}
None => {
// There is no logger anywhere, just write to stderr
let mut logger = StdErrLogger::new();
logger.log(args);
match (*local).logger {
// Use the available logger if we have one
Some(ref mut logger) => return logger.log(args),
None => {}
}
}
None => {}
}
// There is no logger anywhere, just write to stderr
let mut logger = StdErrLogger::new();
logger.log(args);
}
}

40
src/libstd/rt/basic.rs
View File

@ -15,7 +15,8 @@
use prelude::*;
use cast;
use rt::rtio::{EventLoop, IoFactory, RemoteCallback, PausibleIdleCallback};
use rt::rtio::{EventLoop, IoFactory, RemoteCallback, PausibleIdleCallback,
Callback};
use unstable::sync::Exclusive;
use util;
@ -25,9 +26,9 @@ pub fn event_loop() -> ~EventLoop {
}
struct BasicLoop {
work: ~[~fn()], // pending work
idle: Option<*BasicPausible>, // only one is allowed
remotes: ~[(uint, ~fn())],
work: ~[proc()], // pending work
idle: Option<*mut BasicPausible>, // only one is allowed
remotes: ~[(uint, ~Callback)],
next_remote: uint,
messages: Exclusive<~[Message]>
}
@ -86,8 +87,8 @@ impl BasicLoop {
fn message(&mut self, message: Message) {
match message {
RunRemote(i) => {
match self.remotes.iter().find(|& &(id, _)| id == i) {
Some(&(_, ref f)) => (*f)(),
match self.remotes.mut_iter().find(|& &(id, _)| id == i) {
Some(&(_, ref mut f)) => f.call(),
None => unreachable!()
}
}
@ -106,7 +107,7 @@ impl BasicLoop {
match self.idle {
Some(idle) => {
if (*idle).active {
(*(*idle).work.get_ref())();
(*idle).work.call();
}
}
None => {}
@ -144,22 +145,22 @@ impl EventLoop for BasicLoop {
}
}
fn callback(&mut self, f: ~fn()) {
fn callback(&mut self, f: proc()) {
self.work.push(f);
}
// XXX: Seems like a really weird requirement to have an event loop provide.
fn pausible_idle_callback(&mut self) -> ~PausibleIdleCallback {
let callback = ~BasicPausible::new(self);
fn pausible_idle_callback(&mut self, cb: ~Callback) -> ~PausibleIdleCallback {
let callback = ~BasicPausible::new(self, cb);
rtassert!(self.idle.is_none());
unsafe {
let cb_ptr: &*BasicPausible = cast::transmute(&callback);
let cb_ptr: &*mut BasicPausible = cast::transmute(&callback);
self.idle = Some(*cb_ptr);
}
return callback as ~PausibleIdleCallback;
}
fn remote_callback(&mut self, f: ~fn()) -> ~RemoteCallback {
fn remote_callback(&mut self, f: ~Callback) -> ~RemoteCallback {
let id = self.next_remote;
self.next_remote += 1;
self.remotes.push((id, f));
@ -203,36 +204,27 @@ impl Drop for BasicRemote {
struct BasicPausible {
eloop: *mut BasicLoop,
work: Option<~fn()>,
work: ~Callback,
active: bool,
}
impl BasicPausible {
fn new(eloop: &mut BasicLoop) -> BasicPausible {
fn new(eloop: &mut BasicLoop, cb: ~Callback) -> BasicPausible {
BasicPausible {
active: false,
work: None,
work: cb,
eloop: eloop,
}
}
}
impl PausibleIdleCallback for BasicPausible {
fn start(&mut self, f: ~fn()) {
rtassert!(!self.active && self.work.is_none());
self.active = true;
self.work = Some(f);
}
fn pause(&mut self) {
self.active = false;
}
fn resume(&mut self) {
self.active = true;
}
fn close(&mut self) {
self.active = false;
self.work = None;
}
}
impl Drop for BasicPausible {

45
src/libstd/rt/io/fs.rs
View File

@ -587,6 +587,22 @@ pub fn rmdir_recursive(path: &Path) {
rmdir(path);
}
/// Changes the timestamps for a file's last modification and access time.
/// The file at the path specified will have its last access time set to
/// `atime` and its modification time set to `mtime`. The times specified should
/// be in milliseconds.
///
/// # Errors
///
/// This function will raise on the `io_error` condition if an error
/// happens.
// FIXME(#10301) these arguments should not be u64
pub fn change_file_times(path: &Path, atime: u64, mtime: u64) {
do io_raise |io| {
io.fs_utime(&path.to_c_str(), atime, mtime)
};
}
impl Reader for File {
fn read(&mut self, buf: &mut [u8]) -> Option<uint> {
match self.fd.read(buf) {
@ -704,8 +720,8 @@ mod test {
use rt::io;
use str;
use super::{File, rmdir, mkdir, readdir, rmdir_recursive, mkdir_recursive,
copy, unlink, stat, symlink, link, readlink, chmod, chown,
lstat};
copy, unlink, stat, symlink, link, readlink, chmod,
lstat, change_file_times};
fn tmpdir() -> Path {
use os;
@ -1244,4 +1260,29 @@ mod test {
rmdir_recursive(&tmpdir);
}
#[test]
fn utime() {
let tmpdir = tmpdir();
let path = tmpdir.join("a");
File::create(&path);
change_file_times(&path, 1000, 2000);
assert_eq!(path.stat().accessed, 1000);
assert_eq!(path.stat().modified, 2000);
rmdir_recursive(&tmpdir);
}
#[test]
fn utime_noexist() {
let tmpdir = tmpdir();
match io::result(|| change_file_times(&tmpdir.join("a"), 100, 200)) {
Ok(*) => fail!(),
Err(*) => {}
}
rmdir_recursive(&tmpdir);
}
}

11
src/libstd/rt/io/mod.rs
View File

@ -423,7 +423,11 @@ pub fn ignore_io_error<T>(cb: &fn() -> T) -> T {
/// closure if no error occurred.
pub fn result<T>(cb: &fn() -> T) -> Result<T, IoError> {
let mut err = None;
let ret = io_error::cond.trap(|e| err = Some(e)).inside(cb);
let ret = io_error::cond.trap(|e| {
if err.is_none() {
err = Some(e);
}
}).inside(cb);
match err {
Some(e) => Err(e),
None => Ok(ret),
@ -1142,8 +1146,9 @@ pub struct FileStat {
/// The file permissions currently on the file
perm: FilePermission,
// XXX: These time fields are pretty useless without an actual time
// representation, what are the milliseconds relative to?
// FIXME(#10301): These time fields are pretty useless without an actual
// time representation, what are the milliseconds relative
// to?
/// The time that the file was created at, in platform-dependent
/// milliseconds

16
src/libstd/rt/io/native/file.rs
View File

@ -80,18 +80,20 @@ pub type fd_t = libc::c_int;
pub struct FileDesc {
priv fd: fd_t,
priv close_on_drop: bool,
}
impl FileDesc {
/// Create a `FileDesc` from an open C file descriptor.
///
/// The `FileDesc` will take ownership of the specified file descriptor and
/// close it upon destruction.
/// close it upon destruction if the `close_on_drop` flag is true, otherwise
/// it will not close the file descriptor when this `FileDesc` is dropped.
///
/// Note that all I/O operations done on this object will be *blocking*, but
/// they do not require the runtime to be active.
pub fn new(fd: fd_t) -> FileDesc {
FileDesc { fd: fd }
pub fn new(fd: fd_t, close_on_drop: bool) -> FileDesc {
FileDesc { fd: fd, close_on_drop: close_on_drop }
}
}
@ -137,7 +139,9 @@ impl Writer for FileDesc {
impl Drop for FileDesc {
#[fixed_stack_segment] #[inline(never)]
fn drop(&mut self) {
unsafe { libc::close(self.fd); }
if self.close_on_drop {
unsafe { libc::close(self.fd); }
}
}
}
@ -245,8 +249,8 @@ mod tests {
// opening or closing files.
unsafe {
let os::Pipe { input, out } = os::pipe();
let mut reader = FileDesc::new(input);
let mut writer = FileDesc::new(out);
let mut reader = FileDesc::new(input, true);
let mut writer = FileDesc::new(out, true);
writer.write(bytes!("test"));
let mut buf = [0u8, ..4];

6
src/libstd/rt/io/native/process.rs
View File

@ -105,9 +105,9 @@ impl Process {
Process {
pid: res.pid,
handle: res.handle,
input: in_pipe.map(|pipe| file::FileDesc::new(pipe.out)),
output: out_pipe.map(|pipe| file::FileDesc::new(pipe.input)),
error: err_pipe.map(|pipe| file::FileDesc::new(pipe.input)),
input: in_pipe.map(|pipe| file::FileDesc::new(pipe.out, true)),
output: out_pipe.map(|pipe| file::FileDesc::new(pipe.input, true)),
error: err_pipe.map(|pipe| file::FileDesc::new(pipe.input, true)),
exit_code: None,
}
}

8
src/libstd/rt/io/native/stdio.rs
View File

@ -36,10 +36,8 @@ pub struct StdIn {
impl StdIn {
/// Duplicates the stdin file descriptor, returning an io::Reader
#[fixed_stack_segment] #[inline(never)]
pub fn new() -> StdIn {
let fd = unsafe { libc::dup(libc::STDIN_FILENO) };
StdIn { fd: file::FileDesc::new(fd) }
StdIn { fd: file::FileDesc::new(libc::STDIN_FILENO, false) }
}
}
@ -54,10 +52,8 @@ pub struct StdOut {
impl StdOut {
/// Duplicates the specified file descriptor, returning an io::Writer
#[fixed_stack_segment] #[inline(never)]
pub fn new(fd: file::fd_t) -> StdOut {
let fd = unsafe { libc::dup(fd) };
StdOut { fd: file::FileDesc::new(fd) }
StdOut { fd: file::FileDesc::new(fd, false) }
}
}

18
src/libstd/rt/io/stdio.rs
View File

@ -33,7 +33,8 @@ use result::{Ok, Err};
use rt::io::buffered::LineBufferedWriter;
use rt::rtio::{IoFactory, RtioTTY, RtioFileStream, with_local_io,
CloseAsynchronously};
use super::{Reader, Writer, io_error, IoError, OtherIoError};
use super::{Reader, Writer, io_error, IoError, OtherIoError,
standard_error, EndOfFile};
// And so begins the tale of acquiring a uv handle to a stdio stream on all
// platforms in all situations. Our story begins by splitting the world into two
@ -203,6 +204,15 @@ impl Reader for StdReader {
File(ref mut file) => file.read(buf).map(|i| i as uint),
};
match ret {
// When reading a piped stdin, libuv will return 0-length reads when
// stdin reaches EOF. For pretty much all other streams it will
// return an actual EOF error, but apparently for stdin it's a
// little different. Hence, here we convert a 0 length read to an
// end-of-file indicator so the caller knows to stop reading.
Ok(0) => {
io_error::cond.raise(standard_error(EndOfFile));
None
}
Ok(amt) => Some(amt as uint),
Err(e) => {
io_error::cond.raise(e);
@ -277,12 +287,10 @@ impl StdWriter {
}
}
/// Returns whether this tream is attached to a TTY instance or not.
///
/// This is similar to libc's isatty() function
/// Returns whether this stream is attached to a TTY instance or not.
pub fn isatty(&self) -> bool {
match self.inner {
TTY(ref tty) => tty.isatty(),
TTY(*) => true,
File(*) => false,
}
}

14
src/libstd/rt/io/timer.rs
View File

@ -142,14 +142,10 @@ mod test {
fn oneshot_twice() {
do run_in_mt_newsched_task {
let mut timer = Timer::new().unwrap();
let port1 = timer.oneshot(100000000000);
let port1 = timer.oneshot(10000);
let port = timer.oneshot(1);
port.recv();
let port1 = Cell::new(port1);
let ret = do task::try {
port1.take().recv();
};
assert!(ret.is_err());
assert_eq!(port1.try_recv(), None);
}
}
@ -160,11 +156,7 @@ mod test {
let port = timer.oneshot(100000000000);
timer.sleep(1); // this should invalidate the port
let port = Cell::new(port);
let ret = do task::try {
port.take().recv();
};
assert!(ret.is_err());
assert_eq!(port.try_recv(), None);
}
}

12
src/libstd/rt/logging.rs
View File

@ -172,20 +172,18 @@ pub trait Logger {
/// This logger emits output to the stderr of the process, and contains a lazily
/// initialized event-loop driven handle to the stream.
pub struct StdErrLogger {
priv handle: Option<LineBufferedWriter<StdWriter>>,
priv handle: LineBufferedWriter<StdWriter>,
}
impl StdErrLogger {
pub fn new() -> StdErrLogger { StdErrLogger { handle: None } }
pub fn new() -> StdErrLogger {
StdErrLogger { handle: LineBufferedWriter::new(io::stderr()) }
}
}
impl Logger for StdErrLogger {
fn log(&mut self, args: &fmt::Arguments) {
// First time logging? Get a handle to the stderr of this process.
if self.handle.is_none() {
self.handle = Some(LineBufferedWriter::new(io::stderr()));
}
fmt::writeln(self.handle.get_mut_ref() as &mut io::Writer, args);
fmt::writeln(&mut self.handle as &mut io::Writer, args);
}
}

6
src/libstd/rt/macros.rs
View File

@ -34,7 +34,7 @@ macro_rules! rtassert (
( $arg:expr ) => ( {
if ::rt::util::ENFORCE_SANITY {
if !$arg {
rtabort!("assertion failed: {}", stringify!($arg));
rtabort!(" assertion failed: {}", stringify!($arg));
}
}
} )
@ -42,7 +42,7 @@ macro_rules! rtassert (
macro_rules! rtabort (
($($msg:tt)*) => ( {
::rt::util::abort(format!($($msg)*));
($($arg:tt)*) => ( {
::rt::util::abort(format!($($arg)*));
} )
)

17
src/libstd/rt/rtio.rs
View File

@ -24,11 +24,15 @@ use path::Path;
use super::io::{SeekStyle};
use super::io::{FileMode, FileAccess, FileStat, FilePermission};
pub trait Callback {
fn call(&mut self);
}
pub trait EventLoop {
fn run(&mut self);
fn callback(&mut self, ~fn());
fn pausible_idle_callback(&mut self) -> ~PausibleIdleCallback;
fn remote_callback(&mut self, ~fn()) -> ~RemoteCallback;
fn callback(&mut self, proc());
fn pausible_idle_callback(&mut self, ~Callback) -> ~PausibleIdleCallback;
fn remote_callback(&mut self, ~Callback) -> ~RemoteCallback;
/// The asynchronous I/O services. Not all event loops may provide one
// FIXME(#9382) this is an awful interface
@ -121,6 +125,8 @@ pub trait IoFactory {
fn fs_readlink(&mut self, path: &CString) -> Result<Path, IoError>;
fn fs_symlink(&mut self, src: &CString, dst: &CString) -> Result<(), IoError>;
fn fs_link(&mut self, src: &CString, dst: &CString) -> Result<(), IoError>;
fn fs_utime(&mut self, src: &CString, atime: u64, mtime: u64) ->
Result<(), IoError>;
// misc
fn timer_init(&mut self) -> Result<~RtioTimer, IoError>;
@ -209,8 +215,6 @@ pub trait RtioUnixListener {
pub trait RtioUnixAcceptor {
fn accept(&mut self) -> Result<~RtioPipe, IoError>;
fn accept_simultaneously(&mut self) -> Result<(), IoError>;
fn dont_accept_simultaneously(&mut self) -> Result<(), IoError>;
}
pub trait RtioTTY {
@ -218,14 +222,11 @@ pub trait RtioTTY {
fn write(&mut self, buf: &[u8]) -> Result<(), IoError>;
fn set_raw(&mut self, raw: bool) -> Result<(), IoError>;
fn get_winsize(&mut self) -> Result<(int, int), IoError>;
fn isatty(&self) -> bool;
}
pub trait PausibleIdleCallback {
fn start(&mut self, f: ~fn());
fn pause(&mut self);
fn resume(&mut self);
fn close(&mut self);
}
pub trait RtioSignal {}

24
src/libstd/rt/sched.rs
View File

@ -23,7 +23,7 @@ use super::message_queue::MessageQueue;
use rt::kill::BlockedTask;
use rt::local_ptr;
use rt::local::Local;
use rt::rtio::{RemoteCallback, PausibleIdleCallback};
use rt::rtio::{RemoteCallback, PausibleIdleCallback, Callback};
use borrow::{to_uint};
use cell::Cell;
use rand::{XorShiftRng, Rng, Rand};
@ -169,7 +169,8 @@ impl Scheduler {
pub fn bootstrap(mut ~self, task: ~Task) {
// Build an Idle callback.
self.idle_callback = Some(self.event_loop.pausible_idle_callback());
let cb = ~SchedRunner as ~Callback;
self.idle_callback = Some(self.event_loop.pausible_idle_callback(cb));
// Initialize the TLS key.
local_ptr::init_tls_key();
@ -184,7 +185,7 @@ impl Scheduler {
// Before starting our first task, make sure the idle callback
// is active. As we do not start in the sleep state this is
// important.
self.idle_callback.get_mut_ref().start(Scheduler::run_sched_once);
self.idle_callback.get_mut_ref().resume();
// Now, as far as all the scheduler state is concerned, we are
// inside the "scheduler" context. So we can act like the
@ -202,7 +203,7 @@ impl Scheduler {
// Close the idle callback.
let mut sched: ~Scheduler = Local::take();
sched.idle_callback.get_mut_ref().close();
sched.idle_callback.take();
// Make one go through the loop to run the close callback.
sched.run();
@ -454,8 +455,7 @@ impl Scheduler {
// * Task Routing Functions - Make sure tasks send up in the right
// place.
fn process_task(mut ~self, mut task: ~Task,
schedule_fn: SchedulingFn) {
fn process_task(mut ~self, mut task: ~Task, schedule_fn: SchedulingFn) {
rtdebug!("processing a task");
let home = task.take_unwrap_home();
@ -767,7 +767,7 @@ impl Scheduler {
}
pub fn make_handle(&mut self) -> SchedHandle {
let remote = self.event_loop.remote_callback(Scheduler::run_sched_once);
let remote = self.event_loop.remote_callback(~SchedRunner as ~Callback);
return SchedHandle {
remote: remote,
@ -779,7 +779,7 @@ impl Scheduler {
// Supporting types
type SchedulingFn = ~fn(~Scheduler, ~Task);
type SchedulingFn = extern "Rust" fn (~Scheduler, ~Task);
pub enum SchedMessage {
Wake,
@ -802,6 +802,14 @@ impl SchedHandle {
}
}
struct SchedRunner;
impl Callback for SchedRunner {
fn call(&mut self) {
Scheduler::run_sched_once();
}
}
struct CleanupJob {
task: ~Task,
f: UnsafeTaskReceiver

9
src/libstd/rt/task.rs
View File

@ -50,7 +50,7 @@ pub struct Task {
heap: LocalHeap,
priv gc: GarbageCollector,
storage: LocalStorage,
logger: StdErrLogger,
logger: Option<StdErrLogger>,
unwinder: Unwinder,
taskgroup: Option<Taskgroup>,
death: Death,
@ -180,7 +180,7 @@ impl Task {
heap: LocalHeap::new(),
gc: GarbageCollector,
storage: LocalStorage(None),
logger: StdErrLogger::new(),
logger: None,
unwinder: Unwinder { unwinding: false, cause: None },
taskgroup: None,
death: Death::new(),
@ -215,7 +215,7 @@ impl Task {
heap: LocalHeap::new(),
gc: GarbageCollector,
storage: LocalStorage(None),
logger: StdErrLogger::new(),
logger: None,
unwinder: Unwinder { unwinding: false, cause: None },
taskgroup: None,
death: Death::new(),
@ -238,7 +238,7 @@ impl Task {
heap: LocalHeap::new(),
gc: GarbageCollector,
storage: LocalStorage(None),
logger: StdErrLogger::new(),
logger: None,
unwinder: Unwinder { unwinding: false, cause: None },
taskgroup: None,
// FIXME(#7544) make watching optional
@ -320,6 +320,7 @@ impl Task {
}
None => {}
}
self.logger.take();
}
}

4
src/libstd/run.rs
View File

@ -436,13 +436,13 @@ mod tests {
}
fn writeclose(fd: c_int, s: &str) {
let mut writer = file::FileDesc::new(fd);
let mut writer = file::FileDesc::new(fd, true);
writer.write(s.as_bytes());
}
fn readclose(fd: c_int) -> ~str {
let mut res = ~[];
let mut reader = file::FileDesc::new(fd);
let mut reader = file::FileDesc::new(fd, true);
let mut buf = [0, ..1024];
loop {
match reader.read(buf) {

2
src/libuv

@ -1 +1 @@
Subproject commit d88cf5652a1afb23939da0bae86c70ec521b9921
Subproject commit 7ac7e0248b34732e9963cdb8e31f7e612d23d14b

527
src/rt/rust_uv.cpp
View File

@ -31,244 +31,16 @@ rust_uv_loop_new() {
return (void*)uv_loop_new();
}
extern "C" void
rust_uv_loop_delete(uv_loop_t* loop) {
// FIXME: This is a workaround for #1815. libev uses realloc(0) to
// free the loop, which valgrind doesn't like. We have suppressions
// to make valgrind ignore them.
//
// Valgrind also has a sanity check when collecting allocation backtraces
// that the stack pointer must be at least 512 bytes into the stack (at
// least 512 bytes of frames must have come before). When this is not
// the case it doesn't collect the backtrace.
//
// Unfortunately, with our spaghetti stacks that valgrind check triggers
// sometimes and we don't get the backtrace for the realloc(0), it
// fails to be suppressed, and it gets reported as 0 bytes lost
// from a malloc with no backtrace.
//
// This pads our stack with some extra space before deleting the loop
alloca(512);
uv_loop_delete(loop);
}
extern "C" void
rust_uv_loop_set_data(uv_loop_t* loop, void* data) {
loop->data = data;
}
extern "C" void
rust_uv_run(uv_loop_t* loop) {
uv_run(loop, UV_RUN_DEFAULT);
}
extern "C" void
rust_uv_close(uv_handle_t* handle, uv_close_cb cb) {
uv_close(handle, cb);
}
extern "C" void
rust_uv_walk(uv_loop_t* loop, uv_walk_cb cb, void* arg) {
uv_walk(loop, cb, arg);
}
extern "C" void
rust_uv_async_send(uv_async_t* handle) {
uv_async_send(handle);
}
extern "C" int
rust_uv_async_init(uv_loop_t* loop_handle,
uv_async_t* async_handle,
uv_async_cb cb) {
return uv_async_init(loop_handle, async_handle, cb);
}
extern "C" int
rust_uv_timer_init(uv_loop_t* loop, uv_timer_t* timer) {
return uv_timer_init(loop, timer);
}
extern "C" int
rust_uv_timer_start(uv_timer_t* the_timer, uv_timer_cb cb,
int64_t timeout, int64_t repeat) {
return uv_timer_start(the_timer, cb, timeout, repeat);
}
extern "C" int
rust_uv_timer_stop(uv_timer_t* the_timer) {
return uv_timer_stop(the_timer);
}
extern "C" int
rust_uv_tcp_init(uv_loop_t* loop, uv_tcp_t* handle) {
return uv_tcp_init(loop, handle);
}
extern "C" int
rust_uv_tcp_connect(uv_connect_t* connect_ptr,
uv_tcp_t* tcp_ptr,
uv_connect_cb cb,
sockaddr_in* addr_ptr) {
// FIXME ref #2064
sockaddr_in addr = *addr_ptr;
int result = uv_tcp_connect(connect_ptr, tcp_ptr, addr, cb);
return result;
}
extern "C" int
rust_uv_tcp_bind(uv_tcp_t* tcp_server, sockaddr_in* addr_ptr) {
// FIXME ref #2064
sockaddr_in addr = *addr_ptr;
return uv_tcp_bind(tcp_server, addr);
}
extern "C" int
rust_uv_tcp_connect6(uv_connect_t* connect_ptr,
uv_tcp_t* tcp_ptr,
uv_connect_cb cb,
sockaddr_in6* addr_ptr) {
// FIXME ref #2064
sockaddr_in6 addr = *addr_ptr;
int result = uv_tcp_connect6(connect_ptr, tcp_ptr, addr, cb);
return result;
}
extern "C" int
rust_uv_tcp_bind6
(uv_tcp_t* tcp_server, sockaddr_in6* addr_ptr) {
// FIXME ref #2064
sockaddr_in6 addr = *addr_ptr;
return uv_tcp_bind6(tcp_server, addr);
}
extern "C" int
rust_uv_tcp_getpeername
(uv_tcp_t* handle, sockaddr_storage* name) {
// sockaddr_storage is big enough to hold either
// sockaddr_in or sockaddr_in6
int namelen = sizeof(sockaddr_in);
return uv_tcp_getpeername(handle, (sockaddr*)name, &namelen);
}
extern "C" int
rust_uv_tcp_getsockname
(uv_tcp_t* handle, sockaddr_storage* name) {
// sockaddr_storage is big enough to hold either
// sockaddr_in or sockaddr_in6
int namelen = sizeof(sockaddr_storage);
return uv_tcp_getsockname(handle, (sockaddr*)name, &namelen);
}
extern "C" int
rust_uv_tcp_nodelay
(uv_tcp_t* handle, int enable) {
return uv_tcp_nodelay(handle, enable);
}
extern "C" int
rust_uv_tcp_keepalive
(uv_tcp_t* handle, int enable, unsigned int delay) {
return uv_tcp_keepalive(handle, enable, delay);
}
extern "C" int
rust_uv_tcp_simultaneous_accepts
(uv_tcp_t* handle, int enable) {
return uv_tcp_simultaneous_accepts(handle, enable);
}
extern "C" int
rust_uv_udp_init(uv_loop_t* loop, uv_udp_t* handle) {
return uv_udp_init(loop, handle);
}
extern "C" int
rust_uv_udp_bind(uv_udp_t* server, sockaddr_in* addr_ptr, unsigned flags) {
return uv_udp_bind(server, *addr_ptr, flags);
}
extern "C" int
rust_uv_udp_bind6(uv_udp_t* server, sockaddr_in6* addr_ptr, unsigned flags) {
return uv_udp_bind6(server, *addr_ptr, flags);
}
extern "C" int
rust_uv_udp_send(uv_udp_send_t* req, uv_udp_t* handle, uv_buf_t* buf_in,
int buf_cnt, sockaddr_in* addr_ptr, uv_udp_send_cb cb) {
return uv_udp_send(req, handle, buf_in, buf_cnt, *addr_ptr, cb);
}
extern "C" int
rust_uv_udp_send6(uv_udp_send_t* req, uv_udp_t* handle, uv_buf_t* buf_in,
int buf_cnt, sockaddr_in6* addr_ptr, uv_udp_send_cb cb) {
return uv_udp_send6(req, handle, buf_in, buf_cnt, *addr_ptr, cb);
}
extern "C" int
rust_uv_udp_recv_start(uv_udp_t* server, uv_alloc_cb on_alloc, uv_udp_recv_cb on_read) {
return uv_udp_recv_start(server, on_alloc, on_read);
}
extern "C" int
rust_uv_udp_recv_stop(uv_udp_t* server) {
return uv_udp_recv_stop(server);
}
extern "C" uv_udp_t*
rust_uv_get_udp_handle_from_send_req(uv_udp_send_t* send_req) {
return send_req->handle;
}
extern "C" int
rust_uv_udp_getsockname
(uv_udp_t* handle, sockaddr_storage* name) {
// sockaddr_storage is big enough to hold either
// sockaddr_in or sockaddr_in6
int namelen = sizeof(sockaddr_storage);
return uv_udp_getsockname(handle, (sockaddr*)name, &namelen);
}
extern "C" int
rust_uv_udp_set_membership
(uv_udp_t* handle, const char* m_addr, const char* i_addr, uv_membership membership) {
return uv_udp_set_membership(handle, m_addr, i_addr, membership);
}
extern "C" int
rust_uv_udp_set_multicast_loop
(uv_udp_t* handle, int on) {
return uv_udp_set_multicast_loop(handle, on);
}
extern "C" int
rust_uv_udp_set_multicast_ttl
(uv_udp_t* handle, int ttl) {
return uv_udp_set_multicast_ttl(handle, ttl);
}
extern "C" int
rust_uv_udp_set_ttl
(uv_udp_t* handle, int ttl) {
return uv_udp_set_ttl(handle, ttl);
}
extern "C" int
rust_uv_udp_set_broadcast
(uv_udp_t* handle, int on) {
return uv_udp_set_broadcast(handle, on);
}
extern "C" int
rust_uv_listen(uv_stream_t* stream, int backlog,
uv_connection_cb cb) {
return uv_listen(stream, backlog, cb);
}
extern "C" int
rust_uv_accept(uv_stream_t* server, uv_stream_t* client) {
return uv_accept(server, client);
}
extern "C" uv_stream_t*
rust_uv_get_stream_handle_from_connect_req(uv_connect_t* connect) {
return connect->handle;
@ -278,11 +50,6 @@ rust_uv_get_stream_handle_from_write_req(uv_write_t* write_req) {
return write_req->handle;
}
extern "C" void
rust_uv_buf_init(uv_buf_t* out_buf, char* base, size_t len) {
*out_buf = uv_buf_init(base, len);
}
extern "C" uv_loop_t*
rust_uv_get_loop_for_uv_handle(uv_handle_t* handle) {
return handle->loop;
@ -319,178 +86,50 @@ rust_uv_set_data_for_req(uv_req_t* req, void* data) {
req->data = data;
}
extern "C" char*
rust_uv_get_base_from_buf(uv_buf_t buf) {
return buf.base;
}
extern "C" size_t
rust_uv_get_len_from_buf(uv_buf_t buf) {
return buf.len;
}
extern "C" const char*
rust_uv_strerror(int err) {
return uv_strerror(err);
}
extern "C" const char*
rust_uv_err_name(int err) {
return uv_err_name(err);
}
extern "C" int
rust_uv_write(uv_write_t* req, uv_stream_t* handle,
uv_buf_t* bufs, int buf_cnt,
uv_write_cb cb) {
return uv_write(req, handle, bufs, buf_cnt, cb);
}
extern "C" int
rust_uv_read_start(uv_stream_t* stream, uv_alloc_cb on_alloc,
uv_read_cb on_read) {
return uv_read_start(stream, on_alloc, on_read);
rust_sockaddr_size() {
return sizeof(struct sockaddr_storage);
}
extern "C" int
rust_uv_read_stop(uv_stream_t* stream) {
return uv_read_stop(stream);
extern "C" struct sockaddr*
rust_malloc_ip4_addr(char *name, int port) {
struct sockaddr_in *addr = (struct sockaddr_in*) calloc(1, rust_sockaddr_size());
assert(addr != NULL);
addr->sin_port = htons(port);
assert(uv_inet_pton(AF_INET, name, &addr->sin_addr) == 0);
addr->sin_family = AF_INET;
return (struct sockaddr*) addr;
}
extern "C" struct sockaddr_in
rust_uv_ip4_addr(const char* ip, int port) {
struct sockaddr_in addr = uv_ip4_addr(ip, port);
return addr;
}
extern "C" struct sockaddr_in6
rust_uv_ip6_addr(const char* ip, int port) {
return uv_ip6_addr(ip, port);
extern "C" struct sockaddr*
rust_malloc_ip6_addr(char *name, int port) {
struct sockaddr_in6 *addr = (struct sockaddr_in6*) calloc(1, rust_sockaddr_size());
assert(addr != NULL);
addr->sin6_port = htons(port);
assert(uv_inet_pton(AF_INET6, name, &addr->sin6_addr) == 0);
addr->sin6_family = AF_INET6;
return (struct sockaddr*) addr;
}
extern "C" struct sockaddr_in*
rust_uv_ip4_addrp(const char* ip, int port) {
struct sockaddr_in addr = uv_ip4_addr(ip, port);
struct sockaddr_in *addrp = (sockaddr_in*)malloc(sizeof(struct sockaddr_in));
assert(addrp);
memcpy(addrp, &addr, sizeof(struct sockaddr_in));
return addrp;
}
extern "C" struct sockaddr_in6*
rust_uv_ip6_addrp(const char* ip, int port) {
struct sockaddr_in6 addr = uv_ip6_addr(ip, port);
struct sockaddr_in6 *addrp = (sockaddr_in6*)malloc(sizeof(struct sockaddr_in6));
assert(addrp);
memcpy(addrp, &addr, sizeof(struct sockaddr_in6));
return addrp;
}
extern "C" struct sockaddr_storage *
rust_uv_malloc_sockaddr_storage() {
struct sockaddr_storage *ss = (sockaddr_storage *)malloc(sizeof(struct sockaddr_storage));
return ss;
}
extern "C" void
rust_uv_free_sockaddr_storage(struct sockaddr_storage *ss) {
free(ss);
}
extern "C" void
rust_uv_free_ip4_addr(sockaddr_in *addrp) {
free(addrp);
}
extern "C" void
rust_uv_free_ip6_addr(sockaddr_in6 *addrp) {
free(addrp);
}
extern "C" int
rust_uv_ip4_name(struct sockaddr_in* src, char* dst, size_t size) {
return uv_ip4_name(src, dst, size);
}
extern "C" int
rust_uv_ip6_name(struct sockaddr_in6* src, char* dst, size_t size) {
int result = uv_ip6_name(src, dst, size);
return result;
}
extern "C" unsigned int
rust_uv_ip4_port(struct sockaddr_in* src) {
rust_ip4_port(struct sockaddr_in* src) {
return ntohs(src->sin_port);
}
extern "C" unsigned int
rust_uv_ip6_port(struct sockaddr_in6* src) {
rust_ip6_port(struct sockaddr_in6* src) {
return ntohs(src->sin6_port);
}
extern "C" int
rust_uv_getaddrinfo(uv_loop_t* loop, uv_getaddrinfo_t* handle,
uv_getaddrinfo_cb cb,
char* node, char* service,
addrinfo* hints) {
return uv_getaddrinfo(loop, handle, cb, node, service, hints);
}
extern "C" void
rust_uv_freeaddrinfo(addrinfo* res) {
uv_freeaddrinfo(res);
}
extern "C" int
rust_uv_is_ipv4_sockaddr(sockaddr* addr) {
rust_is_ipv4_sockaddr(sockaddr* addr) {
return addr->sa_family == AF_INET;
}
extern "C" int
rust_uv_is_ipv6_sockaddr(sockaddr* addr) {
rust_is_ipv6_sockaddr(sockaddr* addr) {
return addr->sa_family == AF_INET6;
}
extern "C" bool
rust_uv_is_ipv4_addrinfo(addrinfo* input) {
return input->ai_family == AF_INET;
}
extern "C" bool
rust_uv_is_ipv6_addrinfo(addrinfo* input) {
return input->ai_family == AF_INET6;
}
extern "C" addrinfo*
rust_uv_get_next_addrinfo(addrinfo* input) {
return input->ai_next;
}
extern "C" sockaddr_in*
rust_uv_addrinfo_as_sockaddr_in(addrinfo* input) {
return (sockaddr_in*)input->ai_addr;
}
extern "C" sockaddr_in6*
rust_uv_addrinfo_as_sockaddr_in6(addrinfo* input) {
return (sockaddr_in6*)input->ai_addr;
}
extern "C" int
rust_uv_idle_init(uv_loop_t* loop, uv_idle_t* idle) {
return uv_idle_init(loop, idle);
}
extern "C" int
rust_uv_idle_start(uv_idle_t* idle, uv_idle_cb cb) {
return uv_idle_start(idle, cb);
}
extern "C" int
rust_uv_idle_stop(uv_idle_t* idle) {
return uv_idle_stop(idle);
}
extern "C" size_t
rust_uv_handle_size(uintptr_t type) {
return uv_handle_size((uv_handle_type)type);
}
extern "C" size_t
rust_uv_req_size(uintptr_t type) {
return uv_req_size((uv_req_type)type);
}
extern "C" uintptr_t
rust_uv_handle_type_max() {
return UV_HANDLE_TYPE_MAX;
@ -501,33 +140,6 @@ rust_uv_req_type_max() {
return UV_REQ_TYPE_MAX;
}
extern "C" int
rust_uv_fs_open(uv_loop_t* loop, uv_fs_t* req, const char* path, int flags,
int mode, uv_fs_cb cb) {
return uv_fs_open(loop, req, path, flags, mode, cb);
}
extern "C" int
rust_uv_fs_unlink(uv_loop_t* loop, uv_fs_t* req, const char* path, uv_fs_cb cb) {
return uv_fs_unlink(loop, req, path, cb);
}
extern "C" int
rust_uv_fs_write(uv_loop_t* loop, uv_fs_t* req, uv_file fd, void* buf,
size_t len, int64_t offset, uv_fs_cb cb) {
return uv_fs_write(loop, req, fd, buf, len, offset, cb);
}
extern "C" int
rust_uv_fs_read(uv_loop_t* loop, uv_fs_t* req, uv_file fd, void* buf,
size_t len, int64_t offset, uv_fs_cb cb) {
return uv_fs_read(loop, req, fd, buf, len, offset, cb);
}
extern "C" int
rust_uv_fs_close(uv_loop_t* loop, uv_fs_t* req, uv_file fd, uv_fs_cb cb) {
return uv_fs_close(loop, req, fd, cb);
}
extern "C" void
rust_uv_fs_req_cleanup(uv_fs_t* req) {
uv_fs_req_cleanup(req);
}
extern "C" int
rust_uv_get_result_from_fs_req(uv_fs_t* req) {
return req->result;
@ -550,15 +162,6 @@ rust_uv_get_loop_from_getaddrinfo_req(uv_getaddrinfo_t* req) {
return req->loop;
}
extern "C" int
rust_uv_fs_stat(uv_loop_t* loop, uv_fs_t* req, const char* path, uv_fs_cb cb) {
return uv_fs_stat(loop, req, path, cb);
}
extern "C" int
rust_uv_fs_fstat(uv_loop_t* loop, uv_fs_t* req, uv_file file, uv_fs_cb cb) {
return uv_fs_fstat(loop, req, file, cb);
}
extern "C" void
rust_uv_populate_uv_stat(uv_fs_t* req_in, uv_stat_t* stat_out) {
stat_out->st_dev = req_in->statbuf.st_dev;
@ -583,39 +186,6 @@ rust_uv_populate_uv_stat(uv_fs_t* req_in, uv_stat_t* stat_out) {
stat_out->st_birthtim.tv_nsec = req_in->statbuf.st_birthtim.tv_nsec;
}
extern "C" int
rust_uv_fs_mkdir(uv_loop_t* loop, uv_fs_t* req, const char* path, int mode, uv_fs_cb cb) {
return uv_fs_mkdir(loop, req, path, mode, cb);
}
extern "C" int
rust_uv_fs_rmdir(uv_loop_t* loop, uv_fs_t* req, const char* path, uv_fs_cb cb) {
return uv_fs_rmdir(loop, req, path, cb);
}
extern "C" int
rust_uv_fs_readdir(uv_loop_t* loop, uv_fs_t* req, const char* path, int flags, uv_fs_cb cb) {
return uv_fs_readdir(loop, req, path, flags, cb);
}
extern "C" int
rust_uv_fs_rename(uv_loop_t *loop, uv_fs_t* req, const char *path,
const char *to, uv_fs_cb cb) {
return uv_fs_rename(loop, req, path, to, cb);
}
extern "C" int
rust_uv_fs_chmod(uv_loop_t* loop, uv_fs_t* req, const char* path, int mode, uv_fs_cb cb) {
return uv_fs_chmod(loop, req, path, mode, cb);
}
extern "C" int
rust_uv_spawn(uv_loop_t *loop, uv_process_t *p, uv_process_options_t options) {
return uv_spawn(loop, p, options);
}
extern "C" int
rust_uv_process_kill(uv_process_t *p, int signum) {
return uv_process_kill(p, signum);
}
extern "C" void
rust_set_stdio_container_flags(uv_stdio_container_t *c, int flags) {
c->flags = (uv_stdio_flags) flags;
@ -636,58 +206,7 @@ rust_uv_process_pid(uv_process_t* p) {
return p->pid;
}
extern "C" int
rust_uv_pipe_init(uv_loop_t *loop, uv_pipe_t* p, int ipc) {
return uv_pipe_init(loop, p, ipc);
}
extern "C" int
rust_uv_pipe_open(uv_pipe_t *pipe, int file) {
return uv_pipe_open(pipe, file);
}
extern "C" int
rust_uv_pipe_bind(uv_pipe_t *pipe, char *name) {
return uv_pipe_bind(pipe, name);
}
extern "C" void
rust_uv_pipe_connect(uv_connect_t *req, uv_pipe_t *handle,
char *name, uv_connect_cb cb) {
uv_pipe_connect(req, handle, name, cb);
}
extern "C" int
rust_uv_tty_init(uv_loop_t *loop, uv_tty_t *tty, int fd, int readable) {
return uv_tty_init(loop, tty, fd, readable);
}
extern "C" int
rust_uv_tty_set_mode(uv_tty_t *tty, int mode) {
return uv_tty_set_mode(tty, mode);
}
extern "C" int
rust_uv_tty_get_winsize(uv_tty_t *tty, int *width, int *height) {
return uv_tty_get_winsize(tty, width, height);
}
extern "C" int
rust_uv_guess_handle(int fd) {
return uv_guess_handle(fd);
}
extern "C" int
rust_uv_signal_init(uv_loop_t* loop, uv_signal_t* handle) {
return uv_signal_init(loop, handle);
}
extern "C" int
rust_uv_signal_start(uv_signal_t* handle, uv_signal_cb signal_cb, int signum) {
return uv_signal_start(handle, signal_cb, signum);
}
extern "C" int
rust_uv_signal_stop(uv_signal_t* handle) {
return uv_signal_stop(handle);
return uv_guess_handle(fd);
}

3
src/test/run-pass/closure-reform.rs
View File

@ -67,7 +67,8 @@ pub fn main() {
call_that(|x, y| *x + *y - z);
call_cramped(|| 1, || unsafe {
cast::transmute(&100)
static a: uint = 100;
cast::transmute(&a)
});
// External functions

10
src/test/run-pass/rtio-processes.rs
View File

@ -23,8 +23,8 @@
//
// See #9341
use std::rt::io;
use std::rt::io::process::{Process, ProcessConfig, CreatePipe, Ignored};
use std::rt::io::{Reader, Writer};
use std::str;
#[test]
@ -55,10 +55,10 @@ fn smoke_failure() {
cwd: None,
io: io,
};
let p = Process::new(args);
assert!(p.is_some());
let mut p = p.unwrap();
assert!(p.wait() != 0);
match io::result(|| Process::new(args)) {
Ok(*) => fail!(),
Err(*) => {}
}
}
#[test]