native: Implement clone/close_accept for unix

This commits implements {Tcp,Unix}Acceptor::{clone,close_accept} methods for
unix. A windows implementation is coming in a later commit.

The clone implementation is based on atomic reference counting (as with all
other clones), and the close_accept implementation is based on selecting on a
self-pipe which signals that a close has been seen.

5 files changed, 156 insertions, 46 deletions

diff --git a/src/libnative/io/net.rs b/src/libnative/io/net.rs
index 2255578ba80..7a8a363a0a3 100644
--- a/src/libnative/io/net.rs
+++ b/src/libnative/io/net.rs
@@ -14,10 +14,13 @@ use std::mem;
 use std::rt::mutex;
 use std::rt::rtio;
 use std::rt::rtio::{IoResult, IoError};
+use std::sync::atomics;
 
 use super::{retry, keep_going};
 use super::c;
 use super::util;
+use super::file::FileDesc;
+use super::process;
 
 ////////////////////////////////////////////////////////////////////////////////
 // sockaddr and misc bindings
@@ -479,9 +482,26 @@ impl TcpListener {
     pub fn fd(&self) -> sock_t { self.inner.fd }
 
     pub fn native_listen(self, backlog: int) -> IoResult<TcpAcceptor> {
+        try!(util::set_nonblocking(self.fd(), true));
         match unsafe { libc::listen(self.fd(), backlog as libc::c_int) } {
             -1 => Err(last_error()),
-            _ => Ok(TcpAcceptor { listener: self, deadline: 0 })
+
+            #[cfg(unix)]
+            _ => {
+                let (reader, writer) = try!(process::pipe());
+                try!(util::set_nonblocking(reader.fd(), true));
+                try!(util::set_nonblocking(writer.fd(), true));
+                try!(util::set_nonblocking(self.fd(), true));
+                Ok(TcpAcceptor {
+                    inner: Arc::new(AcceptorInner {
+                        listener: self,
+                        reader: reader,
+                        writer: writer,
+                        closed: atomics::AtomicBool::new(false),
+                    }),
+                    deadline: 0,
+                })
+            }
         }
     }
 }
@@ -502,31 +522,46 @@ impl rtio::RtioSocket for TcpListener {
 }
 
 pub struct TcpAcceptor {
-    listener: TcpListener,
+    inner: Arc<AcceptorInner>,
     deadline: u64,
 }
 
