src/libnative/io/pipe_unix.rs

   1 // Copyright 2014 The Rust Project Developers. See the COPYRIGHT
   2 // file at the top-level directory of this distribution and at
   3 // http://rust-lang.org/COPYRIGHT.
   4 //
   5 // Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
   6 // http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
   7 // <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
   8 // option. This file may not be copied, modified, or distributed
   9 // except according to those terms.
  10
  11 use alloc::arc::Arc;
  12 use libc;
  13 use std::c_str::CString;
  14 use std::mem;
  15 use std::rt::mutex;
  16 use std::rt::rtio;
  17 use std::rt::rtio::{IoResult, IoError};
  18
  19 use super::retry;
  20 use super::net;
  21 use super::util;
  22 use super::c;
  23 use super::file::fd_t;
  24
  25 fn unix_socket(ty: libc::c_int) -> IoResult<fd_t> {
  26     match unsafe { libc::socket(libc::AF_UNIX, ty, 0) } {
  27         -1 => Err(super::last_error()),
  28         fd => Ok(fd)
  29     }
  30 }
  31
  32 fn addr_to_sockaddr_un(addr: &CString,
  33                        storage: &mut libc::sockaddr_storage)
  34                        -> IoResult<libc::socklen_t> {
  35     // the sun_path length is limited to SUN_LEN (with null)
  36     assert!(mem::size_of::<libc::sockaddr_storage>() >=
  37             mem::size_of::<libc::sockaddr_un>());
  38     let s = unsafe { &mut *(storage as *mut _ as *mut libc::sockaddr_un) };
  39
  40     let len = addr.len();
  41     if len > s.sun_path.len() - 1 {
  42         #[cfg(unix)] use ERROR = libc::EINVAL;
  43         #[cfg(windows)] use ERROR = libc::WSAEINVAL;
  44         return Err(IoError {
  45             code: ERROR as uint,
  46             extra: 0,
  47             detail: Some("path must be smaller than SUN_LEN".to_string()),
  48         })
  49     }
  50     s.sun_family = libc::AF_UNIX as libc::sa_family_t;
  51     for (slot, value) in s.sun_path.mut_iter().zip(addr.iter()) {
  52         *slot = value;
  53     }
  54
  55     // count the null terminator
  56     let len = mem::size_of::<libc::sa_family_t>() + len + 1;
  57     return Ok(len as libc::socklen_t);
  58 }
  59
  60 struct Inner {
  61     fd: fd_t,
  62
  63     // Unused on Linux, where this lock is not necessary.
  64     #[allow(dead_code)]
  65     lock: mutex::NativeMutex
  66 }
  67
  68 impl Inner {
  69     fn new(fd: fd_t) -> Inner {
  70         Inner { fd: fd, lock: unsafe { mutex::NativeMutex::new() } }
  71     }
  72 }
  73
  74 impl Drop for Inner {
  75     fn drop(&mut self) { unsafe { let _ = libc::close(self.fd); } }
  76 }
  77
  78 fn connect(addr: &CString, ty: libc::c_int,
  79            timeout: Option<u64>) -> IoResult<Inner> {
  80     let mut storage = unsafe { mem::zeroed() };
  81     let len = try!(addr_to_sockaddr_un(addr, &mut storage));
  82     let inner = Inner::new(try!(unix_socket(ty)));
  83     let addrp = &storage as *const _ as *const libc::sockaddr;
  84
  85     match timeout {
  86         None => {
  87             match retry(|| unsafe { libc::connect(inner.fd, addrp, len) }) {
  88                 -1 => Err(super::last_error()),
  89                 _  => Ok(inner)
  90             }
  91         }
  92         Some(timeout_ms) => {
  93             try!(util::connect_timeout(inner.fd, addrp, len, timeout_ms));
  94             Ok(inner)
  95         }
  96     }
  97 }
  98
  99 fn bind(addr: &CString, ty: libc::c_int) -> IoResult<Inner> {
 100     let mut storage = unsafe { mem::zeroed() };
 101     let len = try!(addr_to_sockaddr_un(addr, &mut storage));
 102     let inner = Inner::new(try!(unix_socket(ty)));
 103     let addrp = &storage as *const _ as *const libc::sockaddr;
 104     match unsafe {
 105         libc::bind(inner.fd, addrp, len)
 106     } {
 107         -1 => Err(super::last_error()),
 108         _  => Ok(inner)
 109     }
 110 }
 111
 112 ////////////////////////////////////////////////////////////////////////////////
 113 // Unix Streams
 114 ////////////////////////////////////////////////////////////////////////////////
 115
 116 pub struct UnixStream {
 117     inner: Arc<Inner>,
 118     read_deadline: u64,
 119     write_deadline: u64,
 120 }
 121
 122 impl UnixStream {
 123     pub fn connect(addr: &CString,
 124                    timeout: Option<u64>) -> IoResult<UnixStream> {
 125         connect(addr, libc::SOCK_STREAM, timeout).map(|inner| {
 126             UnixStream::new(Arc::new(inner))
 127         })
 128     }
 129
 130     fn new(inner: Arc<Inner>) -> UnixStream {
 131         UnixStream {
 132             inner: inner,
 133             read_deadline: 0,
 134             write_deadline: 0,
 135         }
 136     }
 137
 138     fn fd(&self) -> fd_t { self.inner.fd }
 139
 140     #[cfg(target_os = "linux")]
 141     fn lock_nonblocking(&self) {}
 142
 143     #[cfg(not(target_os = "linux"))]
 144     fn lock_nonblocking<'a>(&'a self) -> net::Guard<'a> {
 145         let ret = net::Guard {
 146             fd: self.fd(),
 147             guard: unsafe { self.inner.lock.lock() },
 148         };
 149         assert!(util::set_nonblocking(self.fd(), true).is_ok());
 150         ret
 151     }
 152 }
 153
 154 impl rtio::RtioPipe for UnixStream {
 155     fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> {
 156         let fd = self.fd();
 157         let dolock = || self.lock_nonblocking();
 158         let doread = |nb| unsafe {
 159             let flags = if nb {c::MSG_DONTWAIT} else {0};
 160             libc::recv(fd,
 161                        buf.as_mut_ptr() as *mut libc::c_void,
 162                        buf.len() as libc::size_t,
 163                        flags) as libc::c_int
 164         };
 165         net::read(fd, self.read_deadline, dolock, doread)
 166     }
 167
 168     fn write(&mut self, buf: &[u8]) -> IoResult<()> {
 169         let fd = self.fd();
 170         let dolock = || self.lock_nonblocking();
 171         let dowrite = |nb: bool, buf: *const u8, len: uint| unsafe {
 172             let flags = if nb {c::MSG_DONTWAIT} else {0};
 173             libc::send(fd,
 174                        buf as *mut libc::c_void,
 175                        len as libc::size_t,
 176                        flags) as i64
 177         };
 178         match net::write(fd, self.write_deadline, buf, true, dolock, dowrite) {
 179             Ok(_) => Ok(()),
 180             Err(e) => Err(e)
 181         }
 182     }
 183
 184     fn clone(&self) -> Box<rtio::RtioPipe + Send> {
 185         box UnixStream::new(self.inner.clone()) as Box<rtio::RtioPipe + Send>
 186     }
 187
 188     fn close_write(&mut self) -> IoResult<()> {
 189         super::mkerr_libc(unsafe { libc::shutdown(self.fd(), libc::SHUT_WR) })
 190     }
 191     fn close_read(&mut self) -> IoResult<()> {
 192         super::mkerr_libc(unsafe { libc::shutdown(self.fd(), libc::SHUT_RD) })
 193     }
 194     fn set_timeout(&mut self, timeout: Option<u64>) {
 195         let deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
 196         self.read_deadline = deadline;
 197         self.write_deadline = deadline;
 198     }
 199     fn set_read_timeout(&mut self, timeout: Option<u64>) {
 200         self.read_deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
 201     }
 202     fn set_write_timeout(&mut self, timeout: Option<u64>) {
 203         self.write_deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
 204     }
 205 }
 206
 207 ////////////////////////////////////////////////////////////////////////////////
 208 // Unix Listener
 209 ////////////////////////////////////////////////////////////////////////////////
 210
 211 pub struct UnixListener {
 212     inner: Inner,
 213     path: CString,
 214 }
 215
 216 impl UnixListener {
 217     pub fn bind(addr: &CString) -> IoResult<UnixListener> {
 218         bind(addr, libc::SOCK_STREAM).map(|fd| {
 219             UnixListener { inner: fd, path: addr.clone() }
 220         })
 221     }
 222
 223     fn fd(&self) -> fd_t { self.inner.fd }
 224
 225     pub fn native_listen(self, backlog: int) -> IoResult<UnixAcceptor> {
 226         match unsafe { libc::listen(self.fd(), backlog as libc::c_int) } {
 227             -1 => Err(super::last_error()),
 228             _ => Ok(UnixAcceptor { listener: self, deadline: 0 })
 229         }
 230     }
 231 }
 232
 233 impl rtio::RtioUnixListener for UnixListener {
 234     fn listen(self: Box<UnixListener>)
 235               -> IoResult<Box<rtio::RtioUnixAcceptor + Send>> {
 236         self.native_listen(128).map(|a| {
 237             box a as Box<rtio::RtioUnixAcceptor + Send>
 238         })
 239     }
 240 }
 241
 242 pub struct UnixAcceptor {
 243     listener: UnixListener,
 244     deadline: u64,
 245 }
 246
 247 impl UnixAcceptor {
 248     fn fd(&self) -> fd_t { self.listener.fd() }
 249
 250     pub fn native_accept(&mut self) -> IoResult<UnixStream> {
 251         if self.deadline != 0 {
 252             try!(util::await(self.fd(), Some(self.deadline), util::Readable));
 253         }
 254         let mut storage: libc::sockaddr_storage = unsafe { mem::zeroed() };
 255         let storagep = &mut storage as *mut libc::sockaddr_storage;
 256         let size = mem::size_of::<libc::sockaddr_storage>();
 257         let mut size = size as libc::socklen_t;
 258         match retry(|| unsafe {
 259             libc::accept(self.fd(),
 260                          storagep as *mut libc::sockaddr,
 261                          &mut size as *mut libc::socklen_t) as libc::c_int
 262         }) {
 263             -1 => Err(super::last_error()),
 264             fd => Ok(UnixStream::new(Arc::new(Inner::new(fd))))
 265         }
 266     }
 267 }
 268
 269 impl rtio::RtioUnixAcceptor for UnixAcceptor {
 270     fn accept(&mut self) -> IoResult<Box<rtio::RtioPipe + Send>> {
 271         self.native_accept().map(|s| box s as Box<rtio::RtioPipe + Send>)
 272     }
 273     fn set_timeout(&mut self, timeout: Option<u64>) {
 274         self.deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
 275     }
 276 }
 277
 278 impl Drop for UnixListener {
 279     fn drop(&mut self) {
 280         // Unlink the path to the socket to ensure that it doesn't linger. We're
 281         // careful to unlink the path before we close the file descriptor to
 282         // prevent races where we unlink someone else's path.
 283         unsafe {
 284             let _ = libc::unlink(self.path.as_ptr());
 285         }
 286     }
 287 }