Make the uv bindings resilient to linked failure

In the ideal world, uv I/O could be canceled safely at any time. In reality,
however, we are unable to do this. Right now linked failure is fairly flaky as
implemented in the runtime, making it very difficult to test whether the linked
failure mechanisms inside of the uv bindings are ready for this kind of
interaction.

Right now, all constructors will execute in a task::unkillable block, and all
homing I/O operations will prevent linked failure in the duration of the homing
operation. What this means is that tasks which perform I/O are still susceptible
to linked failure, but the I/O operations themselves will never get interrupted.
Instead, the linked failure will be received at the edge of the I/O operation.

Author: alexcrichton

src/librustuv/addrinfo.rs

@@ -9,15 +9,14 @@
 // except according to those terms.
 
 use ai = std::rt::io::net::addrinfo;
-use std::cast;
 use std::libc::c_int;
 use std::ptr::null;
 use std::rt::BlockedTask;
 use std::rt::local::Local;
 use std::rt::sched::Scheduler;
 
 use net;
-use super::{Loop, UvError, Request};
+use super::{Loop, UvError, Request, wait_until_woken_after};
 use uvll;
 
 struct Addrinfo {
@@ -76,20 +75,19 @@ impl GetAddrInfoRequest {
             }
         });
         let hint_ptr = hint.as_ref().map_default(null(), |x| x as *uvll::addrinfo);