+#[cfg(unix)]
+struct AcceptorInner {
+    listener: TcpListener,
+    reader: FileDesc,
+    writer: FileDesc,
+    closed: atomics::AtomicBool,
+}
+
 impl TcpAcceptor {
-    pub fn fd(&self) -> sock_t { self.listener.fd() }
+    pub fn fd(&self) -> sock_t { self.inner.listener.fd() }
 
+    #[cfg(unix)]
     pub fn native_accept(&mut self) -> IoResult<TcpStream> {
-        if self.deadline != 0 {
-            try!(util::await(self.fd(), Some(self.deadline), util::Readable));
-        }
-        unsafe {
-            let mut storage: libc::sockaddr_storage = mem::zeroed();
-            let storagep = &mut storage as *mut libc::sockaddr_storage;
-            let size = mem::size_of::<libc::sockaddr_storage>();
-            let mut size = size as libc::socklen_t;
-            match retry(|| {
-                libc::accept(self.fd(),
-                             storagep as *mut libc::sockaddr,
-                             &mut size as *mut libc::socklen_t) as libc::c_int
-            }) as sock_t {
-                -1 => Err(last_error()),
-                fd => Ok(TcpStream::new(Inner::new(fd))),
+        let deadline = if self.deadline == 0 {None} else {Some(self.deadline)};
+
+        while !self.inner.closed.load(atomics::SeqCst) {
+            unsafe {
+                let mut storage: libc::sockaddr_storage = mem::zeroed();
+                let storagep = &mut storage as *mut libc::sockaddr_storage;
+                let size = mem::size_of::<libc::sockaddr_storage>();
+                let mut size = size as libc::socklen_t;
+                match retry(|| {
+                    libc::accept(self.fd(),
+                                 storagep as *mut libc::sockaddr,
+                                 &mut size as *mut libc::socklen_t) as libc::c_int
+                }) as sock_t {
+                    -1 if util::wouldblock() => {}
+                    -1 => return Err(last_error()),
+                    fd => return Ok(TcpStream::new(Inner::new(fd))),
+                }
             }
+            try!(util::await([self.fd(), self.inner.reader.fd()],
+                             deadline, util::Readable));
         }
+
+        Err(util::eof())
     }
 }
 
@@ -546,6 +581,24 @@ impl rtio::RtioTcpAcceptor for TcpAcceptor {
     fn set_timeout(&mut self, timeout: Option<u64>) {
         self.deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
     }
+
+    fn clone(&self) -> Box<rtio::RtioTcpAcceptor + Send> {
+        box TcpAcceptor {
+            inner: self.inner.clone(),
+            deadline: 0,
+        } as Box<rtio::RtioTcpAcceptor + Send>
+    }
+
+    #[cfg(unix)]
+    fn close_accept(&mut self) -> IoResult<()> {
+        self.inner.closed.store(true, atomics::SeqCst);
+        let mut fd = FileDesc::new(self.inner.writer.fd(), false);
+        match fd.inner_write([0]) {
+            Ok(..) => Ok(()),
+            Err(..) if util::wouldblock() => Ok(()),
+            Err(e) => Err(e),
+        }
+    }
 }
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -817,7 +870,7 @@ pub fn read<T>(fd: sock_t,
             // With a timeout, first we wait for the socket to become
             // readable using select(), specifying the relevant timeout for
             // our previously set deadline.
-            try!(util::await(fd, deadline, util::Readable));
+            try!(util::await([fd], deadline, util::Readable));
 
             // At this point, we're still within the timeout, and we've
             // determined that the socket is readable (as returned by
@@ -871,7 +924,7 @@ pub fn write<T>(fd: sock_t,
         while written < buf.len() && (write_everything || written == 0) {
             // As with read(), first wait for the socket to be ready for
             // the I/O operation.
-            match util::await(fd, deadline, util::Writable) {
+            match util::await([fd], deadline, util::Writable) {
                 Err(ref e) if e.code == libc::EOF as uint && written > 0 => {
                     assert!(deadline.is_some());
                     return Err(util::short_write(written, "short write"))
diff --git a/src/libnative/io/pipe_unix.rs b/src/libnative/io/pipe_unix.rs
index 895b8b5929c..4ad8383e6f8 100644
--- a/src/libnative/io/pipe_unix.rs
+++ b/src/libnative/io/pipe_unix.rs
@@ -15,12 +15,14 @@ use std::mem;
 use std::rt::mutex;
 use std::rt::rtio;
 use std::rt::rtio::{IoResult, IoError};
+use std::sync::atomics;
 
 use super::retry;
 use super::net;
 use super::util;
 use super::c;
-use super::file::fd_t;
+use super::process;
+use super::file::{fd_t, FileDesc};
 
 fn unix_socket(ty: libc::c_int) -> IoResult<fd_t> {
     match unsafe { libc::socket(libc::AF_UNIX, ty, 0) } {
@@ -225,7 +227,23 @@ impl UnixListener {
     pub fn native_listen(self, backlog: int) -> IoResult<UnixAcceptor> {
         match unsafe { libc::listen(self.fd(), backlog as libc::c_int) } {
             -1 => Err(super::last_error()),
-            _ => Ok(UnixAcceptor { listener: self, deadline: 0 })
+
+            #[cfg(unix)]
+            _ => {
+                let (reader, writer) = try!(process::pipe());
+                try!(util::set_nonblocking(reader.fd(), true));
+                try!(util::set_nonblocking(writer.fd(), true));
+                try!(util::set_nonblocking(self.fd(), true));
+                Ok(UnixAcceptor {
+                    inner: Arc::new(AcceptorInner {
+                        listener: self,
+                        reader: reader,
+                        writer: writer,
+                        closed: atomics::AtomicBool::new(false),
+                    }),
+                    deadline: 0,
+                })
+            }
         }
     }
 }
@@ -240,29 +258,45 @@ impl rtio::RtioUnixListener for UnixListener {
 }
 
 pub struct UnixAcceptor {
-    listener: UnixListener,
+    inner: Arc<AcceptorInner>,
     deadline: u64,
 }
 
+#[cfg(unix)]
+struct AcceptorInner {
+    listener: UnixListener,
+    reader: FileDesc,
+    writer: FileDesc,
+    closed: atomics::AtomicBool,
+}
+
 impl UnixAcceptor {
-    fn fd(&self) -> fd_t { self.listener.fd() }
+    fn fd(&self) -> fd_t { self.inner.listener.fd() }
 
     pub fn native_accept(&mut self) -> IoResult<UnixStream> {
-        if self.deadline != 0 {
-            try!(util::await(self.fd(), Some(self.deadline), util::Readable));
-        }
-        let mut storage: libc::sockaddr_storage = unsafe { mem::zeroed() };
-        let storagep = &mut storage as *mut libc::sockaddr_storage;
-        let size = mem::size_of::<libc::sockaddr_storage>();
-        let mut size = size as libc::socklen_t;
-        match retry(|| unsafe {
-            libc::accept(self.fd(),
-                         storagep as *mut libc::sockaddr,
-                         &mut size as *mut libc::socklen_t) as libc::c_int
-        }) {
-            -1 => Err(super::last_error()),
-            fd => Ok(UnixStream::new(Arc::new(Inner::new(fd))))
+        let deadline = if self.deadline == 0 {None} else {Some(self.deadline)};
+
+        while !self.inner.closed.load(atomics::SeqCst) {
+            unsafe {
+                let mut storage: libc::sockaddr_storage = mem::zeroed();
+                let storagep = &mut storage as *mut libc::sockaddr_storage;
+                let size = mem::size_of::<libc::sockaddr_storage>();
+                let mut size = size as libc::socklen_t;
+                match retry(|| {
+                    libc::accept(self.fd(),
+                                 storagep as *mut libc::sockaddr,
+                                 &mut size as *mut libc::socklen_t) as libc::c_int
+                }) {
+                    -1 if util::wouldblock() => {}
+                    -1 => return Err(super::last_error()),
+                    fd => return Ok(UnixStream::new(Arc::new(Inner::new(fd)))),
+                }
+            }
+            try!(util::await([self.fd(), self.inner.reader.fd()],
+                             deadline, util::Readable));
         }
+
+        Err(util::eof())
     }
 }
 
@@ -273,6 +307,24 @@ impl rtio::RtioUnixAcceptor for UnixAcceptor {
     fn set_timeout(&mut self, timeout: Option<u64>) {
         self.deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
     }
+
+    fn clone(&self) -> Box<rtio::RtioUnixAcceptor + Send> {
+        box UnixAcceptor {
+            inner: self.inner.clone(),
+            deadline: 0,
+        } as Box<rtio::RtioUnixAcceptor + Send>
+    }
+
+    #[cfg(unix)]
+    fn close_accept(&mut self) -> IoResult<()> {
+        self.inner.closed.store(true, atomics::SeqCst);
+        let mut fd = FileDesc::new(self.inner.writer.fd(), false);
+        match fd.inner_write([0]) {
+            Ok(..) => Ok(()),
+            Err(..) if util::wouldblock() => Ok(()),
+            Err(e) => Err(e),
+        }
+    }
 }
 
 impl Drop for UnixListener {
diff --git a/src/libnative/io/pipe_windows.rs b/src/libnative/io/pipe_windows.rs
index 717915e5d23..6ad51ee586f 100644
--- a/src/libnative/io/pipe_windows.rs
+++ b/src/libnative/io/pipe_windows.rs
@@ -99,10 +99,10 @@ use super::c;
 use super::util;
 use super::file::to_utf16;
 
-struct Event(libc::HANDLE);
+pub struct Event(libc::HANDLE);
 
 impl Event {
-    fn new(manual_reset: bool, initial_state: bool) -> IoResult<Event> {
+    pub fn new(manual_reset: bool, initial_state: bool) -> IoResult<Event> {
         let event = unsafe {
             libc::CreateEventW(ptr::mut_null(),
                                manual_reset as libc::BOOL,
@@ -116,7 +116,7 @@ impl Event {
         }
     }
 
-    fn handle(&self) -> libc::HANDLE { let Event(handle) = *self; handle }
+    pub fn handle(&self) -> libc::HANDLE { let Event(handle) = *self; handle }
 }
 
 impl Drop for Event {
diff --git a/src/libnative/io/process.rs b/src/libnative/io/process.rs
index d1b28854157..b8ec0cd5496 100644
--- a/src/libnative/io/process.rs
+++ b/src/libnative/io/process.rs
@@ -191,7 +191,7 @@ impl Drop for Process {
     }
 }
 
-fn pipe() -> IoResult<(file::FileDesc, file::FileDesc)> {
+pub fn pipe() -> IoResult<(file::FileDesc, file::FileDesc)> {
     #[cfg(unix)] use libc::EMFILE as ERROR;
     #[cfg(windows)] use libc::WSAEMFILE as ERROR;
     struct Closer { fd: libc::c_int }
diff --git a/src/libnative/io/util.rs b/src/libnative/io/util.rs
index 356805d91de..aec29bc2d03 100644
--- a/src/libnative/io/util.rs
+++ b/src/libnative/io/util.rs
@@ -9,6 +9,7 @@
 // except according to those terms.
 
 use libc;
+use std::cmp;
 use std::mem;
 use std::os;
 use std::ptr;
@@ -166,10 +167,15 @@ pub fn connect_timeout(fd: net::sock_t,
     }
 }
 
-pub fn await(fd: net::sock_t, deadline: Option<u64>,
+pub fn await(fds: &[net::sock_t], deadline: Option<u64>,
              status: SocketStatus) -> IoResult<()> {
     let mut set: c::fd_set = unsafe { mem::zeroed() };
-    c::fd_set(&mut set, fd);
+    let mut max = 0;
+    for &fd in fds.iter() {
+        c::fd_set(&mut set, fd);
+        max = cmp::max(max, fd + 1);
+    }
+
     let (read, write) = match status {
         Readable => (&mut set as *mut _, ptr::mut_null()),
         Writable => (ptr::mut_null(), &mut set as *mut _),
@@ -188,8 +194,7 @@ pub fn await(fd: net::sock_t, deadline: Option<u64>,
                 &mut tv as *mut _
             }
         };
-        let n = if cfg!(windows) {1} else {fd as libc::c_int + 1};
-        let r = unsafe { c::select(n, read, write, ptr::mut_null(), tvp) };
+        let r = unsafe { c::select(max, read, write, ptr::mut_null(), tvp) };
         r
     }) {
         -1 => Err(last_error()),