---

yaml
---
r: 151110
b: refs/heads/try2
c: 7c5d48a
h: refs/heads/master
v: v3

Author: pongad

[refs]

@@ -5,7 +5,7 @@ refs/heads/snap-stage3: 78a7676898d9f80ab540c6df5d4c9ce35bb50463
 refs/heads/try: 519addf6277dbafccbb4159db4b710c37eaa2ec5
 refs/tags/release-0.1: 1f5c5126e96c79d22cb7862f75304136e204f105
 refs/heads/ndm: f3868061cd7988080c30d6d5bf352a5a5fe2460b
-refs/heads/try2: 66486518d5e1e4f0850024386b66a7aa790fc32f
+refs/heads/try2: 7c5d48a09e3648adc3d4f4118808a903758a9dda
 refs/heads/dist-snap: ba4081a5a8573875fed17545846f6f6902c8ba8d
 refs/tags/release-0.2: c870d2dffb391e14efb05aa27898f1f6333a9596
 refs/tags/release-0.3: b5f0d0f648d9a6153664837026ba1be43d3e2503

branches/try2/src/liblibc/lib.rs

@@ -225,7 +225,7 @@ pub use funcs::bsd43::{shutdown};
 #[cfg(windows)] pub use consts::os::extra::{PIPE_UNLIMITED_INSTANCES, ERROR_ACCESS_DENIED};
 #[cfg(windows)] pub use consts::os::extra::{FILE_WRITE_ATTRIBUTES, FILE_READ_ATTRIBUTES};
 #[cfg(windows)] pub use consts::os::extra::{ERROR_PIPE_BUSY, ERROR_IO_PENDING};
-#[cfg(windows)] pub use consts::os::extra::{ERROR_PIPE_CONNECTED, WAIT_OBJECT_0};
+#[cfg(windows)] pub use consts::os::extra::{ERROR_PIPE_CONNECTED};
 #[cfg(windows)] pub use types::os::common::bsd44::{SOCKET};
 #[cfg(windows)] pub use types::os::common::posix01::{stat, utimbuf};
 #[cfg(windows)] pub use types::os::arch::extra::{HANDLE, BOOL, LPSECURITY_ATTRIBUTES};

branches/try2/src/libnative/io/c_win32.rs

@@ -59,6 +59,4 @@ extern "system" {
                       optname: libc::c_int,
                       optval: *mut libc::c_char,
                       optlen: *mut libc::c_int) -> libc::c_int;
-
-    pub fn CancelIo(hFile: libc::HANDLE) -> libc::BOOL;
 }

branches/try2/src/libnative/io/mod.rs

@@ -44,7 +44,6 @@ pub use self::process::Process;
 pub mod addrinfo;
 pub mod net;
 pub mod process;
-mod util;
 
 #[cfg(unix)]
 #[path = "file_unix.rs"]
