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