Runtime removal: refactor pipes and networking

This patch continues the runtime removal by moving pipe and
networking-related code into `sys`.

Because this eliminates APIs in `libnative` and `librustrt`, it is a:

[breaking-change]

This functionality is likely to be available publicly, in some form,
from `std` in the future.

5 files changed, 543 insertions, 7 deletions

diff --git a/src/libstd/sys/unix/mod.rs b/src/libstd/sys/unix/mod.rs
index ad5de2dad48..5a43fd08f90 100644
--- a/src/libstd/sys/unix/mod.rs
+++ b/src/libstd/sys/unix/mod.rs
@@ -8,24 +8,51 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
+#![allow(missing_doc)]
+
 extern crate libc;
 
 use num;
 use prelude::*;
 use io::{mod, IoResult, IoError};
+use sys_common::mkerr_libc;
 
+pub mod c;
 pub mod fs;
 pub mod os;
-pub mod c;
+pub mod tcp;
+pub mod udp;
+pub mod pipe;
+
+pub mod addrinfo {
+    pub use sys_common::net::get_host_addresses;
+}
 
-pub type sock_t = io::file::fd_t;
+// FIXME: move these to c module
+pub type sock_t = self::fs::fd_t;
 pub type wrlen = libc::size_t;
+pub type msglen_t = libc::size_t;
 pub unsafe fn close_sock(sock: sock_t) { let _ = libc::close(sock); }
 
 pub fn last_error() -> IoError {
-    let errno = os::errno() as i32;
-    let mut err = decode_error(errno);
-    err.detail = Some(os::error_string(errno));
+    decode_error_detailed(os::errno() as i32)
+}
+
+pub fn last_net_error() -> IoError {
+    last_error()
+}
+
+extern "system" {
+    fn gai_strerror(errcode: libc::c_int) -> *const libc::c_char;
+}
+
+pub fn last_gai_error(s: libc::c_int) -> IoError {
+    use c_str::CString;
+
+    let mut err = decode_error(s);
+    err.detail = Some(unsafe {
+        CString::new(gai_strerror(s), false).as_str().unwrap().to_string()
+    });
     err
 }
 
@@ -64,6 +91,12 @@ pub fn decode_error(errno: i32) -> IoError {
     IoError { kind: kind, desc: desc, detail: None }
 }
 
+pub fn decode_error_detailed(errno: i32) -> IoError {
+    let mut err = decode_error(errno);
+    err.detail = Some(os::error_string(errno));
+    err
+}
+
 #[inline]
 pub fn retry<I: PartialEq + num::One + Neg<I>> (f: || -> I) -> I {
     let minus_one = -num::one::<I>();
@@ -86,7 +119,10 @@ pub fn wouldblock() -> bool {
     err == libc::EWOULDBLOCK as int || err == libc::EAGAIN as int
 }
 
-pub fn set_nonblocking(fd: net::sock_t, nb: bool) -> IoResult<()> {
+pub 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) }))
+    mkerr_libc(retry(|| unsafe { c::ioctl(fd, c::FIONBIO, &set) }))
 }
+
+// nothing needed on unix platforms
+pub fn init_net() {}
diff --git a/src/libstd/sys/unix/os.rs b/src/libstd/sys/unix/os.rs
index 34699eb27c1..4e495f043bc 100644
--- a/src/libstd/sys/unix/os.rs
+++ b/src/libstd/sys/unix/os.rs
@@ -11,6 +11,8 @@
 use libc;
 use libc::{c_int, c_char};
 use prelude::*;
+use io::IoResult;
+use sys::fs::FileDesc;
 
 use os::TMPBUF_SZ;
 
@@ -99,3 +101,12 @@ pub fn error_string(errno: i32) -> String {
         ::string::raw::from_buf(p as *const u8)
     }
 }
