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