src/libstd/sys/unix/pipe.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 prelude::v1::*;
  12
  13 use libc;
  14 use c_str::CString;
  15 use mem;
  16 use sync::{atomic, Arc, Mutex};
  17 use io::{mod, IoResult, IoError};
  18
  19 use sys::{mod, timer, retry, c, set_nonblocking, wouldblock};
  20 use sys::fs::{fd_t, FileDesc};
  21 use sys_common::net::*;
  22 use sys_common::net::SocketStatus::*;
  23 use sys_common::{eof, mkerr_libc};
  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         return Err(IoError {
  43             kind: io::InvalidInput,
  44             desc: "invalid argument: path must be smaller than SUN_LEN",
  45             detail: None,
  46         })
  47     }
  48     s.sun_family = libc::AF_UNIX as libc::sa_family_t;
  49     for (slot, value) in s.sun_path.iter_mut().zip(addr.iter()) {
  50         *slot = value;
  51     }
  52
  53     // count the null terminator
  54     let len = mem::size_of::<libc::sa_family_t>() + len + 1;
  55     return Ok(len as libc::socklen_t);
  56 }
  57
  58 struct Inner {
  59     fd: fd_t,
  60
  61     // Unused on Linux, where this lock is not necessary.
  62     #[allow(dead_code)]
  63     lock: Mutex<()>,
  64 }
  65
  66 impl Inner {
  67     fn new(fd: fd_t) -> Inner {
  68         Inner { fd: fd, lock: Mutex::new(()) }
  69     }
  70 }
  71
  72 impl Drop for Inner {
  73     fn drop(&mut self) { unsafe { let _ = libc::close(self.fd); } }
  74 }
  75
  76 fn connect(addr: &CString, ty: libc::c_int,
  77            timeout: Option<u64>) -> IoResult<Inner> {
  78     let mut storage = unsafe { mem::zeroed() };
  79     let len = try!(addr_to_sockaddr_un(addr, &mut storage));
  80     let inner = Inner::new(try!(unix_socket(ty)));
  81     let addrp = &storage as *const _ as *const libc::sockaddr;
  82
  83     match timeout {
  84         None => {
  85             match retry(|| unsafe { libc::connect(inner.fd, addrp, len) }) {
  86                 -1 => Err(super::last_error()),
  87                 _  => Ok(inner)
  88             }
  89         }
  90         Some(timeout_ms) => {
  91             try!(connect_timeout(inner.fd, addrp, len, timeout_ms));
  92             Ok(inner)
  93         }
  94     }
  95 }
  96
  97 fn bind(addr: &CString, ty: libc::c_int) -> IoResult<Inner> {
  98     let mut storage = unsafe { mem::zeroed() };
  99     let len = try!(addr_to_sockaddr_un(addr, &mut storage));
 100     let inner = Inner::new(try!(unix_socket(ty)));
 101     let addrp = &storage as *const _ as *const libc::sockaddr;
 102     match unsafe {
 103         libc::bind(inner.fd, addrp, len)
 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     pub 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) -> Guard<'a> {
 143         let ret = Guard {
 144             fd: self.fd(),
 145             guard: unsafe { self.inner.lock.lock().unwrap() },
 146         };
 147         assert!(set_nonblocking(self.fd(), true).is_ok());
 148         ret
 149     }
 150
 151     pub fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> {
 152         let fd = self.fd();
 153         let dolock = |&:| self.lock_nonblocking();
 154         let doread = |&mut: nb| unsafe {
 155             let flags = if nb {c::MSG_DONTWAIT} else {0};
 156             libc::recv(fd,
 157                        buf.as_mut_ptr() as *mut libc::c_void,
 158                        buf.len() as libc::size_t,
 159                        flags) as libc::c_int
 160         };
 161         read(fd, self.read_deadline, dolock, doread)
 162     }
 163
 164     pub fn write(&mut self, buf: &[u8]) -> IoResult<()> {
 165         let fd = self.fd();
 166         let dolock = |&: | self.lock_nonblocking();
 167         let dowrite = |&: nb: bool, buf: *const u8, len: uint| unsafe {
 168             let flags = if nb {c::MSG_DONTWAIT} else {0};
 169             libc::send(fd,
 170                        buf as *const _,
 171                        len as libc::size_t,
 172                        flags) as i64
 173         };
 174         match write(fd, self.write_deadline, buf, true, dolock, dowrite) {
 175             Ok(_) => Ok(()),
 176             Err(e) => Err(e)
 177         }
 178     }
 179
 180     pub fn close_write(&mut self) -> IoResult<()> {
 181         mkerr_libc(unsafe { libc::shutdown(self.fd(), libc::SHUT_WR) })
 182     }
 183
 184     pub fn close_read(&mut self) -> IoResult<()> {
 185         mkerr_libc(unsafe { libc::shutdown(self.fd(), libc::SHUT_RD) })
 186     }
 187
 188     pub fn set_timeout(&mut self, timeout: Option<u64>) {
 189         let deadline = timeout.map(|a| timer::now() + a).unwrap_or(0);
 190         self.read_deadline = deadline;
 191         self.write_deadline = deadline;
 192     }
 193
 194     pub fn set_read_timeout(&mut self, timeout: Option<u64>) {
 195         self.read_deadline = timeout.map(|a| timer::now() + a).unwrap_or(0);
 196     }
 197
 198     pub fn set_write_timeout(&mut self, timeout: Option<u64>) {
 199         self.write_deadline = timeout.map(|a| timer::now() + a).unwrap_or(0);
 200     }
 201 }
 202
 203 impl Clone for UnixStream {
 204     fn clone(&self) -> UnixStream {
 205         UnixStream::new(self.inner.clone())
 206     }
 207 }
 208
 209 ////////////////////////////////////////////////////////////////////////////////
 210 // Unix Listener
 211 ////////////////////////////////////////////////////////////////////////////////
 212
 213 pub struct UnixListener {
 214     inner: Inner,
 215     path: CString,
 216 }
 217
 218 // we currently own the CString, so these impls should be safe
 219 unsafe impl Send for UnixListener {}
 220 unsafe impl Sync for UnixListener {}
 221
 222 impl UnixListener {
 223     pub fn bind(addr: &CString) -> IoResult<UnixListener> {
 224         bind(addr, libc::SOCK_STREAM).map(|fd| {
 225             UnixListener { inner: fd, path: addr.clone() }
 226         })
 227     }
 228
 229     pub fn fd(&self) -> fd_t { self.inner.fd }
 230
 231     pub fn listen(self) -> IoResult<UnixAcceptor> {
 232         match unsafe { libc::listen(self.fd(), 128) } {
 233             -1 => Err(super::last_error()),
 234
 235             _ => {
 236                 let (reader, writer) = try!(unsafe { sys::os::pipe() });
 237                 try!(set_nonblocking(reader.fd(), true));
 238                 try!(set_nonblocking(writer.fd(), true));
 239                 try!(set_nonblocking(self.fd(), true));
 240                 Ok(UnixAcceptor {
 241                     inner: Arc::new(AcceptorInner {
 242                         listener: self,
 243                         reader: reader,
 244                         writer: writer,
 245                         closed: atomic::AtomicBool::new(false),
 246                     }),
 247                     deadline: 0,
 248                 })
 249             }
 250         }
 251     }
 252 }
 253
 254 pub struct UnixAcceptor {
 255     inner: Arc<AcceptorInner>,
 256     deadline: u64,
 257 }
 258
 259 struct AcceptorInner {
 260     listener: UnixListener,
 261     reader: FileDesc,
 262     writer: FileDesc,
 263     closed: atomic::AtomicBool,
 264 }
 265
 266 impl UnixAcceptor {
 267     pub fn fd(&self) -> fd_t { self.inner.listener.fd() }
 268
 269     pub fn accept(&mut self) -> IoResult<UnixStream> {
 270         let deadline = if self.deadline == 0 {None} else {Some(self.deadline)};
 271
 272         while !self.inner.closed.load(atomic::SeqCst) {
 273             unsafe {
 274                 let mut storage: libc::sockaddr_storage = mem::zeroed();
 275                 let storagep = &mut storage as *mut libc::sockaddr_storage;
 276                 let size = mem::size_of::<libc::sockaddr_storage>();
 277                 let mut size = size as libc::socklen_t;
 278                 match retry(|| {
 279                     libc::accept(self.fd(),
 280                                  storagep as *mut libc::sockaddr,
 281                                  &mut size as *mut libc::socklen_t) as libc::c_int
 282                 }) {
 283                     -1 if wouldblock() => {}
 284                     -1 => return Err(super::last_error()),
 285                     fd => return Ok(UnixStream::new(Arc::new(Inner::new(fd)))),
 286                 }
 287             }
 288             try!(await(&[self.fd(), self.inner.reader.fd()],
 289                        deadline, Readable));
 290         }
 291
 292         Err(eof())
 293     }
 294
 295     pub fn set_timeout(&mut self, timeout: Option<u64>) {
 296         self.deadline = timeout.map(|a| timer::now() + a).unwrap_or(0);
 297     }
 298
 299     pub fn close_accept(&mut self) -> IoResult<()> {
 300         self.inner.closed.store(true, atomic::SeqCst);
 301         let fd = FileDesc::new(self.inner.writer.fd(), false);
 302         match fd.write(&[0]) {
 303             Ok(..) => Ok(()),
 304             Err(..) if wouldblock() => Ok(()),
 305             Err(e) => Err(e),
 306         }
 307     }
 308 }
 309
 310 impl Clone for UnixAcceptor {
 311     fn clone(&self) -> UnixAcceptor {
 312         UnixAcceptor { inner: self.inner.clone(), deadline: 0 }
 313     }
 314 }
 315
 316 impl Drop for UnixListener {
 317     fn drop(&mut self) {
 318         // Unlink the path to the socket to ensure that it doesn't linger. We're
 319         // careful to unlink the path before we close the file descriptor to
 320         // prevent races where we unlink someone else's path.
 321         unsafe {
 322             let _ = libc::unlink(self.path.as_ptr());
 323         }
 324     }
 325 }