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