Implement named pipes for windows, touch up unix

* Implementation of pipe_win32 filled out for libnative
* Reorganize pipes to be clone-able
* Fix a few file descriptor leaks on error
* Factor out some common code into shared functions
* Make use of the if_ok!() macro for less indentation

Closes #11201

src/libnative/io/pipe_unix.rs

@@ -8,46 +8,124 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
+use std::c_str::CString;
+use std::cast;
+use std::io;
+use std::libc;
+use std::mem;
+use std::rt::rtio;
+use std::sync::arc::UnsafeArc;
+use std::unstable::intrinsics;
+
+use super::{IoResult, retry};
+use super::file::{keep_going, fd_t};
+
+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) -> IoResult<(libc::sockaddr_storage, uint)> {
+    // 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 mut storage: libc::sockaddr_storage = unsafe { intrinsics::init() };
+    let s: &mut libc::sockaddr_un = unsafe { cast::transmute(&mut storage) };
+
+    let len = addr.len();
+    if len > s.sun_path.len() - 1 {
+        return Err(io::IoError {
+            kind: io::InvalidInput,
+            desc: "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.mut_iter().zip(addr.iter()) {
+        *slot = value;
+    }
+
+    // count the null terminator
+    let len = mem::size_of::<libc::sa_family_t>() + len + 1;
+    return Ok((storage, len));
+}
+
+fn sockaddr_to_unix(storage: &libc::sockaddr_storage,
+                    len: uint) -> IoResult<CString> {
+    match storage.ss_family as libc::c_int {
+        libc::AF_UNIX => {
+            assert!(len as uint <= mem::size_of::<libc::sockaddr_un>());
+            let storage: &libc::sockaddr_un = unsafe {
+                cast::transmute(storage)
+            };
+            unsafe {
+                Ok(CString::new(storage.sun_path.as_ptr(), false).clone())
+            }
+        }
+        _ => Err(io::standard_error(io::InvalidInput))
+    }
+}
+
+struct Inner {
+    fd: fd_t,
+}
+
+impl Drop for Inner {
+    fn drop(&mut self) { unsafe { let _ = libc::close(self.fd); } }
+}
+
+fn connect(addr: &CString, ty: libc::c_int) -> IoResult<Inner> {
+    let (addr, len) = if_ok!(addr_to_sockaddr_un(addr));
+    let inner = Inner { fd: if_ok!(unix_socket(ty)) };
+    let addrp = &addr as *libc::sockaddr_storage;
+    match retry(|| unsafe {
+        libc::connect(inner.fd, addrp as *libc::sockaddr,
+                      len as libc::socklen_t)
+    }) {
+        -1 => Err(super::last_error()),
+        _  => Ok(inner)
+    }
+}
+
+fn bind(addr: &CString, ty: libc::c_int) -> IoResult<Inner> {
+    let (addr, len) = if_ok!(addr_to_sockaddr_un(addr));
+    let inner = Inner { fd: if_ok!(unix_socket(ty)) };
+    let addrp = &addr as *libc::sockaddr_storage;
+    match unsafe {
+        libc::bind(inner.fd, addrp as *libc::sockaddr, len as libc::socklen_t)
+    } {
+        -1 => Err(super::last_error()),
+        _  => Ok(inner)
+    }
+}
+
 ////////////////////////////////////////////////////////////////////////////////
-// Unix
+// Unix Streams
 ////////////////////////////////////////////////////////////////////////////////
 
 pub struct UnixStream {
-    priv fd: sock_t,
+    priv inner: UnsafeArc<Inner>,
 }
 
 impl UnixStream {
-    pub fn connect(addr: &CString, ty: libc::c_int) -> IoResult<UnixStream> {
-        unix_socket(ty).and_then(|fd| {
-            match addr_to_sockaddr_un(addr) {
-                Err(e)          => return Err(e),
-                Ok((addr, len)) => {
-                    let ret = UnixStream{ fd: fd };
-                    let addrp = &addr as *libc::sockaddr_storage;
-                    match retry(|| {
-                      libc::connect(fd, addrp as *libc::sockaddr,
-                                     len as libc::socklen_t)
-                    }) {
-                        -1 => return Err(super::last_error()),
-                        _  => return Ok(ret)
-                    }
-                }
-            }
+    pub fn connect(addr: &CString) -> IoResult<UnixStream> {
+        connect(addr, libc::SOCK_STREAM).map(|inner| {
+            UnixStream { inner: UnsafeArc::new(inner) }
         })
     }
 
-    pub fn fd(&self) -> sock_t { self.fd }
+    fn fd(&self) -> fd_t { unsafe { (*self.inner.get()).fd } }
 }
 
 impl rtio::RtioPipe for UnixStream {
     fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> {
-        let ret = retry(|| {
-            unsafe {
-                libc::recv(self.fd,
-                           buf.as_ptr() as *mut libc::c_void,
-                           buf.len() as wrlen,
-                           0) as libc::c_int
-            }
+        let ret = retry(|| unsafe {
+            libc::recv(self.fd(),
+                       buf.as_ptr() as *mut libc::c_void,
+                       buf.len() as libc::size_t,
+                       0) as libc::c_int
         });
         if ret == 0 {
             Err(io::standard_error(io::EndOfFile))
@@ -57,14 +135,13 @@ impl rtio::RtioPipe for UnixStream {
             Ok(ret as uint)
         }
     }
+
     fn write(&mut self, buf: &[u8]) -> IoResult<()> {
-        let ret = keep_going(buf, |buf, len| {
-            unsafe {
-                libc::send(self.fd,
-                           buf as *mut libc::c_void,
-                           len as wrlen,
-                           0) as i64
-            }
+        let ret = keep_going(buf, |buf, len| unsafe {
+            libc::send(self.fd(),
+                       buf as *mut libc::c_void,
+                       len as libc::size_t,
+                       0) as i64
         });
         if ret < 0 {
             Err(super::last_error())
@@ -72,10 +149,10 @@ impl rtio::RtioPipe for UnixStream {
             Ok(())
         }
     }
-}
 
-impl Drop for UnixStream {
-    fn drop(&mut self) { unsafe { close(self.fd); } }
+    fn clone(&self) -> ~rtio::RtioPipe {
+        ~UnixStream { inner: self.inner.clone() } as ~rtio::RtioPipe
+    }
 }
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -83,176 +160,89 @@ impl Drop for UnixStream {
 ////////////////////////////////////////////////////////////////////////////////
 
 pub struct UnixDatagram {
-    priv fd: sock_t,
+    priv inner: UnsafeArc<Inner>,
 }
 
 impl UnixDatagram {
-    pub fn connect(addr: &CString, ty: libc::c_int) -> IoResult<UnixDatagram> {
-        unsafe {
-            unix_socket(ty).and_then(|fd| {
-                match addr_to_sockaddr_un(addr) {
-                    Err(e)          => return Err(e),
-                    Ok((addr, len)) => {
-                        let ret = UnixDatagram{ fd: fd };
-                        let addrp = &addr as *libc::sockaddr_storage;
-                        match retry(|| {
-                          libc::connect(fd, addrp as *libc::sockaddr,
-                                         len as libc::socklen_t)
-                        }) {
-                            -1 => return Err(super::last_error()),
-                            _  => return Ok(ret)
-                        }
-                    }
-                }
-            })
-        }
+    pub fn connect(addr: &CString) -> IoResult<UnixDatagram> {
+        connect(addr, libc::SOCK_DGRAM).map(|inner| {
+            UnixDatagram { inner: UnsafeArc::new(inner) }
+        })
     }
 
     pub fn bind(addr: &CString) -> IoResult<UnixDatagram> {
-        unsafe {
-            unix_socket(libc::SOCK_DGRAM).and_then(|fd| {
-                match addr_to_sockaddr_un(addr) {
-                    Err(e)          => return Err(e),
-                    Ok((addr, len)) => {
-                        let ret = UnixDatagram{ fd: fd };
-                        let addrp = &addr as *libc::sockaddr_storage;
-                        match libc::bind(fd, addrp as *libc::sockaddr,
-                                         len as libc::socklen_t) {
-                            -1 => return Err(super::last_error()),
-                            _  => return Ok(ret)
-                        }
-                    }
-                }
-            })
-        }
+        bind(addr, libc::SOCK_DGRAM).map(|inner| {
+            UnixDatagram { inner: UnsafeArc::new(inner) }
+        })
     }
 
-    pub fn fd(&self) -> sock_t { self.fd }
-}
+    fn fd(&self) -> fd_t { unsafe { (*self.inner.get()).fd } }
 
-impl rtio::RtioPipe for UnixDatagram {
-    fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> {
-        let ret = retry(|| {
-            unsafe {
-                libc::recv(self.fd,
+    pub fn recvfrom(&mut self, buf: &mut [u8]) -> IoResult<(uint, CString)> {
+        let mut storage: libc::sockaddr_storage = unsafe { intrinsics::init() };
+        let storagep = &mut storage as *mut libc::sockaddr_storage;
+        let mut addrlen: libc::socklen_t =
+                mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
+        let ret = retry(|| unsafe {
+            libc::recvfrom(self.fd(),
                            buf.as_ptr() as *mut libc::c_void,
-                           buf.len() as wrlen,
-                           0) as libc::c_int
-            }
+                           buf.len() as libc::size_t,
+                           0,
+                           storagep as *mut libc::sockaddr,
+                           &mut addrlen) as libc::c_int
         });
-        if ret == 0 {
-            Err(io::standard_error(io::EndOfFile))
-        } else if ret < 0 {
-            Err(super::last_error())
-        } else {
-            Ok(ret as uint)
-        }
+        if ret < 0 { return Err(super::last_error()) }
+        sockaddr_to_unix(&storage, addrlen as uint).and_then(|addr| {
+            Ok((ret as uint, addr))
+        })
     }
-    fn write(&mut self, buf: &[u8]) -> IoResult<()> {
-        let ret = keep_going(buf, |buf, len| {
-            unsafe {
-                libc::send(self.fd,
-                           buf as *mut libc::c_void,
-                           len as wrlen,
-                           0) as i64
-            }
+
+    pub fn sendto(&mut self, buf: &[u8], dst: &CString) -> IoResult<()> {
+        let (dst, len) = if_ok!(addr_to_sockaddr_un(dst));
+        let dstp = &dst as *libc::sockaddr_storage;
+        let ret = retry(|| unsafe {
+            libc::sendto(self.fd(),
+                         buf.as_ptr() as *libc::c_void,
+                         buf.len() as libc::size_t,
+                         0,
+                         dstp as *libc::sockaddr,
+                         len as libc::socklen_t) as libc::c_int
         });
-        if ret < 0 {
-            Err(super::last_error())
-        } else {
-            Ok(())
-        }
-    }
-}
-
-impl rtio::RtioDatagramPipe for UnixDatagram {
-    fn recvfrom(&mut self, buf: &mut [u8]) -> IoResult<(uint, CString)> {
-        unsafe {
-            let mut storage: libc::sockaddr_storage = intrinsics::init();
-            let storagep = &mut storage as *mut libc::sockaddr_storage;
-            let mut addrlen: libc::socklen_t =
-                    mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
-            let ret = retry(|| {
-                libc::recvfrom(self.fd,
-                               buf.as_ptr() as *mut libc::c_void,
-                               buf.len() as msglen_t,
-                               0,
-                               storagep as *mut libc::sockaddr,
-                               &mut addrlen) as libc::c_int
-            });
-            if ret < 0 { return Err(super::last_error()) }
-            sockaddr_to_unix(&storage, addrlen as uint).and_then(|addr| {
-                Ok((ret as uint, addr))
-            })
-        }
-    }
-
-    fn sendto(&mut self, buf: &[u8], dst: &CString) -> IoResult<()> {
-        match addr_to_sockaddr_un(dst) {
-            Err(e)         => Err(e),
-            Ok((dst, len)) => {
-                let dstp = &dst as *libc::sockaddr_storage;
-                unsafe {
-                    let ret = retry(|| {
-                        libc::sendto(self.fd,
-                                     buf.as_ptr() as *libc::c_void,
-                                     buf.len() as msglen_t,
-                                     0,
-                                     dstp as *libc::sockaddr,
-                                     len as libc::socklen_t) as libc::c_int
-                    });
-                    match ret {
-                        -1 => Err(super::last_error()),
-                        n if n as uint != buf.len() => {
-                            Err(io::IoError {
-                                kind: io::OtherIoError,
-                                desc: "couldn't send entire packet at once",
-                                detail: None,
-                            })
-                        }
-                        _ => Ok(())
-                    }
-                }
+        match ret {
+            -1 => Err(super::last_error()),
+            n if n as uint != buf.len() => {
+                Err(io::IoError {
+                    kind: io::OtherIoError,
+                    desc: "couldn't send entire packet at once",
+                    detail: None,
+                })
             }
+            _ => Ok(())
         }
     }
+
+    pub fn clone(&mut self) -> UnixDatagram {
+        UnixDatagram { inner: self.inner.clone() }
+    }
 }
 
-impl Drop for UnixDatagram {
-    fn drop(&mut self) { unsafe { close(self.fd); } }
-}
 ////////////////////////////////////////////////////////////////////////////////
 // Unix Listener
 ////////////////////////////////////////////////////////////////////////////////
 
 pub struct UnixListener {
-    priv fd: sock_t,
+    priv inner: Inner,
 }
 
 impl UnixListener {
     pub fn bind(addr: &CString) -> IoResult<UnixListener> {
-        unsafe {
-            unix_socket(libc::SOCK_STREAM).and_then(|fd| {
-                match addr_to_sockaddr_un(addr) {
-                    Err(e)          => return Err(e),
-                    Ok((addr, len)) => {
-                        let ret = UnixListener{ fd: fd };
-                        let addrp = &addr as *libc::sockaddr_storage;
-                        match libc::bind(fd, addrp as *libc::sockaddr,
-                                         len as libc::socklen_t) {
-                            -1 => return Err(super::last_error()),
-                            _  => return Ok(ret)
-                        }
-                    }
-                }
-            })
-        }
+        bind(addr, libc::SOCK_STREAM).map(|fd| UnixListener { inner: fd })
     }
 
-    pub fn fd(&self) -> sock_t { self.fd }
+    fn fd(&self) -> fd_t { self.inner.fd }
 
     pub fn native_listen(self, backlog: int) -> IoResult<UnixAcceptor> {
-        match unsafe { libc::listen(self.fd, backlog as libc::c_int) } {
+        match unsafe { libc::listen(self.fd(), backlog as libc::c_int) } {
             -1 => Err(super::last_error()),
             _ => Ok(UnixAcceptor { listener: self })
         }
@@ -265,16 +255,12 @@ impl rtio::RtioUnixListener for UnixListener {
     }
 }
 
-impl Drop for UnixListener {
-    fn drop(&mut self) { unsafe { close(self.fd); } }
-}
-
 pub struct UnixAcceptor {
     priv listener: UnixListener,
 }
 
 impl UnixAcceptor {
-    pub fn fd(&self) -> sock_t { self.listener.fd }
+    fn fd(&self) -> fd_t { self.listener.fd() }
 
     pub fn native_accept(&mut self) -> IoResult<UnixStream> {
         let mut storage: libc::sockaddr_storage = unsafe { intrinsics::init() };
@@ -285,9 +271,9 @@ impl UnixAcceptor {
             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(super::last_error()),
-            fd => Ok(UnixStream { fd: fd })
+            fd => Ok(UnixStream { inner: UnsafeArc::new(Inner { fd: fd }) })
         }
     }
 }
@@ -297,4 +283,3 @@ impl rtio::RtioUnixAcceptor for UnixAcceptor {
         self.native_accept().map(|s| ~s as ~rtio::RtioPipe)
     }
 }
-