auto merge of #9721 : klutzy/rust/uv-net-read-fix, r=alexcrichton

See #9605 for detailed information.

This also fixes two tests of #8811.

src/libstd/rt/io/mod.rs

@@ -371,6 +371,7 @@ pub enum IoErrorKind {
     Closed,
     ConnectionRefused,
     ConnectionReset,
+    NotConnected,
     BrokenPipe,
     PathAlreadyExists,
     PathDoesntExist,
@@ -390,6 +391,7 @@ impl ToStr for IoErrorKind {
             Closed => ~"Closed",
             ConnectionRefused => ~"ConnectionRefused",
             ConnectionReset => ~"ConnectionReset",
+            NotConnected => ~"NotConnected",
             BrokenPipe => ~"BrokenPipe",
             PathAlreadyExists => ~"PathAlreadyExists",
             PathDoesntExist => ~"PathDoesntExist",

src/libstd/rt/io/net/tcp.rs

@@ -306,7 +306,6 @@ mod test {
     }
 
     #[test]
-    #[ignore(cfg(windows))] // FIXME #8811
     fn read_eof_twice_ip4() {
         do run_in_mt_newsched_task {
             let addr = next_test_ip4();
@@ -321,8 +320,16 @@ mod test {
                 let mut buf = [0];
                 let nread = stream.read(buf);
                 assert!(nread.is_none());
-                let nread = stream.read(buf);
-                assert!(nread.is_none());
+                do read_error::cond.trap(|e| {
+                    if cfg!(windows) {
+                        assert_eq!(e.kind, NotConnected);
+                    } else {
+                        fail2!();
+                    }
+                }).inside {
+                    let nread = stream.read(buf);
+                    assert!(nread.is_none());
+                }
             }
 
             do spawntask {
@@ -334,7 +341,6 @@ mod test {
     }
 
     #[test]
-    #[ignore(cfg(windows))] // FIXME #8811
     fn read_eof_twice_ip6() {
         do run_in_mt_newsched_task {
             let addr = next_test_ip6();
@@ -349,8 +355,16 @@ mod test {
                 let mut buf = [0];
                 let nread = stream.read(buf);
                 assert!(nread.is_none());
-                let nread = stream.read(buf);
-                assert!(nread.is_none());
+                do read_error::cond.trap(|e| {
+                    if cfg!(windows) {
+                        assert_eq!(e.kind, NotConnected);
+                    } else {
+                        fail2!();
+                    }
+                }).inside {
+                    let nread = stream.read(buf);
+                    assert!(nread.is_none());
+                }
             }
 
             do spawntask {

src/libstd/rt/uv/mod.rs

@@ -267,6 +267,7 @@ pub fn uv_error_to_io_error(uverr: UvError) -> IoError {
             EACCES => PermissionDenied,
             ECONNREFUSED => ConnectionRefused,
             ECONNRESET => ConnectionReset,
+            ENOTCONN => NotConnected,
             EPIPE => BrokenPipe,
             err => {
                 rtdebug!("uverr.code {}", err as int);

src/libstd/rt/uv/net.rs

@@ -14,7 +14,7 @@ use rt::uv::uvll;
 use rt::uv::uvll::*;
 use rt::uv::{AllocCallback, ConnectionCallback, ReadCallback, UdpReceiveCallback, UdpSendCallback};
 use rt::uv::{Loop, Watcher, Request, UvError, Buf, NativeHandle, NullCallback,
-             status_to_maybe_uv_error};
+             status_to_maybe_uv_error, vec_to_uv_buf};
 use rt::io::net::ip::{SocketAddr, Ipv4Addr, Ipv6Addr};
 use vec;
 use str;
@@ -147,7 +147,18 @@ impl StreamWatcher {
             data.read_cb = Some(cb);
         }
 
-        unsafe { uvll::read_start(self.native_handle(), alloc_cb, read_cb); }
+        let ret = unsafe { uvll::read_start(self.native_handle(), alloc_cb, read_cb) };
+
+        if ret != 0 {
+            // uvll::read_start failed, so read_cb will not be called.
+            // Call it manually for scheduling.
+            call_read_cb(self.native_handle(), ret as ssize_t);
+        }
+
+        fn call_read_cb(stream: *uvll::uv_stream_t, errno: ssize_t) {
+            #[fixed_stack_segment]; #[inline(never)];
+            read_cb(stream, errno, vec_to_uv_buf(~[]));
+        }
 
         extern fn alloc_cb(stream: *uvll::uv_stream_t, suggested_size: size_t) -> Buf {
             let mut stream_watcher: StreamWatcher = NativeHandle::from_native_handle(stream);

src/libstd/rt/uv/uvll.rs

@@ -53,6 +53,7 @@ pub mod errors {
     pub static EACCES: c_int = -4093;
     pub static ECONNREFUSED: c_int = -4079;
     pub static ECONNRESET: c_int = -4078;
+    pub static ENOTCONN: c_int = -4054;
     pub static EPIPE: c_int = -4048;
 }
 #[cfg(not(windows))]
@@ -63,6 +64,7 @@ pub mod errors {
     pub static EACCES: c_int = -libc::EACCES;
     pub static ECONNREFUSED: c_int = -libc::ECONNREFUSED;
     pub static ECONNRESET: c_int = -libc::ECONNRESET;
+    pub static ENOTCONN: c_int = -libc::ENOTCONN;
     pub static EPIPE: c_int = -libc::EPIPE;
 }