src/libnative/io/net.rs

   1 // Copyright 2013-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::mem;
  14 use std::ptr;
  15 use std::rt::mutex;
  16 use std::rt::rtio;
  17 use std::rt::rtio::{IoResult, IoError};
  18 use std::sync::atomics;
  19
  20 use super::{retry, keep_going};
  21 use super::c;
  22 use super::util;
  23
  24 #[cfg(unix)] use super::process;
  25 #[cfg(unix)] use super::file::FileDesc;
  26
  27 pub use self::os::{init, sock_t, last_error};
  28
  29 ////////////////////////////////////////////////////////////////////////////////
  30 // sockaddr and misc bindings
  31 ////////////////////////////////////////////////////////////////////////////////
  32
  33 pub fn htons(u: u16) -> u16 {
  34     u.to_be()
  35 }
  36 pub fn ntohs(u: u16) -> u16 {
  37     Int::from_be(u)
  38 }
  39
  40 enum InAddr {
  41     InAddr(libc::in_addr),
  42     In6Addr(libc::in6_addr),
  43 }
  44
  45 fn ip_to_inaddr(ip: rtio::IpAddr) -> InAddr {
  46     match ip {
  47         rtio::Ipv4Addr(a, b, c, d) => {
  48             let ip = (a as u32 << 24) |
  49                      (b as u32 << 16) |
  50                      (c as u32 <<  8) |
  51                      (d as u32 <<  0);
  52             InAddr(libc::in_addr {
  53                 s_addr: Int::from_be(ip)
  54             })
  55         }
  56         rtio::Ipv6Addr(a, b, c, d, e, f, g, h) => {
  57             In6Addr(libc::in6_addr {
  58                 s6_addr: [
  59                     htons(a),
  60                     htons(b),
  61                     htons(c),
  62                     htons(d),
  63                     htons(e),
  64                     htons(f),
  65                     htons(g),
  66                     htons(h),
  67                 ]
  68             })
  69         }
  70     }
  71 }
  72
  73 fn addr_to_sockaddr(addr: rtio::SocketAddr,
  74                     storage: &mut libc::sockaddr_storage)
  75                     -> libc::socklen_t {
  76     unsafe {
  77         let len = match ip_to_inaddr(addr.ip) {
  78             InAddr(inaddr) => {
  79                 let storage = storage as *mut _ as *mut libc::sockaddr_in;
  80                 (*storage).sin_family = libc::AF_INET as libc::sa_family_t;
  81                 (*storage).sin_port = htons(addr.port);
  82                 (*storage).sin_addr = inaddr;
  83                 mem::size_of::<libc::sockaddr_in>()
  84             }
  85             In6Addr(inaddr) => {
  86                 let storage = storage as *mut _ as *mut libc::sockaddr_in6;
  87                 (*storage).sin6_family = libc::AF_INET6 as libc::sa_family_t;
  88                 (*storage).sin6_port = htons(addr.port);
  89                 (*storage).sin6_addr = inaddr;
  90                 mem::size_of::<libc::sockaddr_in6>()
  91             }
  92         };
  93         return len as libc::socklen_t;
  94     }
  95 }
  96
  97 fn socket(addr: rtio::SocketAddr, ty: libc::c_int) -> IoResult<sock_t> {
  98     unsafe {
  99         let fam = match addr.ip {
 100             rtio::Ipv4Addr(..) => libc::AF_INET,
 101             rtio::Ipv6Addr(..) => libc::AF_INET6,
 102         };
 103         match libc::socket(fam, ty, 0) {
 104             -1 => Err(os::last_error()),
 105             fd => Ok(fd),
 106         }
 107     }
 108 }
 109
 110 fn setsockopt<T>(fd: sock_t, opt: libc::c_int, val: libc::c_int,
 111                  payload: T) -> IoResult<()> {
 112     unsafe {
 113         let payload = &payload as *const T as *const libc::c_void;
 114         let ret = libc::setsockopt(fd, opt, val,
 115                                    payload,
 116                                    mem::size_of::<T>() as libc::socklen_t);
 117         if ret != 0 {
 118             Err(os::last_error())
 119         } else {
 120             Ok(())
 121         }
 122     }
 123 }
 124
 125 pub fn getsockopt<T: Copy>(fd: sock_t, opt: libc::c_int,
 126                            val: libc::c_int) -> IoResult<T> {
 127     unsafe {
 128         let mut slot: T = mem::zeroed();
 129         let mut len = mem::size_of::<T>() as libc::socklen_t;
 130         let ret = c::getsockopt(fd, opt, val,
 131                                 &mut slot as *mut _ as *mut _,
 132                                 &mut len);
 133         if ret != 0 {
 134             Err(os::last_error())
 135         } else {
 136             assert!(len as uint == mem::size_of::<T>());
 137             Ok(slot)
 138         }
 139     }
 140 }
 141
 142 fn sockname(fd: sock_t,
 143             f: unsafe extern "system" fn(sock_t, *mut libc::sockaddr,
 144                                          *mut libc::socklen_t) -> libc::c_int)
 145     -> IoResult<rtio::SocketAddr>
 146 {
 147     let mut storage: libc::sockaddr_storage = unsafe { mem::zeroed() };
 148     let mut len = mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
 149     unsafe {
 150         let storage = &mut storage as *mut libc::sockaddr_storage;
 151         let ret = f(fd,
 152                     storage as *mut libc::sockaddr,
 153                     &mut len as *mut libc::socklen_t);
 154         if ret != 0 {
 155             return Err(os::last_error())
 156         }
 157     }
 158     return sockaddr_to_addr(&storage, len as uint);
 159 }
 160
 161 pub fn sockaddr_to_addr(storage: &libc::sockaddr_storage,
 162                         len: uint) -> IoResult<rtio::SocketAddr> {
 163     match storage.ss_family as libc::c_int {
 164         libc::AF_INET => {
 165             assert!(len as uint >= mem::size_of::<libc::sockaddr_in>());
 166             let storage: &libc::sockaddr_in = unsafe {
 167                 mem::transmute(storage)
 168             };
 169             let ip = (storage.sin_addr.s_addr as u32).to_be();
 170             let a = (ip >> 24) as u8;
 171             let b = (ip >> 16) as u8;
 172             let c = (ip >>  8) as u8;
 173             let d = (ip >>  0) as u8;
 174             Ok(rtio::SocketAddr {
 175                 ip: rtio::Ipv4Addr(a, b, c, d),
 176                 port: ntohs(storage.sin_port),
 177             })
 178         }
 179         libc::AF_INET6 => {
 180             assert!(len as uint >= mem::size_of::<libc::sockaddr_in6>());
 181             let storage: &libc::sockaddr_in6 = unsafe {
 182                 mem::transmute(storage)
 183             };
 184             let a = ntohs(storage.sin6_addr.s6_addr[0]);
 185             let b = ntohs(storage.sin6_addr.s6_addr[1]);
 186             let c = ntohs(storage.sin6_addr.s6_addr[2]);
 187             let d = ntohs(storage.sin6_addr.s6_addr[3]);
 188             let e = ntohs(storage.sin6_addr.s6_addr[4]);
 189             let f = ntohs(storage.sin6_addr.s6_addr[5]);
 190             let g = ntohs(storage.sin6_addr.s6_addr[6]);
 191             let h = ntohs(storage.sin6_addr.s6_addr[7]);
 192             Ok(rtio::SocketAddr {
 193                 ip: rtio::Ipv6Addr(a, b, c, d, e, f, g, h),
 194                 port: ntohs(storage.sin6_port),
 195             })
 196         }
 197         _ => {
 198             #[cfg(unix)] use libc::EINVAL as ERROR;
 199             #[cfg(windows)] use libc::WSAEINVAL as ERROR;
 200             Err(IoError {
 201                 code: ERROR as uint,
 202                 extra: 0,
 203                 detail: None,
 204             })
 205         }
 206     }
 207 }
 208
 209 ////////////////////////////////////////////////////////////////////////////////
 210 // TCP streams
 211 ////////////////////////////////////////////////////////////////////////////////
 212
 213 pub struct TcpStream {
 214     inner: Arc<Inner>,
 215     read_deadline: u64,
 216     write_deadline: u64,
 217 }
 218
 219 struct Inner {
 220     fd: sock_t,
 221
 222     // Unused on Linux, where this lock is not necessary.
 223     #[allow(dead_code)]
 224     lock: mutex::NativeMutex
 225 }
 226
 227 pub struct Guard<'a> {
 228     pub fd: sock_t,
 229     pub guard: mutex::LockGuard<'a>,
 230 }
 231
 232 impl Inner {
 233     fn new(fd: sock_t) -> Inner {
 234         Inner { fd: fd, lock: unsafe { mutex::NativeMutex::new() } }
 235     }
 236 }
 237
 238 impl TcpStream {
 239     pub fn connect(addr: rtio::SocketAddr,
 240                    timeout: Option<u64>) -> IoResult<TcpStream> {
 241         let fd = try!(socket(addr, libc::SOCK_STREAM));
 242         let ret = TcpStream::new(Inner::new(fd));
 243
 244         let mut storage = unsafe { mem::zeroed() };
 245         let len = addr_to_sockaddr(addr, &mut storage);
 246         let addrp = &storage as *const _ as *const libc::sockaddr;
 247
 248         match timeout {
 249             Some(timeout) => {
 250                 try!(util::connect_timeout(fd, addrp, len, timeout));
 251                 Ok(ret)
 252             },
 253             None => {
 254                 match retry(|| unsafe { libc::connect(fd, addrp, len) }) {
 255                     -1 => Err(os::last_error()),
 256                     _ => Ok(ret),
 257                 }
 258             }
 259         }
 260     }
 261
 262     fn new(inner: Inner) -> TcpStream {
 263         TcpStream {
 264             inner: Arc::new(inner),
 265             read_deadline: 0,
 266             write_deadline: 0,
 267         }
 268     }
 269
 270     pub fn fd(&self) -> sock_t { self.inner.fd }
 271
 272     fn set_nodelay(&mut self, nodelay: bool) -> IoResult<()> {
 273         setsockopt(self.fd(), libc::IPPROTO_TCP, libc::TCP_NODELAY,
 274                    nodelay as libc::c_int)
 275     }
 276
 277     fn set_keepalive(&mut self, seconds: Option<uint>) -> IoResult<()> {
 278         let ret = setsockopt(self.fd(), libc::SOL_SOCKET, libc::SO_KEEPALIVE,
 279                              seconds.is_some() as libc::c_int);
 280         match seconds {
 281             Some(n) => ret.and_then(|()| self.set_tcp_keepalive(n)),
 282             None => ret,
 283         }
 284     }
 285
 286     #[cfg(target_os = "macos")]
 287     #[cfg(target_os = "ios")]
 288     fn set_tcp_keepalive(&mut self, seconds: uint) -> IoResult<()> {
 289         setsockopt(self.fd(), libc::IPPROTO_TCP, libc::TCP_KEEPALIVE,
 290                    seconds as libc::c_int)
 291     }
 292     #[cfg(target_os = "freebsd")]
 293     #[cfg(target_os = "dragonfly")]
 294     fn set_tcp_keepalive(&mut self, seconds: uint) -> IoResult<()> {
 295         setsockopt(self.fd(), libc::IPPROTO_TCP, libc::TCP_KEEPIDLE,
 296                    seconds as libc::c_int)
 297     }
 298     #[cfg(not(target_os = "macos"), not(target_os = "ios"), not(target_os = "freebsd"),
 299       not(target_os = "dragonfly"))]
 300     fn set_tcp_keepalive(&mut self, _seconds: uint) -> IoResult<()> {
 301         Ok(())
 302     }
 303
 304     #[cfg(target_os = "linux")]
 305     fn lock_nonblocking(&self) {}
 306
 307     #[cfg(not(target_os = "linux"))]
 308     fn lock_nonblocking<'a>(&'a self) -> Guard<'a> {
 309         let ret = Guard {
 310             fd: self.fd(),
 311             guard: unsafe { self.inner.lock.lock() },
 312         };
 313         assert!(util::set_nonblocking(self.fd(), true).is_ok());
 314         ret
 315     }
 316 }
 317
 318 #[cfg(windows)] type wrlen = libc::c_int;
 319 #[cfg(not(windows))] type wrlen = libc::size_t;
 320
 321 impl rtio::RtioTcpStream for TcpStream {
 322     fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> {
 323         let fd = self.fd();
 324         let dolock = || self.lock_nonblocking();
 325         let doread = |nb| unsafe {
 326             let flags = if nb {c::MSG_DONTWAIT} else {0};
 327             libc::recv(fd,
 328                        buf.as_mut_ptr() as *mut libc::c_void,
 329                        buf.len() as wrlen,
 330                        flags) as libc::c_int
 331         };
 332         read(fd, self.read_deadline, dolock, doread)
 333     }
 334
 335     fn write(&mut self, buf: &[u8]) -> IoResult<()> {
 336         let fd = self.fd();
 337         let dolock = || self.lock_nonblocking();
 338         let dowrite = |nb: bool, buf: *const u8, len: uint| unsafe {
 339             let flags = if nb {c::MSG_DONTWAIT} else {0};
 340             libc::send(fd,
 341                        buf as *mut libc::c_void,
 342                        len as wrlen,
 343                        flags) as i64
 344         };
 345         match write(fd, self.write_deadline, buf, true, dolock, dowrite) {
 346             Ok(_) => Ok(()),
 347             Err(e) => Err(e)
 348         }
 349     }
 350     fn peer_name(&mut self) -> IoResult<rtio::SocketAddr> {
 351         sockname(self.fd(), libc::getpeername)
 352     }
 353     fn control_congestion(&mut self) -> IoResult<()> {
 354         self.set_nodelay(false)
 355     }
 356     fn nodelay(&mut self) -> IoResult<()> {
 357         self.set_nodelay(true)
 358     }
 359     fn keepalive(&mut self, delay_in_seconds: uint) -> IoResult<()> {
 360         self.set_keepalive(Some(delay_in_seconds))
 361     }
 362     fn letdie(&mut self) -> IoResult<()> {
 363         self.set_keepalive(None)
 364     }
 365
 366     fn clone(&self) -> Box<rtio::RtioTcpStream + Send> {
 367         box TcpStream {
 368             inner: self.inner.clone(),
 369             read_deadline: 0,
 370             write_deadline: 0,
 371         } as Box<rtio::RtioTcpStream + Send>
 372     }
 373
 374     fn close_write(&mut self) -> IoResult<()> {
 375         super::mkerr_libc(unsafe { libc::shutdown(self.fd(), libc::SHUT_WR) })
 376     }
 377     fn close_read(&mut self) -> IoResult<()> {
 378         super::mkerr_libc(unsafe { libc::shutdown(self.fd(), libc::SHUT_RD) })
 379     }
 380
 381     fn set_timeout(&mut self, timeout: Option<u64>) {
 382         let deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
 383         self.read_deadline = deadline;
 384         self.write_deadline = deadline;
 385     }
 386     fn set_read_timeout(&mut self, timeout: Option<u64>) {
 387         self.read_deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
 388     }
 389     fn set_write_timeout(&mut self, timeout: Option<u64>) {
 390         self.write_deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
 391     }
 392 }
 393
 394 impl rtio::RtioSocket for TcpStream {
 395     fn socket_name(&mut self) -> IoResult<rtio::SocketAddr> {
 396         sockname(self.fd(), libc::getsockname)
 397     }
 398 }
 399
 400 impl Drop for Inner {
 401     fn drop(&mut self) { unsafe { os::close(self.fd); } }
 402 }
 403
 404 #[unsafe_destructor]
 405 impl<'a> Drop for Guard<'a> {
 406     fn drop(&mut self) {
 407         assert!(util::set_nonblocking(self.fd, false).is_ok());
 408     }
 409 }
 410
 411 ////////////////////////////////////////////////////////////////////////////////
 412 // TCP listeners
 413 ////////////////////////////////////////////////////////////////////////////////
 414
 415 pub struct TcpListener {
 416     inner: Inner,
 417 }
 418
 419 impl TcpListener {
 420     pub fn bind(addr: rtio::SocketAddr) -> IoResult<TcpListener> {
 421         let fd = try!(socket(addr, libc::SOCK_STREAM));
 422         let ret = TcpListener { inner: Inner::new(fd) };
 423
 424         let mut storage = unsafe { mem::zeroed() };
 425         let len = addr_to_sockaddr(addr, &mut storage);
 426         let addrp = &storage as *const _ as *const libc::sockaddr;
 427
 428         // On platforms with Berkeley-derived sockets, this allows
 429         // to quickly rebind a socket, without needing to wait for
 430         // the OS to clean up the previous one.
 431         if cfg!(unix) {
 432             try!(setsockopt(fd, libc::SOL_SOCKET, libc::SO_REUSEADDR,
 433                             1 as libc::c_int));
 434         }
 435
 436         match unsafe { libc::bind(fd, addrp, len) } {
 437             -1 => Err(os::last_error()),
 438             _ => Ok(ret),
 439         }
 440     }
 441
 442     pub fn fd(&self) -> sock_t { self.inner.fd }
 443
 444     pub fn native_listen(self, backlog: int) -> IoResult<TcpAcceptor> {
 445         match unsafe { libc::listen(self.fd(), backlog as libc::c_int) } {
 446             -1 => Err(os::last_error()),
 447
 448             #[cfg(unix)]
 449             _ => {
 450                 let (reader, writer) = try!(process::pipe());
 451                 try!(util::set_nonblocking(reader.fd(), true));
 452                 try!(util::set_nonblocking(writer.fd(), true));
 453                 try!(util::set_nonblocking(self.fd(), true));
 454                 Ok(TcpAcceptor {
 455                     inner: Arc::new(AcceptorInner {
 456                         listener: self,
 457                         reader: reader,
 458                         writer: writer,
 459                         closed: atomics::AtomicBool::new(false),
 460                     }),
 461                     deadline: 0,
 462                 })
 463             }
 464
 465             #[cfg(windows)]
 466             _ => {
 467                 let accept = try!(os::Event::new());
 468                 let ret = unsafe {
 469                     c::WSAEventSelect(self.fd(), accept.handle(), c::FD_ACCEPT)
 470                 };
 471                 if ret != 0 {
 472                     return Err(os::last_error())
 473                 }
 474                 Ok(TcpAcceptor {
 475                     inner: Arc::new(AcceptorInner {
 476                         listener: self,
 477                         abort: try!(os::Event::new()),
 478                         accept: accept,
 479                         closed: atomics::AtomicBool::new(false),
 480                     }),
 481                     deadline: 0,
 482                 })
 483             }
 484         }
 485     }
 486 }
 487
 488 impl rtio::RtioTcpListener for TcpListener {
 489     fn listen(self: Box<TcpListener>)
 490               -> IoResult<Box<rtio::RtioTcpAcceptor + Send>> {
 491         self.native_listen(128).map(|a| {
 492             box a as Box<rtio::RtioTcpAcceptor + Send>
 493         })
 494     }
 495 }
 496
 497 impl rtio::RtioSocket for TcpListener {
 498     fn socket_name(&mut self) -> IoResult<rtio::SocketAddr> {
 499         sockname(self.fd(), libc::getsockname)
 500     }
 501 }
 502
 503 pub struct TcpAcceptor {
 504     inner: Arc<AcceptorInner>,
 505     deadline: u64,
 506 }
 507
 508 #[cfg(unix)]
 509 struct AcceptorInner {
 510     listener: TcpListener,
 511     reader: FileDesc,
 512     writer: FileDesc,
 513     closed: atomics::AtomicBool,
 514 }
 515
 516 #[cfg(windows)]
 517 struct AcceptorInner {
 518     listener: TcpListener,
 519     abort: os::Event,
 520     accept: os::Event,
 521     closed: atomics::AtomicBool,
 522 }
 523
 524 impl TcpAcceptor {
 525     pub fn fd(&self) -> sock_t { self.inner.listener.fd() }
 526
 527     #[cfg(unix)]
 528     pub fn native_accept(&mut self) -> IoResult<TcpStream> {
 529         let deadline = if self.deadline == 0 {None} else {Some(self.deadline)};
 530
 531         while !self.inner.closed.load(atomics::SeqCst) {
 532             match retry(|| unsafe {
 533                 libc::accept(self.fd(), ptr::mut_null(), ptr::mut_null())
 534             }) {
 535                 -1 if util::wouldblock() => {}
 536                 -1 => return Err(os::last_error()),
 537                 fd => return Ok(TcpStream::new(Inner::new(fd as sock_t))),
 538             }
 539             try!(util::await([self.fd(), self.inner.reader.fd()],
 540                              deadline, util::Readable));
 541         }
 542
 543         Err(util::eof())
 544     }
 545
 546     #[cfg(windows)]
 547     pub fn native_accept(&mut self) -> IoResult<TcpStream> {
 548         let events = [self.inner.abort.handle(), self.inner.accept.handle()];
 549
 550         while !self.inner.closed.load(atomics::SeqCst) {
 551             let ms = if self.deadline == 0 {
 552                 c::WSA_INFINITE as u64
 553             } else {
 554                 let now = ::io::timer::now();
 555                 if self.deadline < now {0} else {now - self.deadline}
 556             };
 557             let ret = unsafe {
 558                 c::WSAWaitForMultipleEvents(2, events.as_ptr(), libc::FALSE,
 559                                             ms as libc::DWORD, libc::FALSE)
 560             };
 561             match ret {
 562                 c::WSA_WAIT_TIMEOUT => {
 563                     return Err(util::timeout("accept timed out"))
 564                 }
 565                 c::WSA_WAIT_FAILED => return Err(os::last_error()),
 566                 c::WSA_WAIT_EVENT_0 => break,
 567                 n => assert_eq!(n, c::WSA_WAIT_EVENT_0 + 1),
 568             }
 569             println!("woke up");
 570
 571             let mut wsaevents: c::WSANETWORKEVENTS = unsafe { mem::zeroed() };
 572             let ret = unsafe {
 573                 c::WSAEnumNetworkEvents(self.fd(), events[1], &mut wsaevents)
 574             };
 575             if ret != 0 { return Err(os::last_error()) }
 576
 577             if wsaevents.lNetworkEvents & c::FD_ACCEPT == 0 { continue }
 578             match unsafe {
 579                 libc::accept(self.fd(), ptr::mut_null(), ptr::mut_null())
 580             } {
 581                 -1 if util::wouldblock() => {}
 582                 -1 => return Err(os::last_error()),
 583                 fd => return Ok(TcpStream::new(Inner::new(fd))),
 584             }
 585         }
 586
 587         Err(util::eof())
 588     }
 589 }
 590
 591 impl rtio::RtioSocket for TcpAcceptor {
 592     fn socket_name(&mut self) -> IoResult<rtio::SocketAddr> {
 593         sockname(self.fd(), libc::getsockname)
 594     }
 595 }
 596
 597 impl rtio::RtioTcpAcceptor for TcpAcceptor {
 598     fn accept(&mut self) -> IoResult<Box<rtio::RtioTcpStream + Send>> {
 599         self.native_accept().map(|s| box s as Box<rtio::RtioTcpStream + Send>)
 600     }
 601
 602     fn accept_simultaneously(&mut self) -> IoResult<()> { Ok(()) }
 603     fn dont_accept_simultaneously(&mut self) -> IoResult<()> { Ok(()) }
 604     fn set_timeout(&mut self, timeout: Option<u64>) {
 605         self.deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
 606     }
 607
 608     fn clone(&self) -> Box<rtio::RtioTcpAcceptor + Send> {
 609         box TcpAcceptor {
 610             inner: self.inner.clone(),
 611             deadline: 0,
 612         } as Box<rtio::RtioTcpAcceptor + Send>
 613     }
 614
 615     #[cfg(unix)]
 616     fn close_accept(&mut self) -> IoResult<()> {
 617         self.inner.closed.store(true, atomics::SeqCst);
 618         let mut fd = FileDesc::new(self.inner.writer.fd(), false);
 619         match fd.inner_write([0]) {
 620             Ok(..) => Ok(()),
 621             Err(..) if util::wouldblock() => Ok(()),
 622             Err(e) => Err(e),
 623         }
 624     }
 625
 626     #[cfg(windows)]
 627     fn close_accept(&mut self) -> IoResult<()> {
 628         self.inner.closed.store(true, atomics::SeqCst);
 629         let ret = unsafe { c::WSASetEvent(self.inner.abort.handle()) };
 630         if ret == libc::TRUE {
 631             Ok(())
 632         } else {
 633             Err(os::last_error())
 634         }
 635     }
 636 }
 637
 638 ////////////////////////////////////////////////////////////////////////////////
 639 // UDP
 640 ////////////////////////////////////////////////////////////////////////////////
 641
 642 pub struct UdpSocket {
 643     inner: Arc<Inner>,
 644     read_deadline: u64,
 645     write_deadline: u64,
 646 }
 647
 648 impl UdpSocket {
 649     pub fn bind(addr: rtio::SocketAddr) -> IoResult<UdpSocket> {
 650         let fd = try!(socket(addr, libc::SOCK_DGRAM));
 651         let ret = UdpSocket {
 652             inner: Arc::new(Inner::new(fd)),
 653             read_deadline: 0,
 654             write_deadline: 0,
 655         };
 656
 657         let mut storage = unsafe { mem::zeroed() };
 658         let len = addr_to_sockaddr(addr, &mut storage);
 659         let addrp = &storage as *const _ as *const libc::sockaddr;
 660
 661         match unsafe { libc::bind(fd, addrp, len) } {
 662             -1 => Err(os::last_error()),
 663             _ => Ok(ret),
 664         }
 665     }
 666
 667     pub fn fd(&self) -> sock_t { self.inner.fd }
 668
 669     pub fn set_broadcast(&mut self, on: bool) -> IoResult<()> {
 670         setsockopt(self.fd(), libc::SOL_SOCKET, libc::SO_BROADCAST,
 671                    on as libc::c_int)
 672     }
 673
 674     pub fn set_multicast_loop(&mut self, on: bool) -> IoResult<()> {
 675         setsockopt(self.fd(), libc::IPPROTO_IP, libc::IP_MULTICAST_LOOP,
 676                    on as libc::c_int)
 677     }
 678
 679     pub fn set_membership(&mut self, addr: rtio::IpAddr,
 680                           opt: libc::c_int) -> IoResult<()> {
 681         match ip_to_inaddr(addr) {
 682             InAddr(addr) => {
 683                 let mreq = libc::ip_mreq {
 684                     imr_multiaddr: addr,
 685                     // interface == INADDR_ANY
 686                     imr_interface: libc::in_addr { s_addr: 0x0 },
 687                 };
 688                 setsockopt(self.fd(), libc::IPPROTO_IP, opt, mreq)
 689             }
 690             In6Addr(addr) => {
 691                 let mreq = libc::ip6_mreq {
 692                     ipv6mr_multiaddr: addr,
 693                     ipv6mr_interface: 0,
 694                 };
 695                 setsockopt(self.fd(), libc::IPPROTO_IPV6, opt, mreq)
 696             }
 697         }
 698     }
 699
 700     #[cfg(target_os = "linux")]
 701     fn lock_nonblocking(&self) {}
 702
 703     #[cfg(not(target_os = "linux"))]
 704     fn lock_nonblocking<'a>(&'a self) -> Guard<'a> {
 705         let ret = Guard {
 706             fd: self.fd(),
 707             guard: unsafe { self.inner.lock.lock() },
 708         };
 709         assert!(util::set_nonblocking(self.fd(), true).is_ok());
 710         ret
 711     }
 712 }
 713
 714 impl rtio::RtioSocket for UdpSocket {
 715     fn socket_name(&mut self) -> IoResult<rtio::SocketAddr> {
 716         sockname(self.fd(), libc::getsockname)
 717     }
 718 }
 719
 720 #[cfg(windows)] type msglen_t = libc::c_int;
 721 #[cfg(unix)]    type msglen_t = libc::size_t;
 722
 723 impl rtio::RtioUdpSocket for UdpSocket {
 724     fn recv_from(&mut self, buf: &mut [u8]) -> IoResult<(uint, rtio::SocketAddr)> {
 725         let fd = self.fd();
 726         let mut storage: libc::sockaddr_storage = unsafe { mem::zeroed() };
 727         let storagep = &mut storage as *mut _ as *mut libc::sockaddr;
 728         let mut addrlen: libc::socklen_t =
 729                 mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
 730
 731         let dolock = || self.lock_nonblocking();
 732         let n = try!(read(fd, self.read_deadline, dolock, |nb| unsafe {
 733             let flags = if nb {c::MSG_DONTWAIT} else {0};
 734             libc::recvfrom(fd,
 735                            buf.as_mut_ptr() as *mut libc::c_void,
 736                            buf.len() as msglen_t,
 737                            flags,
 738                            storagep,
 739                            &mut addrlen) as libc::c_int
 740         }));
 741         sockaddr_to_addr(&storage, addrlen as uint).and_then(|addr| {
 742             Ok((n as uint, addr))
 743         })
 744     }
 745
 746     fn send_to(&mut self, buf: &[u8], dst: rtio::SocketAddr) -> IoResult<()> {
 747         let mut storage = unsafe { mem::zeroed() };
 748         let dstlen = addr_to_sockaddr(dst, &mut storage);
 749         let dstp = &storage as *const _ as *const libc::sockaddr;
 750
 751         let fd = self.fd();
 752         let dolock = || self.lock_nonblocking();
 753         let dowrite = |nb, buf: *const u8, len: uint| unsafe {
 754             let flags = if nb {c::MSG_DONTWAIT} else {0};
 755             libc::sendto(fd,
 756                          buf as *const libc::c_void,
 757                          len as msglen_t,
 758                          flags,
 759                          dstp,
 760                          dstlen) as i64
 761         };
 762
 763         let n = try!(write(fd, self.write_deadline, buf, false, dolock, dowrite));
 764         if n != buf.len() {
 765             Err(util::short_write(n, "couldn't send entire packet at once"))
 766         } else {
 767             Ok(())
 768         }
 769     }
 770
 771     fn join_multicast(&mut self, multi: rtio::IpAddr) -> IoResult<()> {
 772         match multi {
 773             rtio::Ipv4Addr(..) => {
 774                 self.set_membership(multi, libc::IP_ADD_MEMBERSHIP)
 775             }
 776             rtio::Ipv6Addr(..) => {
 777                 self.set_membership(multi, libc::IPV6_ADD_MEMBERSHIP)
 778             }
 779         }
 780     }
 781     fn leave_multicast(&mut self, multi: rtio::IpAddr) -> IoResult<()> {
 782         match multi {
 783             rtio::Ipv4Addr(..) => {
 784                 self.set_membership(multi, libc::IP_DROP_MEMBERSHIP)
 785             }
 786             rtio::Ipv6Addr(..) => {
 787                 self.set_membership(multi, libc::IPV6_DROP_MEMBERSHIP)
 788             }
 789         }
 790     }
 791
 792     fn loop_multicast_locally(&mut self) -> IoResult<()> {
 793         self.set_multicast_loop(true)
 794     }
 795     fn dont_loop_multicast_locally(&mut self) -> IoResult<()> {
 796         self.set_multicast_loop(false)
 797     }
 798
 799     fn multicast_time_to_live(&mut self, ttl: int) -> IoResult<()> {
 800         setsockopt(self.fd(), libc::IPPROTO_IP, libc::IP_MULTICAST_TTL,
 801                    ttl as libc::c_int)
 802     }
 803     fn time_to_live(&mut self, ttl: int) -> IoResult<()> {
 804         setsockopt(self.fd(), libc::IPPROTO_IP, libc::IP_TTL, ttl as libc::c_int)
 805     }
 806
 807     fn hear_broadcasts(&mut self) -> IoResult<()> {
 808         self.set_broadcast(true)
 809     }
 810     fn ignore_broadcasts(&mut self) -> IoResult<()> {
 811         self.set_broadcast(false)
 812     }
 813
 814     fn clone(&self) -> Box<rtio::RtioUdpSocket + Send> {
 815         box UdpSocket {
 816             inner: self.inner.clone(),
 817             read_deadline: 0,
 818             write_deadline: 0,
 819         } as Box<rtio::RtioUdpSocket + Send>
 820     }
 821
 822     fn set_timeout(&mut self, timeout: Option<u64>) {
 823         let deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
 824         self.read_deadline = deadline;
 825         self.write_deadline = deadline;
 826     }
 827     fn set_read_timeout(&mut self, timeout: Option<u64>) {
 828         self.read_deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
 829     }
 830     fn set_write_timeout(&mut self, timeout: Option<u64>) {
 831         self.write_deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
 832     }
 833 }
 834
 835 ////////////////////////////////////////////////////////////////////////////////
 836 // Timeout helpers
 837 //
 838 // The read/write functions below are the helpers for reading/writing a socket
 839 // with a possible deadline specified. This is generally viewed as a timed out
 840 // I/O operation.
 841 //
 842 // From the application's perspective, timeouts apply to the I/O object, not to
 843 // the underlying file descriptor (it's one timeout per object). This means that
 844 // we can't use the SO_RCVTIMEO and corresponding send timeout option.
 845 //
 846 // The next idea to implement timeouts would be to use nonblocking I/O. An
 847 // invocation of select() would wait (with a timeout) for a socket to be ready.
 848 // Once its ready, we can perform the operation. Note that the operation *must*
 849 // be nonblocking, even though select() says the socket is ready. This is
 850 // because some other thread could have come and stolen our data (handles can be
 851 // cloned).
 852 //
 853 // To implement nonblocking I/O, the first option we have is to use the
 854 // O_NONBLOCK flag. Remember though that this is a global setting, affecting all
 855 // I/O objects, so this was initially viewed as unwise.
 856 //
 857 // It turns out that there's this nifty MSG_DONTWAIT flag which can be passed to
 858 // send/recv, but the niftiness wears off once you realize it only works well on
 859 // linux [1] [2]. This means that it's pretty easy to get a nonblocking
 860 // operation on linux (no flag fiddling, no affecting other objects), but not on
 861 // other platforms.
 862 //
 863 // To work around this constraint on other platforms, we end up using the
 864 // original strategy of flipping the O_NONBLOCK flag. As mentioned before, this
 865 // could cause other objects' blocking operations to suddenly become
 866 // nonblocking. To get around this, a "blocking operation" which returns EAGAIN
 867 // falls back to using the same code path as nonblocking operations, but with an
 868 // infinite timeout (select + send/recv). This helps emulate blocking
 869 // reads/writes despite the underlying descriptor being nonblocking, as well as
 870 // optimizing the fast path of just hitting one syscall in the good case.
 871 //
 872 // As a final caveat, this implementation uses a mutex so only one thread is
 873 // doing a nonblocking operation at at time. This is the operation that comes
 874 // after the select() (at which point we think the socket is ready). This is
 875 // done for sanity to ensure that the state of the O_NONBLOCK flag is what we
 876 // expect (wouldn't want someone turning it on when it should be off!). All
 877 // operations performed in the lock are *nonblocking* to avoid holding the mutex
 878 // forever.
 879 //
 880 // So, in summary, linux uses MSG_DONTWAIT and doesn't need mutexes, everyone
 881 // else uses O_NONBLOCK and mutexes with some trickery to make sure blocking
 882 // reads/writes are still blocking.
 883 //
 884 // Fun, fun!
 885 //
 886 // [1] http://twistedmatrix.com/pipermail/twisted-commits/2012-April/034692.html
 887 // [2] http://stackoverflow.com/questions/19819198/does-send-msg-dontwait
 888
 889 pub fn read<T>(fd: sock_t,
 890                deadline: u64,
 891                lock: || -> T,
 892                read: |bool| -> libc::c_int) -> IoResult<uint> {
 893     let mut ret = -1;
 894     if deadline == 0 {
 895         ret = retry(|| read(false));
 896     }
 897
 898     if deadline != 0 || (ret == -1 && util::wouldblock()) {
 899         let deadline = match deadline {
 900             0 => None,
 901             n => Some(n),
 902         };
 903         loop {
 904             // With a timeout, first we wait for the socket to become
 905             // readable using select(), specifying the relevant timeout for
 906             // our previously set deadline.
 907             try!(util::await([fd], deadline, util::Readable));
 908
 909             // At this point, we're still within the timeout, and we've
 910             // determined that the socket is readable (as returned by
 911             // select). We must still read the socket in *nonblocking* mode
 912             // because some other thread could come steal our data. If we
 913             // fail to read some data, we retry (hence the outer loop) and
 914             // wait for the socket to become readable again.
 915             let _guard = lock();
 916             match retry(|| read(deadline.is_some())) {
 917                 -1 if util::wouldblock() => { assert!(deadline.is_some()); }
 918                 -1 => return Err(os::last_error()),
 919                n => { ret = n; break }
 920             }
 921         }
 922     }
 923
 924     match ret {
 925         0 => Err(util::eof()),
 926         n if n < 0 => Err(os::last_error()),
 927         n => Ok(n as uint)
 928     }
 929 }
 930
 931 pub fn write<T>(fd: sock_t,
 932                 deadline: u64,
 933                 buf: &[u8],
 934                 write_everything: bool,
 935                 lock: || -> T,
 936                 write: |bool, *const u8, uint| -> i64) -> IoResult<uint> {
 937     let mut ret = -1;
 938     let mut written = 0;
 939     if deadline == 0 {
 940         if write_everything {
 941             ret = keep_going(buf, |inner, len| {
 942                 written = buf.len() - len;
 943                 write(false, inner, len)
 944             });
 945         } else {
 946             ret = retry(|| {
 947                 write(false, buf.as_ptr(), buf.len()) as libc::c_int
 948             }) as i64;
 949             if ret > 0 { written = ret as uint; }
 950         }
 951     }
 952
 953     if deadline != 0 || (ret == -1 && util::wouldblock()) {
 954         let deadline = match deadline {
 955             0 => None,
 956             n => Some(n),
 957         };
 958         while written < buf.len() && (write_everything || written == 0) {
 959             // As with read(), first wait for the socket to be ready for
 960             // the I/O operation.
 961             match util::await([fd], deadline, util::Writable) {
 962                 Err(ref e) if e.code == libc::EOF as uint && written > 0 => {
 963                     assert!(deadline.is_some());
 964                     return Err(util::short_write(written, "short write"))
 965                 }
 966                 Err(e) => return Err(e),
 967                 Ok(()) => {}
 968             }
 969
 970             // Also as with read(), we use MSG_DONTWAIT to guard ourselves
 971             // against unforeseen circumstances.
 972             let _guard = lock();
 973             let ptr = buf.slice_from(written).as_ptr();
 974             let len = buf.len() - written;
 975             match retry(|| write(deadline.is_some(), ptr, len) as libc::c_int) {
 976                 -1 if util::wouldblock() => {}
 977                 -1 => return Err(os::last_error()),
 978                 n => { written += n as uint; }
 979             }
 980         }
 981         ret = 0;
 982     }
 983     if ret < 0 {
 984         Err(os::last_error())
 985     } else {
 986         Ok(written)
 987     }
 988 }
 989
 990 #[cfg(windows)]
 991 mod os {
 992     use libc;
 993     use std::mem;
 994     use std::rt::rtio::{IoError, IoResult};
 995
 996     use io::c;
 997
 998     pub type sock_t = libc::SOCKET;
 999     pub struct Event(c::WSAEVENT);
1000
1001     impl Event {
1002         pub fn new() -> IoResult<Event> {
1003             let event = unsafe { c::WSACreateEvent() };
1004             if event == c::WSA_INVALID_EVENT {
1005                 Err(last_error())
1006             } else {
1007                 Ok(Event(event))
1008             }
1009         }
1010
1011         pub fn handle(&self) -> c::WSAEVENT { let Event(handle) = *self; handle }
1012     }
1013
1014     impl Drop for Event {
1015         fn drop(&mut self) {
1016             unsafe { let _ = c::WSACloseEvent(self.handle()); }
1017         }
1018     }
1019
1020     pub fn init() {
1021         unsafe {
1022             use std::rt::mutex::{StaticNativeMutex, NATIVE_MUTEX_INIT};
1023             static mut INITIALIZED: bool = false;
1024             static mut LOCK: StaticNativeMutex = NATIVE_MUTEX_INIT;
1025
1026             let _guard = LOCK.lock();
1027             if !INITIALIZED {
1028                 let mut data: c::WSADATA = mem::zeroed();
1029                 let ret = c::WSAStartup(0x202,      // version 2.2
1030                                         &mut data);
1031                 assert_eq!(ret, 0);
1032                 INITIALIZED = true;
1033             }
1034         }
1035     }
1036
1037     pub fn last_error() -> IoError {
1038         use std::os;
1039         let code = unsafe { c::WSAGetLastError() as uint };
1040         IoError {
1041             code: code,
1042             extra: 0,
1043             detail: Some(os::error_string(code)),
1044         }
1045     }
1046
1047     pub unsafe fn close(sock: sock_t) { let _ = libc::closesocket(sock); }
1048 }
1049
1050 #[cfg(unix)]
1051 mod os {
1052     use libc;
1053     use std::rt::rtio::IoError;
1054     use io;
1055
1056     pub type sock_t = io::file::fd_t;
1057
1058     pub fn init() {}
1059     pub fn last_error() -> IoError { io::last_error() }
1060     pub unsafe fn close(sock: sock_t) { let _ = libc::close(sock); }
1061 }