std: Add an experimental connect_timeout function

This adds a `TcpStream::connect_timeout` function in order to assist opening
connections with a timeout (cc #13523). There isn't really much design space for
this specific operation (unlike timing out normal blocking reads/writes), so I
am fairly confident that this is the correct interface for this function.

The function is marked #[experimental] because it takes a u64 timeout argument,
and the u64 type is likely to change in the future.

6 files changed, 284 insertions, 112 deletions

diff --git a/src/libnative/io/c_unix.rs b/src/libnative/io/c_unix.rs
new file mode 100644
index 00000000000..e2bf515a1e5
--- /dev/null
+++ b/src/libnative/io/c_unix.rs
@@ -0,0 +1,76 @@
+// Copyright 2014 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.
+
+//! C definitions used by libnative that don't belong in liblibc
+
+pub use self::select::fd_set;
+
+use libc;
+
+#[cfg(target_os = "macos")]
+#[cfg(target_os = "freebsd")]
+pub static FIONBIO: libc::c_ulong = 0x8004667e;
+#[cfg(target_os = "linux")]
+#[cfg(target_os = "android")]
+pub static FIONBIO: libc::c_ulong = 0x5421;
+#[cfg(target_os = "macos")]
+#[cfg(target_os = "freebsd")]
+pub static FIOCLEX: libc::c_ulong = 0x20006601;
+#[cfg(target_os = "linux")]
+#[cfg(target_os = "android")]
+pub static FIOCLEX: libc::c_ulong = 0x5451;
+
+extern {
+    pub fn gettimeofday(timeval: *mut libc::timeval,
+                        tzp: *libc::c_void) -> libc::c_int;
+    pub fn select(nfds: libc::c_int,
+                  readfds: *fd_set,
+                  writefds: *fd_set,
+                  errorfds: *fd_set,
+                  timeout: *libc::timeval) -> libc::c_int;
+    pub fn getsockopt(sockfd: libc::c_int,
+                      level: libc::c_int,
+                      optname: libc::c_int,
+                      optval: *mut libc::c_void,
+                      optlen: *mut libc::socklen_t) -> libc::c_int;
+    pub fn ioctl(fd: libc::c_int, req: libc::c_ulong, ...) -> libc::c_int;
+
+}
+
+#[cfg(target_os = "macos")]
+mod select {
+    pub static FD_SETSIZE: uint = 1024;
+
+    pub struct fd_set {
+        fds_bits: [i32, ..(FD_SETSIZE / 32)]
+    }
+
+    pub fn fd_set(set: &mut fd_set, fd: i32) {
+        set.fds_bits[(fd / 32) as uint] |= 1 << (fd % 32);
+    }
+}
+
+#[cfg(target_os = "android")]
+#[cfg(target_os = "freebsd")]
+#[cfg(target_os = "linux")]
+mod select {
+    use std::uint;
+
+    pub static FD_SETSIZE: uint = 1024;
+
+    pub struct fd_set {
+        fds_bits: [uint, ..(FD_SETSIZE / uint::BITS)]
+    }
+
+    pub fn fd_set(set: &mut fd_set, fd: i32) {
+        let fd = fd as uint;
+        set.fds_bits[fd / uint::BITS] |= 1 << (fd % uint::BITS);
+    }
+}
diff --git a/src/libnative/io/c_win32.rs b/src/libnative/io/c_win32.rs
new file mode 100644
index 00000000000..8d75a673914
--- /dev/null
+++ b/src/libnative/io/c_win32.rs
@@ -0,0 +1,62 @@
+// Copyright 2014 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.
+
+//! C definitions used by libnative that don't belong in liblibc
+
+#![allow(type_overflow)]
+
+use libc;
+
+pub static WSADESCRIPTION_LEN: uint = 256;
+pub static WSASYS_STATUS_LEN: uint = 128;
+pub static FIONBIO: libc::c_long = 0x8004667e;
+static FD_SETSIZE: uint = 64;
+
+pub struct WSADATA {
+    pub wVersion: libc::WORD,
+    pub wHighVersion: libc::WORD,
+    pub szDescription: [u8, ..WSADESCRIPTION_LEN + 1],
+    pub szSystemStatus: [u8, ..WSASYS_STATUS_LEN + 1],
+    pub iMaxSockets: u16,
+    pub iMaxUdpDg: u16,
+    pub lpVendorInfo: *u8,
+}
+
+pub type LPWSADATA = *mut WSADATA;
+
+pub struct fd_set {
+    fd_count: libc::c_uint,
+    fd_array: [libc::SOCKET, ..FD_SETSIZE],
+}
+
+pub fn fd_set(set: &mut fd_set, s: libc::SOCKET) {
+    set.fd_array[set.fd_count as uint] = s;
+    set.fd_count += 1;
+}
+
+#[link(name = "ws2_32")]
+extern "system" {
+    pub fn WSAStartup(wVersionRequested: libc::WORD,
+                      lpWSAData: LPWSADATA) -> libc::c_int;
+    pub fn WSAGetLastError() -> libc::c_int;
+
+    pub fn ioctlsocket(s: libc::SOCKET, cmd: libc::c_long,
+                       argp: *mut libc::c_ulong) -> libc::c_int;
+    pub fn select(nfds: libc::c_int,
+                  readfds: *mut fd_set,
+                  writefds: *mut fd_set,
+                  exceptfds: *mut fd_set,
+                  timeout: *libc::timeval) -> libc::c_int;
+    pub fn getsockopt(sockfd: libc::SOCKET,
+                      level: libc::c_int,
+                      optname: libc::c_int,
+                      optval: *mut libc::c_char,
+                      optlen: *mut libc::c_int) -> libc::c_int;
+}
diff --git a/src/libnative/io/mod.rs b/src/libnative/io/mod.rs
index 78d17bc8d74..19cb5c5f1d4 100644
--- a/src/libnative/io/mod.rs
+++ b/src/libnative/io/mod.rs
@@ -71,6 +71,9 @@ pub mod pipe;
 #[path = "pipe_win32.rs"]
 pub mod pipe;
 
+#[cfg(unix)]    #[path = "c_unix.rs"]  mod c;
+#[cfg(windows)] #[path = "c_win32.rs"] mod c;
+
 mod timer_helper;
 
 pub type IoResult<T> = Result<T, IoError>;
@@ -161,8 +164,9 @@ impl IoFactory {
 
 impl rtio::IoFactory for IoFactory {
     // networking
-    fn tcp_connect(&mut self, addr: SocketAddr) -> IoResult<~RtioTcpStream:Send> {
-        net::TcpStream::connect(addr).map(|s| ~s as ~RtioTcpStream:Send)
+    fn tcp_connect(&mut self, addr: SocketAddr,
+                   timeout: Option<u64>) -> IoResult<~RtioTcpStream:Send> {
+        net::TcpStream::connect(addr, timeout).map(|s| ~s as ~RtioTcpStream:Send)
     }
     fn tcp_bind(&mut self, addr: SocketAddr) -> IoResult<~RtioTcpListener:Send> {
         net::TcpListener::bind(addr).map(|s| ~s as ~RtioTcpListener:Send)
diff --git a/src/libnative/io/net.rs b/src/libnative/io/net.rs
index 2e64b82a84a..be597761b1a 100644
--- a/src/libnative/io/net.rs
+++ b/src/libnative/io/net.rs
@@ -8,15 +8,17 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
+use libc;
 use std::cast;
 use std::io::net::ip;
 use std::io;
-use libc;
 use std::mem;
+use std::ptr;
 use std::rt::rtio;
 use std::sync::arc::UnsafeArc;
 
 use super::{IoResult, retry, keep_going};
+use super::c;
 
 ////////////////////////////////////////////////////////////////////////////////
 // sockaddr and misc bindings
@@ -115,12 +117,26 @@ fn setsockopt<T>(fd: sock_t, opt: libc::c_int, val: libc::c_int,
     }
 }
 
+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;
+        let ret = c::getsockopt(fd, opt, val,
+                                &mut slot as *mut _ as *mut _,
+                                &mut len);
+        if ret != 0 {
+            Err(last_error())
+        } else {
+            assert!(len as uint == mem::size_of::<T>());
+            Ok(slot)
+        }
+    }
+}
+
 #[cfg(windows)]
 fn last_error() -> io::IoError {
-    extern "system" {
-        fn WSAGetLastError() -> libc::c_int;
-    }
-    io::IoError::from_errno(unsafe { WSAGetLastError() } as uint, true)
+    io::IoError::from_errno(unsafe { c::WSAGetLastError() } as uint, true)
 }
 
 #[cfg(not(windows))]
@@ -197,24 +213,6 @@ pub fn init() {}
 
 #[cfg(windows)]
 pub fn init() {
-    static WSADESCRIPTION_LEN: uint = 256;
-    static WSASYS_STATUS_LEN: uint = 128;
-    struct WSADATA {
-        wVersion: libc::WORD,
-        wHighVersion: libc::WORD,
-        szDescription: [u8, ..WSADESCRIPTION_LEN + 1],
-        szSystemStatus: [u8, ..WSASYS_STATUS_LEN + 1],
-        iMaxSockets: u16,
-        iMaxUdpDg: u16,
-        lpVendorInfo: *u8,
-    }
-    type LPWSADATA = *mut WSADATA;
-
-    #[link(name = "ws2_32")]
-    extern "system" {
-        fn WSAStartup(wVersionRequested: libc::WORD,
-                       lpWSAData: LPWSADATA) -> libc::c_int;
-    }
 
     unsafe {
         use std::unstable::mutex::{StaticNativeMutex, NATIVE_MUTEX_INIT};
@@ -223,9 +221,9 @@ pub fn init() {
 
         let _guard = LOCK.lock();
         if !INITIALIZED {
-            let mut data: WSADATA = mem::init();
-            let ret = WSAStartup(0x202,      // version 2.2
-                                 &mut data);
+            let mut data: c::WSADATA = mem::init();
+            let ret = c::WSAStartup(0x202,      // version 2.2
+                                    &mut data);
             assert_eq!(ret, 0);
             INITIALIZED = true;
         }
@@ -245,22 +243,118 @@ struct Inner {
 }
 
 impl TcpStream {
-    pub fn connect(addr: ip::SocketAddr) -> IoResult<TcpStream> {
-        unsafe {
-            socket(addr, libc::SOCK_STREAM).and_then(|fd| {
-                let (addr, len) = addr_to_sockaddr(addr);
-                let addrp = &addr as *libc::sockaddr_storage;
-                let inner = Inner { fd: fd };
-                let ret = TcpStream { inner: UnsafeArc::new(inner) };
-                match retry(|| {
-                    libc::connect(fd, addrp as *libc::sockaddr,
-                                  len as libc::socklen_t)
-                }) {
+    pub fn connect(addr: ip::SocketAddr,
+                   timeout: Option<u64>) -> IoResult<TcpStream> {
+        let fd = try!(socket(addr, libc::SOCK_STREAM));
+        let (addr, len) = addr_to_sockaddr(addr);
+        let inner = Inner { fd: fd };
+        let ret = TcpStream { inner: UnsafeArc::new(inner) };
+
+        let len = len as libc::socklen_t;
+        let addrp = &addr as *_ as *libc::sockaddr;
+        match timeout {
+            Some(timeout) => {
+                try!(TcpStream::connect_timeout(fd, addrp, len, timeout));
+                Ok(ret)
+            },
+            None => {
+                match retry(|| unsafe { libc::connect(fd, addrp, len) }) {
                     -1 => Err(last_error()),
                     _ => Ok(ret),
                 }
+            }
+        }
+    }
+
+    // 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: u64) -> IoResult<()> {
+        use std::os;
+        #[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) {
+                    0 => Err(io::IoError {
+                        kind: io::TimedOut,
+                        desc: "connection timed out",
+                        detail: None,
+                    }),
+                    -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 timeout = timeout - (::io::timer::now() - start);
+                let tv = libc::timeval {
+                    tv_sec: (timeout / 1000) as libc::time_t,
+                    tv_usec: ((timeout % 1000) * 1000) as libc::suseconds_t,
+                };
+                c::select(fd + 1, ptr::null(), set as *mut _ as *_,
+                          ptr::null(), &tv)
             })
         }
+        #[cfg(windows)]
+        fn await(_fd: sock_t, set: &mut c::fd_set, timeout: u64) -> libc::c_int {
+            let tv = libc::timeval {
+                tv_sec: (timeout / 1000) as libc::time_t,
+                tv_usec: ((timeout % 1000) * 1000) as libc::suseconds_t,
+            };
+            unsafe { c::select(1, ptr::mut_null(), set, ptr::mut_null(), &tv) }
+        }
     }
 
     pub fn fd(&self) -> sock_t {
diff --git a/src/libnative/io/process.rs b/src/libnative/io/process.rs
index a29a5b631c6..efdab990d18 100644
--- a/src/libnative/io/process.rs
+++ b/src/libnative/io/process.rs
@@ -454,7 +454,7 @@ fn spawn_process_os(config: p::ProcessConfig,
                     err_fd: c_int) -> IoResult<SpawnProcessResult> {
     use libc::funcs::posix88::unistd::{fork, dup2, close, chdir, execvp};
     use libc::funcs::bsd44::getdtablesize;
-    use libc::c_ulong;
+    use io::c;
 
     mod rustrt {
         extern {
@@ -475,16 +475,7 @@ fn spawn_process_os(config: p::ProcessConfig,
     }
 
     unsafe fn set_cloexec(fd: c_int) {
-        extern { fn ioctl(fd: c_int, req: c_ulong) -> c_int; }
-
-        #[cfg(target_os = "macos")]
-        #[cfg(target_os = "freebsd")]
-        static FIOCLEX: c_ulong = 0x20006601;
-        #[cfg(target_os = "linux")]
-        #[cfg(target_os = "android")]
-        static FIOCLEX: c_ulong = 0x5451;
-
-        let ret = ioctl(fd, FIOCLEX);
+        let ret = c::ioctl(fd, c::FIOCLEX);
         assert_eq!(ret, 0);
     }
 
diff --git a/src/libnative/io/timer_unix.rs b/src/libnative/io/timer_unix.rs
index 0a38a6ff0be..e5d4a6bb02b 100644
--- a/src/libnative/io/timer_unix.rs
+++ b/src/libnative/io/timer_unix.rs
@@ -53,8 +53,9 @@ use std::ptr;
 use std::rt::rtio;
 use std::sync::atomics;
 
-use io::file::FileDesc;
 use io::IoResult;
+use io::c;
+use io::file::FileDesc;
 use io::timer_helper;
 
 pub struct Timer {
@@ -84,16 +85,16 @@ pub enum Req {
 }
 
 // returns the current time (in milliseconds)
-fn now() -> u64 {
+pub fn now() -> u64 {
     unsafe {
         let mut now: libc::timeval = mem::init();
-        assert_eq!(imp::gettimeofday(&mut now, ptr::null()), 0);
+        assert_eq!(c::gettimeofday(&mut now, ptr::null()), 0);
         return (now.tv_sec as u64) * 1000 + (now.tv_usec as u64) / 1000;
     }
 }
 
 fn helper(input: libc::c_int, messages: Receiver<Req>) {
-    let mut set: imp::fd_set = unsafe { mem::init() };
+    let mut set: c::fd_set = unsafe { mem::init() };
 
     let mut fd = FileDesc::new(input, true);
     let mut timeout: libc::timeval = unsafe { mem::init() };
@@ -150,9 +151,9 @@ fn helper(input: libc::c_int, messages: Receiver<Req>) {
             &timeout as *libc::timeval
         };
 
-        imp::fd_set(&mut set, input);
+        c::fd_set(&mut set, input);
         match unsafe {
-            imp::select(input + 1, &set, ptr::null(), ptr::null(), timeout)
+            c::select(input + 1, &set, ptr::null(), ptr::null(), timeout)
         } {
             // timed out
             0 => signal(&mut active, &mut dead),
@@ -283,59 +284,3 @@ impl Drop for Timer {
         self.inner = Some(self.inner());
     }
 }
-
-#[cfg(target_os = "macos")]
-mod imp {
-    use libc;
-
-    pub static FD_SETSIZE: uint = 1024;
-
-    pub struct fd_set {
-        fds_bits: [i32, ..(FD_SETSIZE / 32)]
-    }
-
-    pub fn fd_set(set: &mut fd_set, fd: i32) {
-        set.fds_bits[(fd / 32) as uint] |= 1 << (fd % 32);
-    }
-
-    extern {
-        pub fn select(nfds: libc::c_int,
-                      readfds: *fd_set,
-                      writefds: *fd_set,
-                      errorfds: *fd_set,
-                      timeout: *libc::timeval) -> libc::c_int;
-
-        pub fn gettimeofday(timeval: *mut libc::timeval,
-                            tzp: *libc::c_void) -> libc::c_int;
-    }
-}
-
-#[cfg(target_os = "android")]
-#[cfg(target_os = "freebsd")]
-#[cfg(target_os = "linux")]
-mod imp {
-    use libc;
-    use std::uint;
-
-    pub static FD_SETSIZE: uint = 1024;
-
-    pub struct fd_set {
-        fds_bits: [uint, ..(FD_SETSIZE / uint::BITS)]
-    }
-
-    pub fn fd_set(set: &mut fd_set, fd: i32) {
-        let fd = fd as uint;
-        set.fds_bits[fd / uint::BITS] |= 1 << (fd % uint::BITS);
-    }
-
-    extern {
-        pub fn select(nfds: libc::c_int,
-                      readfds: *fd_set,
-                      writefds: *fd_set,
-                      errorfds: *fd_set,
-                      timeout: *libc::timeval) -> libc::c_int;
-
-        pub fn gettimeofday(timeval: *mut libc::timeval,
-                            tzp: *libc::c_void) -> libc::c_int;
-    }
-}