+
+pub unsafe fn pipe() -> IoResult<(FileDesc, FileDesc)> {
+    let mut fds = [0, ..2];
+    if libc::pipe(fds.as_mut_ptr()) == 0 {
+        Ok((FileDesc::new(fds[0], true), FileDesc::new(fds[1], true)))
+    } else {
+        Err(super::last_error())
+    }
+}
diff --git a/src/libstd/sys/unix/pipe.rs b/src/libstd/sys/unix/pipe.rs
new file mode 100644
index 00000000000..67384848a94
--- /dev/null
+++ b/src/libstd/sys/unix/pipe.rs
@@ -0,0 +1,321 @@
+// 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.
+
+use alloc::arc::Arc;
+use libc;
+use c_str::CString;
+use mem;
+use rt::mutex;
+use sync::atomic;
+use io::{mod, IoResult, IoError};
+use prelude::*;
+
+use sys::{mod, timer, retry, c, set_nonblocking, wouldblock};
+use sys::fs::{fd_t, FileDesc};
+use sys_common::net::*;
+use sys_common::{eof, mkerr_libc};
+
+fn unix_socket(ty: libc::c_int) -> IoResult<fd_t> {
+    match unsafe { libc::socket(libc::AF_UNIX, ty, 0) } {
+        -1 => Err(super::last_error()),
+        fd => Ok(fd)
+    }
+}
+
+fn addr_to_sockaddr_un(addr: &CString,
+                       storage: &mut libc::sockaddr_storage)
+                       -> IoResult<libc::socklen_t> {
+    // the sun_path length is limited to SUN_LEN (with null)
+    assert!(mem::size_of::<libc::sockaddr_storage>() >=
+            mem::size_of::<libc::sockaddr_un>());
+    let s = unsafe { &mut *(storage as *mut _ as *mut libc::sockaddr_un) };
+
+    let len = addr.len();
+    if len > s.sun_path.len() - 1 {
+        return Err(IoError {
+            kind: io::InvalidInput,
+            desc: "invalid argument: path must be smaller than SUN_LEN",
+            detail: None,
+        })
+    }
+    s.sun_family = libc::AF_UNIX as libc::sa_family_t;
+    for (slot, value) in s.sun_path.iter_mut().zip(addr.iter()) {
+        *slot = value;
+    }
+
+    // count the null terminator
+    let len = mem::size_of::<libc::sa_family_t>() + len + 1;
+    return Ok(len as libc::socklen_t);
+}
+
+struct Inner {
+    fd: fd_t,
+
+    // Unused on Linux, where this lock is not necessary.
+    #[allow(dead_code)]
+    lock: mutex::NativeMutex
+}
+
+impl Inner {
+    fn new(fd: fd_t) -> Inner {
+        Inner { fd: fd, lock: unsafe { mutex::NativeMutex::new() } }
+    }
+}
+
+impl Drop for Inner {
+    fn drop(&mut self) { unsafe { let _ = libc::close(self.fd); } }
+}
+
+fn connect(addr: &CString, ty: libc::c_int,
+           timeout: Option<u64>) -> IoResult<Inner> {
+    let mut storage = unsafe { mem::zeroed() };
+    let len = try!(addr_to_sockaddr_un(addr, &mut storage));
+    let inner = Inner::new(try!(unix_socket(ty)));
+    let addrp = &storage as *const _ as *const libc::sockaddr;
+
+    match timeout {
+        None => {
+            match retry(|| unsafe { libc::connect(inner.fd, addrp, len) }) {
+                -1 => Err(super::last_error()),
+                _  => Ok(inner)
+            }
+        }
+        Some(timeout_ms) => {
+            try!(connect_timeout(inner.fd, addrp, len, timeout_ms));
+            Ok(inner)
+        }
+    }
+}
+
+fn bind(addr: &CString, ty: libc::c_int) -> IoResult<Inner> {
+    let mut storage = unsafe { mem::zeroed() };
+    let len = try!(addr_to_sockaddr_un(addr, &mut storage));
+    let inner = Inner::new(try!(unix_socket(ty)));
+    let addrp = &storage as *const _ as *const libc::sockaddr;
+    match unsafe {
+        libc::bind(inner.fd, addrp, len)
+    } {
+        -1 => Err(super::last_error()),
+        _  => Ok(inner)
+    }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+// Unix Streams
+////////////////////////////////////////////////////////////////////////////////
+
+pub struct UnixStream {
+    inner: Arc<Inner>,
+    read_deadline: u64,
+    write_deadline: u64,
+}
+
+impl UnixStream {
+    pub fn connect(addr: &CString,
+                   timeout: Option<u64>) -> IoResult<UnixStream> {
+        connect(addr, libc::SOCK_STREAM, timeout).map(|inner| {
+            UnixStream::new(Arc::new(inner))
+        })
+    }
+
+    fn new(inner: Arc<Inner>) -> UnixStream {
+        UnixStream {
+            inner: inner,
+            read_deadline: 0,
+            write_deadline: 0,
+        }
+    }
+
+    fn fd(&self) -> fd_t { self.inner.fd }
+
+    #[cfg(target_os = "linux")]
+    fn lock_nonblocking(&self) {}
+
+    #[cfg(not(target_os = "linux"))]
+    fn lock_nonblocking<'a>(&'a self) -> Guard<'a> {
+        let ret = Guard {
+            fd: self.fd(),
+            guard: unsafe { self.inner.lock.lock() },
+        };
+        assert!(set_nonblocking(self.fd(), true).is_ok());
+        ret
+    }
+
+    pub fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> {
+        let fd = self.fd();
+        let dolock = || self.lock_nonblocking();
+        let doread = |nb| unsafe {
+            let flags = if nb {c::MSG_DONTWAIT} else {0};
+            libc::recv(fd,
+                       buf.as_mut_ptr() as *mut libc::c_void,
+                       buf.len() as libc::size_t,
+                       flags) as libc::c_int
+        };
+        read(fd, self.read_deadline, dolock, doread)
+    }
+
+    pub fn write(&mut self, buf: &[u8]) -> IoResult<()> {
+        let fd = self.fd();
+        let dolock = || self.lock_nonblocking();
+        let dowrite = |nb: bool, buf: *const u8, len: uint| unsafe {
+            let flags = if nb {c::MSG_DONTWAIT} else {0};
+            libc::send(fd,
+                       buf as *const _,
+                       len as libc::size_t,
+                       flags) as i64
+        };
+        match write(fd, self.write_deadline, buf, true, dolock, dowrite) {
+            Ok(_) => Ok(()),
+            Err(e) => Err(e)
+        }
+    }
+
+    pub fn close_write(&mut self) -> IoResult<()> {
+        mkerr_libc(unsafe { libc::shutdown(self.fd(), libc::SHUT_WR) })
+    }
+
+    pub fn close_read(&mut self) -> IoResult<()> {
+        mkerr_libc(unsafe { libc::shutdown(self.fd(), libc::SHUT_RD) })
+    }
+
+    pub fn set_timeout(&mut self, timeout: Option<u64>) {
+        let deadline = timeout.map(|a| timer::now() + a).unwrap_or(0);
+        self.read_deadline = deadline;
+        self.write_deadline = deadline;
+    }
+
+    pub fn set_read_timeout(&mut self, timeout: Option<u64>) {
+        self.read_deadline = timeout.map(|a| timer::now() + a).unwrap_or(0);
+    }
+
+    pub fn set_write_timeout(&mut self, timeout: Option<u64>) {
+        self.write_deadline = timeout.map(|a| timer::now() + a).unwrap_or(0);
+    }
+}
+
+impl Clone for UnixStream {
+    fn clone(&self) -> UnixStream {
+        UnixStream::new(self.inner.clone())
+    }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+// Unix Listener
+////////////////////////////////////////////////////////////////////////////////
+
+pub struct UnixListener {
+    inner: Inner,
+    path: CString,
+}
+
+impl UnixListener {
+    pub fn bind(addr: &CString) -> IoResult<UnixListener> {
+        bind(addr, libc::SOCK_STREAM).map(|fd| {
+            UnixListener { inner: fd, path: addr.clone() }
+        })
+    }
+
+    fn fd(&self) -> fd_t { self.inner.fd }
+
+    pub fn listen(self) -> IoResult<UnixAcceptor> {
+        match unsafe { libc::listen(self.fd(), 128) } {
+            -1 => Err(super::last_error()),
+
+            _ => {
+                let (reader, writer) = try!(unsafe { sys::os::pipe() });
+                try!(set_nonblocking(reader.fd(), true));
+                try!(set_nonblocking(writer.fd(), true));
+                try!(set_nonblocking(self.fd(), true));
+                Ok(UnixAcceptor {
+                    inner: Arc::new(AcceptorInner {
+                        listener: self,
+                        reader: reader,
+                        writer: writer,
+                        closed: atomic::AtomicBool::new(false),
+                    }),
+                    deadline: 0,
+                })
+            }
+        }
+    }
+}
+
+pub struct UnixAcceptor {
+    inner: Arc<AcceptorInner>,
+    deadline: u64,
+}
+
+struct AcceptorInner {
+    listener: UnixListener,
+    reader: FileDesc,
+    writer: FileDesc,
+    closed: atomic::AtomicBool,
+}
+
+impl UnixAcceptor {
+    fn fd(&self) -> fd_t { self.inner.listener.fd() }
+
+    pub fn accept(&mut self) -> IoResult<UnixStream> {
+        let deadline = if self.deadline == 0 {None} else {Some(self.deadline)};
+
+        while !self.inner.closed.load(atomic::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 wouldblock() => {}
+                    -1 => return Err(super::last_error()),
+                    fd => return Ok(UnixStream::new(Arc::new(Inner::new(fd)))),
+                }
+            }
+            try!(await([self.fd(), self.inner.reader.fd()],
+                             deadline, Readable));
+        }
+
+        Err(eof())
+    }
+
+    pub fn set_timeout(&mut self, timeout: Option<u64>) {
+        self.deadline = timeout.map(|a| timer::now() + a).unwrap_or(0);
+    }
+
+    pub fn close_accept(&mut self) -> IoResult<()> {
+        self.inner.closed.store(true, atomic::SeqCst);
+        let fd = FileDesc::new(self.inner.writer.fd(), false);
+        match fd.write([0]) {
+            Ok(..) => Ok(()),
+            Err(..) if wouldblock() => Ok(()),
+            Err(e) => Err(e),
+        }
+    }
+}
+
+impl Clone for UnixAcceptor {
+    fn clone(&self) -> UnixAcceptor {
+        UnixAcceptor { inner: self.inner.clone(), deadline: 0 }
+    }
+}
+
+impl Drop for UnixListener {
+    fn drop(&mut self) {
+        // Unlink the path to the socket to ensure that it doesn't linger. We're
+        // careful to unlink the path before we close the file descriptor to
+        // prevent races where we unlink someone else's path.
+        unsafe {
+            let _ = libc::unlink(self.path.as_ptr());
+        }
+    }
+}
diff --git a/src/libstd/sys/unix/tcp.rs b/src/libstd/sys/unix/tcp.rs
new file mode 100644
index 00000000000..962475e4177
--- /dev/null
+++ b/src/libstd/sys/unix/tcp.rs
@@ -0,0 +1,157 @@
+// 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.
+
+use io::net::ip;
+use io::IoResult;
+use libc;
+use mem;
+use ptr;
+use prelude::*;
+use super::{last_error, last_net_error, retry, sock_t};
+use sync::{Arc, atomic};
+use sys::fs::FileDesc;
+use sys::{set_nonblocking, wouldblock};
+use sys;
+use sys_common;
+use sys_common::net::*;
+
+pub use sys_common::net::TcpStream;
+
+////////////////////////////////////////////////////////////////////////////////
+// TCP listeners
+////////////////////////////////////////////////////////////////////////////////
+
+pub struct TcpListener {
+    pub inner: FileDesc,
+}
+
+impl TcpListener {
+    pub fn bind(addr: ip::SocketAddr) -> IoResult<TcpListener> {
+        let fd = try!(socket(addr, libc::SOCK_STREAM));
+        let ret = TcpListener { inner: FileDesc::new(fd, true) };
+
+        let mut storage = unsafe { mem::zeroed() };
+        let len = addr_to_sockaddr(addr, &mut storage);
+        let addrp = &storage as *const _ as *const libc::sockaddr;
+
+        // On platforms with Berkeley-derived sockets, this allows
+        // to quickly rebind a socket, without needing to wait for
+        // the OS to clean up the previous one.
+        try!(setsockopt(fd, libc::SOL_SOCKET, libc::SO_REUSEADDR, 1 as libc::c_int));
+
+
+        match unsafe { libc::bind(fd, addrp, len) } {
+            -1 => Err(last_error()),
+            _ => Ok(ret),
+        }
+    }
+
+    pub fn fd(&self) -> sock_t { self.inner.fd() }
+
+    pub fn listen(self, backlog: int) -> IoResult<TcpAcceptor> {
+        match unsafe { libc::listen(self.fd(), backlog as libc::c_int) } {
+            -1 => Err(last_net_error()),
+            _ => {
+                let (reader, writer) = try!(unsafe { sys::os::pipe() });
+                try!(set_nonblocking(reader.fd(), true));
+                try!(set_nonblocking(writer.fd(), true));
+                try!(set_nonblocking(self.fd(), true));
+                Ok(TcpAcceptor {
+                    inner: Arc::new(AcceptorInner {
+                        listener: self,
+                        reader: reader,
+                        writer: writer,
+                        closed: atomic::AtomicBool::new(false),
+                    }),
+                    deadline: 0,
+                })
+            }
+        }
+    }
+
+    pub fn socket_name(&mut self) -> IoResult<ip::SocketAddr> {
+        sockname(self.fd(), libc::getsockname)
+    }
+}
+
+pub struct TcpAcceptor {
+    inner: Arc<AcceptorInner>,
+    deadline: u64,
+}
+
+struct AcceptorInner {
+    listener: TcpListener,
+    reader: FileDesc,
+    writer: FileDesc,
+    closed: atomic::AtomicBool,
+}
+
+impl TcpAcceptor {
+    pub fn fd(&self) -> sock_t { self.inner.listener.fd() }
+
+    pub fn accept(&mut self) -> IoResult<TcpStream> {
+        // In implementing accept, the two main concerns are dealing with
+        // close_accept() and timeouts. The unix implementation is based on a
+        // nonblocking accept plus a call to select(). Windows ends up having
+        // an entirely separate implementation than unix, which is explained
+        // below.
+        //
+        // To implement timeouts, all blocking is done via select() instead of
+        // accept() by putting the socket in non-blocking mode. Because
+        // select() takes a timeout argument, we just pass through the timeout
+        // to select().
+        //
+        // To implement close_accept(), we have a self-pipe to ourselves which
+        // is passed to select() along with the socket being accepted on. The
+        // self-pipe is never written to unless close_accept() is called.
+        let deadline = if self.deadline == 0 {None} else {Some(self.deadline)};
+
+        while !self.inner.closed.load(atomic::SeqCst) {
+            match retry(|| unsafe {
+                libc::accept(self.fd(), ptr::null_mut(), ptr::null_mut())
+            }) {
+                -1 if wouldblock() => {}
+                -1 => return Err(last_net_error()),
+                fd => return Ok(TcpStream::new(fd as sock_t)),
+            }
+            try!(await([self.fd(), self.inner.reader.fd()],
+                             deadline, Readable));
+        }
+
+        Err(sys_common::eof())
+    }
+
+    pub fn socket_name(&mut self) -> IoResult<ip::SocketAddr> {
+        sockname(self.fd(), libc::getsockname)
+    }
+
+    pub fn set_timeout(&mut self, timeout: Option<u64>) {
+        self.deadline = timeout.map(|a| sys::timer::now() + a).unwrap_or(0);
+    }
+
+    pub fn close_accept(&mut self) -> IoResult<()> {
+        self.inner.closed.store(true, atomic::SeqCst);
+        let fd = FileDesc::new(self.inner.writer.fd(), false);
+        match fd.write([0]) {
+            Ok(..) => Ok(()),
+            Err(..) if wouldblock() => Ok(()),
+            Err(e) => Err(e),
+        }
+    }
+}
+
+impl Clone for TcpAcceptor {
+    fn clone(&self) -> TcpAcceptor {
+        TcpAcceptor {
+            inner: self.inner.clone(),
+            deadline: 0,
+        }
+    }
+}
diff --git a/src/libstd/sys/unix/udp.rs b/src/libstd/sys/unix/udp.rs
new file mode 100644
index 00000000000..50f8fb828ad
--- /dev/null
+++ b/src/libstd/sys/unix/udp.rs
@@ -0,0 +1,11 @@
+// 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.
+
+pub use sys_common::net::UdpSocket;