-        let req = Request::new(uvll::UV_GETADDRINFO);
+        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 };
-                req.set_data(&cx);
-                req.defuse();
-                let scheduler: ~Scheduler = Local::take();
-                do scheduler.deschedule_running_task_and_then |_, task| {
-                    cx.slot = Some(task);
+
+                do wait_until_woken_after(&mut cx.slot) {
+                    req.set_data(&cx);
                 }
 
                 match cx.status {
@@ -105,8 +103,8 @@ impl GetAddrInfoRequest {
                                  status: c_int,
                                  res: *uvll::addrinfo) {
             let req = Request::wrap(req);
-            if status == uvll::ECANCELED { return }
-            let cx: &mut Ctx = unsafe { cast::transmute(req.get_data()) };
+            assert!(status != uvll::ECANCELED);
+            let cx: &mut Ctx = unsafe { req.get_data() };
             cx.status = status;
             cx.addrinfo = Some(Addrinfo { handle: res });
 
@@ -191,25 +189,23 @@ pub fn accum_addrinfo(addr: &Addrinfo) -> ~[ai::Info] {
 mod test {
     use std::rt::io::net::ip::{SocketAddr, Ipv4Addr};
     use super::*;
-    use super::super::run_uv_loop;
+    use super::super::local_loop;
 
     #[test]
     fn getaddrinfo_test() {
-        do run_uv_loop |l| {
-            match GetAddrInfoRequest::run(l, 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);
+        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;
                 }
-                Err(e) => fail!("{:?}", e),
+                assert!(found_local);
             }
+            Err(e) => fail!("{:?}", e),
         }
     }
 }

src/librustuv/async.rs

@@ -131,11 +131,12 @@ mod test_remote {
     use std::rt::tube::Tube;
 
     use super::*;
-    use super::super::run_uv_loop;
+    use super::super::local_loop;
 
-    // Make sure that we can fire watchers in remote threads
+    // Make sure that we can fire watchers in remote threads and that they
+    // actually trigger what they say they will.
     #[test]
-    fn test_uv_remote() {
+    fn smoke_test() {
         struct MyCallback(Option<Tube<int>>);
         impl Callback for MyCallback {
             fn call(&mut self) {
@@ -147,35 +148,15 @@ mod test_remote {
             }
         }
 
-        do run_uv_loop |l| {
-            let mut tube = Tube::new();
-            let cb = ~MyCallback(Some(tube.clone()));
-            let watcher = Cell::new(AsyncWatcher::new(l, cb as ~Callback));
-
-            let thread = do Thread::start {
-                watcher.take().fire();
-            };
+        let mut tube = Tube::new();
+        let cb = ~MyCallback(Some(tube.clone()));
+        let watcher = Cell::new(AsyncWatcher::new(local_loop(), cb as ~Callback));
 
-            assert_eq!(tube.recv(), 1);
-            thread.join();
-        }
-    }
-
-    #[test]
-    fn smoke_test() {
-        static mut hits: uint = 0;
+        let thread = do Thread::start {
+            watcher.take().fire();
+        };
 
-        struct MyCallback;
-        impl Callback for MyCallback {
-            fn call(&mut self) {
-                unsafe { hits += 1; }
-            }
-        }
-
-        do run_uv_loop |l| {
-            let mut watcher = AsyncWatcher::new(l, ~MyCallback as ~Callback);
-            watcher.fire();
-        }
-        assert!(unsafe { hits > 0 });
+        assert_eq!(tube.recv(), 1);
+        thread.join();
     }
 }

src/librustuv/file.rs

@@ -15,14 +15,14 @@ use std::cast;
 use std::libc::{c_int, c_char, c_void, c_uint};
 use std::libc;
 use std::rt::BlockedTask;
-use std::rt::io;
 use std::rt::io::{FileStat, IoError};
-use std::rt::rtio;
+use std::rt::io;
 use std::rt::local::Local;
+use std::rt::rtio;
 use std::rt::sched::{Scheduler, SchedHandle};
 use std::vec;
 
-use super::{Loop, UvError, uv_error_to_io_error};
+use super::{Loop, UvError, uv_error_to_io_error, wait_until_woken_after};
 use uvio::HomingIO;
 use uvll;
 
@@ -305,10 +305,8 @@ fn execute(f: &fn(*uvll::uv_fs_t, uvll::uv_fs_cb) -> c_int)
         0 => {
             req.fired = true;
             let mut slot = None;
-            unsafe { uvll::set_data_for_req(req.req, &slot) }
-            let sched: ~Scheduler = Local::take();
-            do sched.deschedule_running_task_and_then |_, task| {
-                slot = Some(task);
+            do wait_until_woken_after(&mut slot) {
+                unsafe { uvll::set_data_for_req(req.req, &slot) }
             }
             match req.get_result() {
                 n if n < 0 => Err(UvError(n)),
@@ -454,123 +452,113 @@ mod test {
     use std::str;
     use std::vec;
     use super::*;
-    use super::super::{run_uv_loop};
+    use l = super::super::local_loop;
 
     #[test]
     fn file_test_full_simple_sync() {
-        do run_uv_loop |l| {
-            let create_flags = O_RDWR | O_CREAT;
-            let read_flags = O_RDONLY;
-            let mode = S_IWUSR | S_IRUSR;
-            let path_str = "./tmp/file_full_simple_sync.txt";
-
-            {
-                // open/create
-                let result = FsRequest::open(l, &path_str.to_c_str(),
-                                             create_flags as int, mode as int);
-                assert!(result.is_ok());
-                let result = result.unwrap();
-                let fd = result.fd;
-
-                // write
-                let result = FsRequest::write(l, fd, "hello".as_bytes(), -1);
-                assert!(result.is_ok());
-            }
+        let create_flags = O_RDWR | O_CREAT;
+        let read_flags = O_RDONLY;
+        let mode = S_IWUSR | S_IRUSR;
+        let path_str = "./tmp/file_full_simple_sync.txt";
+
+        {
+            // open/create
+            let result = FsRequest::open(l(), &path_str.to_c_str(),
+                                         create_flags as int, mode as int);
+            assert!(result.is_ok());
+            let result = result.unwrap();
+            let fd = result.fd;
 
-            {
-                // re-open
-                let result = FsRequest::open(l, &path_str.to_c_str(),
-                                             read_flags as int, 0);
-                assert!(result.is_ok());
-                let result = result.unwrap();
-                let fd = result.fd;
-
-                // read
-                let mut read_mem = vec::from_elem(1000, 0u8);
-                let result = FsRequest::read(l, fd, read_mem, 0);
-                assert!(result.is_ok());
-
-                let nread = result.unwrap();
-                assert!(nread > 0);
-                let read_str = str::from_utf8(read_mem.slice(0, nread as uint));
-                assert_eq!(read_str, ~"hello");
-            }
-            // unlink
-            let result = FsRequest::unlink(l, &path_str.to_c_str());
+            // write
+            let result = FsRequest::write(l(), fd, "hello".as_bytes(), -1);
             assert!(result.is_ok());
         }
+
+        {
+            // re-open
+            let result = FsRequest::open(l(), &path_str.to_c_str(),
+                                         read_flags as int, 0);
+            assert!(result.is_ok());
+            let result = result.unwrap();
+            let fd = result.fd;
+
+            // read
+            let mut read_mem = vec::from_elem(1000, 0u8);
+            let result = FsRequest::read(l(), fd, read_mem, 0);
+            assert!(result.is_ok());
+
+            let nread = result.unwrap();
+            assert!(nread > 0);
+            let read_str = str::from_utf8(read_mem.slice(0, nread as uint));
+            assert_eq!(read_str, ~"hello");
+        }
+        // unlink
+        let result = FsRequest::unlink(l(), &path_str.to_c_str());
+        assert!(result.is_ok());
     }
 
     #[test]
     fn file_test_stat() {
-        do run_uv_loop |l| {
-            let path = &"./tmp/file_test_stat_simple".to_c_str();
-            let create_flags = (O_RDWR | O_CREAT) as int;
-            let mode = (S_IWUSR | S_IRUSR) as int;
+        let path = &"./tmp/file_test_stat_simple".to_c_str();
+        let create_flags = (O_RDWR | O_CREAT) as int;
+        let mode = (S_IWUSR | S_IRUSR) as int;
 
-            let result = FsRequest::open(l, path, create_flags, mode);
-            assert!(result.is_ok());
-            let file = result.unwrap();
+        let result = FsRequest::open(l(), path, create_flags, mode);
+        assert!(result.is_ok());
+        let file = result.unwrap();
 
-            let result = FsRequest::write(l, file.fd, "hello".as_bytes(), 0);
-            assert!(result.is_ok());
+        let result = FsRequest::write(l(), file.fd, "hello".as_bytes(), 0);
+        assert!(result.is_ok());
 
-            let result = FsRequest::stat(l, path);
-            assert!(result.is_ok());
-            assert_eq!(result.unwrap().size, 5);
+        let result = FsRequest::stat(l(), path);
+        assert!(result.is_ok());
+        assert_eq!(result.unwrap().size, 5);
 
-            fn free<T>(_: T) {}
-            free(file);
+        fn free<T>(_: T) {}
+        free(file);
 
-            let result = FsRequest::unlink(l, path);
-            assert!(result.is_ok());
-        }
+        let result = FsRequest::unlink(l(), path);
+        assert!(result.is_ok());
     }
 
     #[test]
     fn file_test_mk_rm_dir() {
-        do run_uv_loop |l| {
-            let path = &"./tmp/mk_rm_dir".to_c_str();
-            let mode = S_IWUSR | S_IRUSR;
+        let path = &"./tmp/mk_rm_dir".to_c_str();
+        let mode = S_IWUSR | S_IRUSR;
 
-            let result = FsRequest::mkdir(l, path, mode);
-            assert!(result.is_ok());
+        let result = FsRequest::mkdir(l(), path, mode);
+        assert!(result.is_ok());
 
-            let result = FsRequest::stat(l, path);
-            assert!(result.is_ok());
-            assert!(result.unwrap().kind == io::TypeDirectory);
+        let result = FsRequest::stat(l(), path);
+        assert!(result.is_ok());
+        assert!(result.unwrap().kind == io::TypeDirectory);
 
-            let result = FsRequest::rmdir(l, path);
-            assert!(result.is_ok());
+        let result = FsRequest::rmdir(l(), path);
+        assert!(result.is_ok());
 
-            let result = FsRequest::stat(l, path);
-            assert!(result.is_err());
-        }
+        let result = FsRequest::stat(l(), path);
+        assert!(result.is_err());
     }
 
     #[test]
     fn file_test_mkdir_chokes_on_double_create() {
-        do run_uv_loop |l| {
-            let path = &"./tmp/double_create_dir".to_c_str();
-            let mode = S_IWUSR | S_IRUSR;
-
-            let result = FsRequest::stat(l, path);
-            assert!(result.is_err(), "{:?}", result);
-            let result = FsRequest::mkdir(l, path, mode as c_int);
-            assert!(result.is_ok(), "{:?}", result);
-            let result = FsRequest::mkdir(l, path, mode as c_int);
-            assert!(result.is_err(), "{:?}", result);
-            let result = FsRequest::rmdir(l, path);
-            assert!(result.is_ok(), "{:?}", result);
-        }
+        let path = &"./tmp/double_create_dir".to_c_str();
+        let mode = S_IWUSR | S_IRUSR;
+
+        let result = FsRequest::stat(l(), path);
+        assert!(result.is_err(), "{:?}", result);
+        let result = FsRequest::mkdir(l(), path, mode as c_int);
+        assert!(result.is_ok(), "{:?}", result);
+        let result = FsRequest::mkdir(l(), path, mode as c_int);
+        assert!(result.is_err(), "{:?}", result);
+        let result = FsRequest::rmdir(l(), path);
+        assert!(result.is_ok(), "{:?}", result);
     }
 
     #[test]
     fn file_test_rmdir_chokes_on_nonexistant_path() {
-        do run_uv_loop |l| {
-            let path = &"./tmp/never_existed_dir".to_c_str();
-            let result = FsRequest::rmdir(l, path);
-            assert!(result.is_err());
-        }
+        let path = &"./tmp/never_existed_dir".to_c_str();
+        let result = FsRequest::rmdir(l(), path);
+        assert!(result.is_err());
     }
 }