@@ -178,9 +177,8 @@ impl rtio::IoFactory for IoFactory {
     fn unix_bind(&mut self, path: &CString) -> IoResult<~RtioUnixListener:Send> {
         pipe::UnixListener::bind(path).map(|s| ~s as ~RtioUnixListener:Send)
     }
-    fn unix_connect(&mut self, path: &CString,
-                    timeout: Option<u64>) -> IoResult<~RtioPipe:Send> {
-        pipe::UnixStream::connect(path, timeout).map(|s| ~s as ~RtioPipe:Send)
+    fn unix_connect(&mut self, path: &CString) -> IoResult<~RtioPipe:Send> {
+        pipe::UnixStream::connect(path).map(|s| ~s as ~RtioPipe:Send)
     }
     fn get_host_addresses(&mut self, host: Option<&str>, servname: Option<&str>,
                           hint: Option<ai::Hint>) -> IoResult<~[ai::Info]> {

branches/try2/src/libnative/io/net.rs

@@ -13,12 +13,13 @@ use std::cast;
 use std::io::net::ip;
 use std::io;
 use std::mem;
+use std::os;
+use std::ptr;
 use std::rt::rtio;
 use std::sync::arc::UnsafeArc;
 
 use super::{IoResult, retry, keep_going};
 use super::c;
-use super::util;
 
 ////////////////////////////////////////////////////////////////////////////////
 // sockaddr and misc bindings
@@ -117,8 +118,8 @@ fn setsockopt<T>(fd: sock_t, opt: libc::c_int, val: libc::c_int,
     }
 }
 
-pub fn getsockopt<T: Copy>(fd: sock_t, opt: libc::c_int,
-                           val: libc::c_int) -> IoResult<T> {
+fn getsockopt<T: Copy>(fd: sock_t, opt: libc::c_int,
+                       val: libc::c_int) -> IoResult<T> {
     unsafe {
         let mut slot: T = mem::init();
         let mut len = mem::size_of::<T>() as libc::socklen_t;
@@ -144,6 +145,21 @@ fn last_error() -> io::IoError {
     super::last_error()
 }
 
+fn ms_to_timeval(ms: u64) -> libc::timeval {
+    libc::timeval {
+        tv_sec: (ms / 1000) as libc::time_t,
+        tv_usec: ((ms % 1000) * 1000) as libc::suseconds_t,
+    }
+}
+
+fn timeout(desc: &'static str) -> io::IoError {
+    io::IoError {
+        kind: io::TimedOut,
+        desc: desc,
+        detail: None,
+    }
+}
+
 #[cfg(windows)] unsafe fn close(sock: sock_t) { let _ = libc::closesocket(sock); }
 #[cfg(unix)]    unsafe fn close(sock: sock_t) { let _ = libc::close(sock); }
 
@@ -254,7 +270,7 @@ impl TcpStream {
         let addrp = &addr as *_ as *libc::sockaddr;
         match timeout {
             Some(timeout) => {
-                try!(util::connect_timeout(fd, addrp, len, timeout));
+                try!(TcpStream::connect_timeout(fd, addrp, len, timeout));
                 Ok(ret)
             },
             None => {
@@ -266,6 +282,84 @@ impl TcpStream {
         }
     }
 
+    // See http://developerweb.net/viewtopic.php?id=3196 for where this is
+    // derived from.
+    fn connect_timeout(fd: sock_t,
+                       addrp: *libc::sockaddr,
+                       len: libc::socklen_t,
+                       timeout_ms: u64) -> IoResult<()> {
+        #[cfg(unix)]    use INPROGRESS = libc::EINPROGRESS;
+        #[cfg(windows)] use INPROGRESS = libc::WSAEINPROGRESS;
+        #[cfg(unix)]    use WOULDBLOCK = libc::EWOULDBLOCK;
+        #[cfg(windows)] use WOULDBLOCK = libc::WSAEWOULDBLOCK;
+
+        // Make sure the call to connect() doesn't block
+        try!(set_nonblocking(fd, true));
+
+        let ret = match unsafe { libc::connect(fd, addrp, len) } {
+            // If the connection is in progress, then we need to wait for it to
+            // finish (with a timeout). The current strategy for doing this is
+            // to use select() with a timeout.
+            -1 if os::errno() as int == INPROGRESS as int ||
+                  os::errno() as int == WOULDBLOCK as int => {
+                let mut set: c::fd_set = unsafe { mem::init() };
+                c::fd_set(&mut set, fd);
+                match await(fd, &mut set, timeout_ms) {
+                    0 => Err(timeout("connection timed out")),
+                    -1 => Err(last_error()),
+                    _ => {
+                        let err: libc::c_int = try!(
+                            getsockopt(fd, libc::SOL_SOCKET, libc::SO_ERROR));
+                        if err == 0 {
+                            Ok(())
+                        } else {
+                            Err(io::IoError::from_errno(err as uint, true))
+                        }
+                    }
+                }
+            }
+
+            -1 => Err(last_error()),
+            _ => Ok(()),
+        };
+
+        // be sure to turn blocking I/O back on
+        try!(set_nonblocking(fd, false));
+        return ret;
+
+        #[cfg(unix)]
+        fn set_nonblocking(fd: sock_t, nb: bool) -> IoResult<()> {
+            let set = nb as libc::c_int;
+            super::mkerr_libc(retry(|| unsafe { c::ioctl(fd, c::FIONBIO, &set) }))
+        }
+        #[cfg(windows)]
+        fn set_nonblocking(fd: sock_t, nb: bool) -> IoResult<()> {
+            let mut set = nb as libc::c_ulong;
+            if unsafe { c::ioctlsocket(fd, c::FIONBIO, &mut set) != 0 } {
+                Err(last_error())
+            } else {
+                Ok(())
+            }
+        }
+
+        #[cfg(unix)]
+        fn await(fd: sock_t, set: &mut c::fd_set, timeout: u64) -> libc::c_int {
+            let start = ::io::timer::now();
+            retry(|| unsafe {
+                // Recalculate the timeout each iteration (it is generally
+                // undefined what the value of the 'tv' is after select
+                // returns EINTR).
+                let tv = ms_to_timeval(timeout - (::io::timer::now() - start));
+                c::select(fd + 1, ptr::null(), &*set, ptr::null(), &tv)
+            })
+        }
+        #[cfg(windows)]
+        fn await(_fd: sock_t, set: &mut c::fd_set, timeout: u64) -> libc::c_int {
+            let tv = ms_to_timeval(timeout);
+            unsafe { c::select(1, ptr::null(), &*set, ptr::null(), &tv) }
+        }
+    }
+
     pub fn fd(&self) -> sock_t {
         // This unsafety is fine because it's just a read-only arc
         unsafe { (*self.inner.get()).fd }
@@ -439,7 +533,7 @@ impl TcpAcceptor {
 
     pub fn native_accept(&mut self) -> IoResult<TcpStream> {
         if self.deadline != 0 {
-            try!(util::accept_deadline(self.fd(), self.deadline));
+            try!(self.accept_deadline());
         }
         unsafe {
             let mut storage: libc::sockaddr_storage = mem::init();
@@ -456,6 +550,25 @@ impl TcpAcceptor {
             }
         }
     }
+
+    fn accept_deadline(&mut self) -> IoResult<()> {
+        let mut set: c::fd_set = unsafe { mem::init() };
+        c::fd_set(&mut set, self.fd());
+
+        match retry(|| {
+            // If we're past the deadline, then pass a 0 timeout to select() so
+            // we can poll the status of the socket.
+            let now = ::io::timer::now();
+            let ms = if self.deadline > now {0} else {self.deadline - now};
+            let tv = ms_to_timeval(ms);
+            let n = if cfg!(windows) {1} else {self.fd() as libc::c_int + 1};
+            unsafe { c::select(n, &set, ptr::null(), ptr::null(), &tv) }
+        }) {
+            -1 => Err(last_error()),
+            0 => Err(timeout("accept timed out")),
+            _ => return Ok(()),
+        }
+    }
 }
 
 impl rtio::RtioSocket for TcpAcceptor {
@@ -472,7 +585,10 @@ impl rtio::RtioTcpAcceptor for TcpAcceptor {
     fn accept_simultaneously(&mut self) -> IoResult<()> { Ok(()) }
     fn dont_accept_simultaneously(&mut self) -> IoResult<()> { Ok(()) }
     fn set_timeout(&mut self, timeout: Option<u64>) {
-        self.deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
+        self.deadline = match timeout {
+            None => 0,
+            Some(t) => ::io::timer::now() + t,
+        };
     }
 }