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

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.

Author: bors

mk/rt.mk

@@ -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 $$@

src/librustuv/addrinfo.rs

@@ -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,
 }
 
-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;
-    }
+struct Ctx {
+    slot: Option<BlockedTask>,
+    status: c_int,
+    addrinfo: Option<Addrinfo>,
+}
 
-    pub fn getaddrinfo(&mut self, loop_: &Loop, node: Option<&str>,
-                       service: Option<&str>, hints: Option<ai::Hint>,
-                       cb: GetAddrInfoCallback) {
+pub struct GetAddrInfoRequest;
 
+impl GetAddrInfoRequest {
+    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,37 +53,13 @@ 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| {
                 if hint.flags & (aival as uint) != 0 {
                     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);
+
+        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 };
+
+                do wait_until_woken_after(&mut cx.slot) {
+                    req.set_data(&cx);
+                }
 
-        self.get_req_data().getaddrinfo_cb = Some(wrapper_cb);
-
-        unsafe {
-            assert!(0 == uvll::getaddrinfo(loop_.native_handle(),
-                                           self.native_handle(),
-                                           getaddrinfo_cb,
-                                           c_node_ptr,
-                                           c_service_ptr,
-                                           hint_ptr));
-        }
-
-        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))
+                }
             }
-        }
-    }
+            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
-        };
-        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 });
 
-    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;
+            let sched: ~Scheduler = Local::take();
+            sched.resume_blocked_task_immediately(cx.slot.take_unwrap());
         }
     }
+}
 
-    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();
     }
 }