src/libstd/net/tcp.rs

   1 // Copyright 2015 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 #![unstable(feature = "tcp", reason = "remaining functions have not been \
  12                                        scrutinized enough to be stabilized")]
  13
  14 use prelude::v1::*;
  15 use io::prelude::*;
  16
  17 use fmt;
  18 use io;
  19 use net::{ToSocketAddrs, SocketAddr, Shutdown};
  20 use sys_common::net as net_imp;
  21 use sys_common::{AsInner, FromInner};
  22 use time::Duration;
  23
  24 /// A structure which represents a TCP stream between a local socket and a
  25 /// remote socket.
  26 ///
  27 /// The socket will be closed when the value is dropped.
  28 ///
  29 /// # Examples
  30 ///
  31 /// ```no_run
  32 /// use std::io::prelude::*;
  33 /// use std::net::TcpStream;
  34 ///
  35 /// {
  36 ///     let mut stream = TcpStream::connect("127.0.0.1:34254").unwrap();
  37 ///
  38 ///     // ignore the Result
  39 ///     let _ = stream.write(&[1]);
  40 ///     let _ = stream.read(&mut [0; 128]); // ignore here too
  41 /// } // the stream is closed here
  42 /// ```
  43 #[stable(feature = "rust1", since = "1.0.0")]
  44 pub struct TcpStream(net_imp::TcpStream);
  45
  46 /// A structure representing a socket server.
  47 ///
  48 /// # Examples
  49 ///
  50 /// ```no_run
  51 /// use std::net::{TcpListener, TcpStream};
  52 /// use std::thread;
  53 ///
  54 /// let listener = TcpListener::bind("127.0.0.1:80").unwrap();
  55 ///
  56 /// fn handle_client(stream: TcpStream) {
  57 ///     // ...
  58 /// }
  59 ///
  60 /// // accept connections and process them, spawning a new thread for each one
  61 /// for stream in listener.incoming() {
  62 ///     match stream {
  63 ///         Ok(stream) => {
  64 ///             thread::spawn(move|| {
  65 ///                 // connection succeeded
  66 ///                 handle_client(stream)
  67 ///             });
  68 ///         }
  69 ///         Err(e) => { /* connection failed */ }
  70 ///     }
  71 /// }
  72 ///
  73 /// // close the socket server
  74 /// drop(listener);
  75 /// ```
  76 #[stable(feature = "rust1", since = "1.0.0")]
  77 pub struct TcpListener(net_imp::TcpListener);
  78
  79 /// An infinite iterator over the connections from a `TcpListener`.
  80 ///
  81 /// This iterator will infinitely yield `Some` of the accepted connections. It
  82 /// is equivalent to calling `accept` in a loop.
  83 #[stable(feature = "rust1", since = "1.0.0")]
  84 pub struct Incoming<'a> { listener: &'a TcpListener }
  85
  86 impl TcpStream {
  87     /// Opens a TCP connection to a remote host.
  88     ///
  89     /// `addr` is an address of the remote host. Anything which implements
  90     /// `ToSocketAddrs` trait can be supplied for the address; see this trait
  91     /// documentation for concrete examples.
  92     #[stable(feature = "rust1", since = "1.0.0")]
  93     pub fn connect<A: ToSocketAddrs>(addr: A) -> io::Result<TcpStream> {
  94         super::each_addr(addr, net_imp::TcpStream::connect).map(TcpStream)
  95     }
  96
  97     /// Returns the socket address of the remote peer of this TCP connection.
  98     #[stable(feature = "rust1", since = "1.0.0")]
  99     pub fn peer_addr(&self) -> io::Result<SocketAddr> {
 100         self.0.peer_addr()
 101     }
 102
 103     /// Returns the socket address of the local half of this TCP connection.
 104     #[stable(feature = "rust1", since = "1.0.0")]
 105     pub fn local_addr(&self) -> io::Result<SocketAddr> {
 106         self.0.socket_addr()
 107     }
 108
 109     /// Shuts down the read, write, or both halves of this connection.
 110     ///
 111     /// This function will cause all pending and future I/O on the specified
 112     /// portions to return immediately with an appropriate value (see the
 113     /// documentation of `Shutdown`).
 114     #[stable(feature = "rust1", since = "1.0.0")]
 115     pub fn shutdown(&self, how: Shutdown) -> io::Result<()> {
 116         self.0.shutdown(how)
 117     }
 118
 119     /// Creates a new independently owned handle to the underlying socket.
 120     ///
 121     /// The returned `TcpStream` is a reference to the same stream that this
 122     /// object references. Both handles will read and write the same stream of
 123     /// data, and options set on one stream will be propagated to the other
 124     /// stream.
 125     #[stable(feature = "rust1", since = "1.0.0")]
 126     pub fn try_clone(&self) -> io::Result<TcpStream> {
 127         self.0.duplicate().map(TcpStream)
 128     }
 129
 130     /// Sets the nodelay flag on this connection to the boolean specified.
 131     pub fn set_nodelay(&self, nodelay: bool) -> io::Result<()> {
 132         self.0.set_nodelay(nodelay)
 133     }
 134
 135     /// Sets the keepalive timeout to the timeout specified.
 136     ///
 137     /// If the value specified is `None`, then the keepalive flag is cleared on
 138     /// this connection. Otherwise, the keepalive timeout will be set to the
 139     /// specified time, in seconds.
 140     pub fn set_keepalive(&self, seconds: Option<u32>) -> io::Result<()> {
 141         self.0.set_keepalive(seconds)
 142     }
 143
 144     /// Sets the read timeout to the timeout specified.
 145     ///
 146     /// If the value specified is `None`, then `read` calls will block
 147     /// indefinitely. It is an error to pass the zero `Duration` to this
 148     /// method.
 149     #[unstable(feature = "socket_timeout", reason = "RFC 1047 - recently added")]
 150     pub fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
 151         self.0.set_read_timeout(dur)
 152     }
 153
 154     /// Sets the write timeout to the timeout specified.
 155     ///
 156     /// If the value specified is `None`, then `write` calls will block
 157     /// indefinitely. It is an error to pass the zero `Duration` to this
 158     /// method.
 159     #[unstable(feature = "socket_timeout", reason = "RFC 1047 - recently added")]
 160     pub fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
 161         self.0.set_write_timeout(dur)
 162     }
 163
 164     /// Returns the read timeout of this socket.
 165     ///
 166     /// If the timeout is `None`, then `read` calls will block indefinitely.
 167     ///
 168     /// # Note
 169     ///
 170     /// Some platforms do not provide access to the current timeout.
 171     #[unstable(feature = "socket_timeout", reason = "RFC 1047 - recently added")]
 172     pub fn read_timeout(&self) -> io::Result<Option<Duration>> {
 173         self.0.read_timeout()
 174     }
 175
 176     /// Returns the write timeout of this socket.
 177     ///
 178     /// If the timeout is `None`, then `write` calls will block indefinitely.
 179     ///
 180     /// # Note
 181     ///
 182     /// Some platforms do not provide access to the current timeout.
 183     #[unstable(feature = "socket_timeout", reason = "RFC 1047 - recently added")]
 184     pub fn write_timeout(&self) -> io::Result<Option<Duration>> {
 185         self.0.write_timeout()
 186     }
 187 }
 188
 189 #[stable(feature = "rust1", since = "1.0.0")]
 190 impl Read for TcpStream {
 191     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> { self.0.read(buf) }
 192 }
 193 #[stable(feature = "rust1", since = "1.0.0")]
 194 impl Write for TcpStream {
 195     fn write(&mut self, buf: &[u8]) -> io::Result<usize> { self.0.write(buf) }
 196     fn flush(&mut self) -> io::Result<()> { Ok(()) }
 197 }
 198 #[stable(feature = "rust1", since = "1.0.0")]
 199 impl<'a> Read for &'a TcpStream {
 200     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> { self.0.read(buf) }
 201 }
 202 #[stable(feature = "rust1", since = "1.0.0")]
 203 impl<'a> Write for &'a TcpStream {
 204     fn write(&mut self, buf: &[u8]) -> io::Result<usize> { self.0.write(buf) }
 205     fn flush(&mut self) -> io::Result<()> { Ok(()) }
 206 }
 207
 208 impl AsInner<net_imp::TcpStream> for TcpStream {
 209     fn as_inner(&self) -> &net_imp::TcpStream { &self.0 }
 210 }
 211
 212 impl FromInner<net_imp::TcpStream> for TcpStream {
 213     fn from_inner(inner: net_imp::TcpStream) -> TcpStream { TcpStream(inner) }
 214 }
 215
 216 impl fmt::Debug for TcpStream {
 217     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
 218         self.0.fmt(f)
 219     }
 220 }
 221
 222 impl TcpListener {
 223     /// Creates a new `TcpListener` which will be bound to the specified
 224     /// address.
 225     ///
 226     /// The returned listener is ready for accepting connections.
 227     ///
 228     /// Binding with a port number of 0 will request that the OS assigns a port
 229     /// to this listener. The port allocated can be queried via the
 230     /// `socket_addr` function.
 231     ///
 232     /// The address type can be any implementer of `ToSocketAddrs` trait. See
 233     /// its documentation for concrete examples.
 234     #[stable(feature = "rust1", since = "1.0.0")]
 235     pub fn bind<A: ToSocketAddrs>(addr: A) -> io::Result<TcpListener> {
 236         super::each_addr(addr, net_imp::TcpListener::bind).map(TcpListener)
 237     }
 238
 239     /// Returns the local socket address of this listener.
 240     #[stable(feature = "rust1", since = "1.0.0")]
 241     pub fn local_addr(&self) -> io::Result<SocketAddr> {
 242         self.0.socket_addr()
 243     }
 244
 245     /// Creates a new independently owned handle to the underlying socket.
 246     ///
 247     /// The returned `TcpListener` is a reference to the same socket that this
 248     /// object references. Both handles can be used to accept incoming
 249     /// connections and options set on one listener will affect the other.
 250     #[stable(feature = "rust1", since = "1.0.0")]
 251     pub fn try_clone(&self) -> io::Result<TcpListener> {
 252         self.0.duplicate().map(TcpListener)
 253     }
 254
 255     /// Accept a new incoming connection from this listener.
 256     ///
 257     /// This function will block the calling thread until a new TCP connection
 258     /// is established. When established, the corresponding `TcpStream` and the
 259     /// remote peer's address will be returned.
 260     #[stable(feature = "rust1", since = "1.0.0")]
 261     pub fn accept(&self) -> io::Result<(TcpStream, SocketAddr)> {
 262         self.0.accept().map(|(a, b)| (TcpStream(a), b))
 263     }
 264
 265     /// Returns an iterator over the connections being received on this
 266     /// listener.
 267     ///
 268     /// The returned iterator will never return `None` and will also not yield
 269     /// the peer's `SocketAddr` structure.
 270     #[stable(feature = "rust1", since = "1.0.0")]
 271     pub fn incoming(&self) -> Incoming {
 272         Incoming { listener: self }
 273     }
 274 }
 275
 276 #[stable(feature = "rust1", since = "1.0.0")]
 277 impl<'a> Iterator for Incoming<'a> {
 278     type Item = io::Result<TcpStream>;
 279     fn next(&mut self) -> Option<io::Result<TcpStream>> {
 280         Some(self.listener.accept().map(|p| p.0))
 281     }
 282 }
 283
 284 impl AsInner<net_imp::TcpListener> for TcpListener {
 285     fn as_inner(&self) -> &net_imp::TcpListener { &self.0 }
 286 }
 287
 288 impl FromInner<net_imp::TcpListener> for TcpListener {
 289     fn from_inner(inner: net_imp::TcpListener) -> TcpListener {
 290         TcpListener(inner)
 291     }
 292 }
 293
 294 impl fmt::Debug for TcpListener {
 295     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
 296         self.0.fmt(f)
 297     }
 298 }
 299
 300 #[cfg(test)]
 301 mod tests {
 302     use prelude::v1::*;
 303
 304     use io::ErrorKind;
 305     use io::prelude::*;
 306     use net::*;
 307     use net::test::{next_test_ip4, next_test_ip6};
 308     use sync::mpsc::channel;
 309     use sys_common::AsInner;
 310     use time::Duration;
 311     use thread;
 312
 313     fn each_ip(f: &mut FnMut(SocketAddr)) {
 314         f(next_test_ip4());
 315         f(next_test_ip6());
 316     }
 317
 318     macro_rules! t {
 319         ($e:expr) => {
 320             match $e {
 321                 Ok(t) => t,
 322                 Err(e) => panic!("received error for `{}`: {}", stringify!($e), e),
 323             }
 324         }
 325     }
 326
 327     #[test]
 328     fn bind_error() {
 329         match TcpListener::bind("1.1.1.1:9999") {
 330             Ok(..) => panic!(),
 331             Err(e) =>
 332                 assert_eq!(e.kind(), ErrorKind::AddrNotAvailable),
 333         }
 334     }
 335
 336     #[test]
 337     fn connect_error() {
 338         match TcpStream::connect("0.0.0.0:1") {
 339             Ok(..) => panic!(),
 340             Err(e) => assert!(e.kind() == ErrorKind::ConnectionRefused ||
 341                               e.kind() == ErrorKind::InvalidInput ||
 342                               e.kind() == ErrorKind::AddrInUse ||
 343                               e.kind() == ErrorKind::AddrNotAvailable,
 344                               "bad error: {} {:?}", e, e.kind()),
 345         }
 346     }
 347
 348     #[test]
 349     fn listen_localhost() {
 350         let socket_addr = next_test_ip4();
 351         let listener = t!(TcpListener::bind(&socket_addr));
 352
 353         let _t = thread::spawn(move || {
 354             let mut stream = t!(TcpStream::connect(&("localhost",
 355                                                      socket_addr.port())));
 356             t!(stream.write(&[144]));
 357         });
 358
 359         let mut stream = t!(listener.accept()).0;
 360         let mut buf = [0];
 361         t!(stream.read(&mut buf));
 362         assert!(buf[0] == 144);
 363     }
 364
 365     #[test]
 366     fn connect_ip4_loopback() {
 367         let addr = next_test_ip4();
 368         let acceptor = t!(TcpListener::bind(&addr));
 369
 370         let _t = thread::spawn(move|| {
 371             let mut stream = t!(TcpStream::connect(&("127.0.0.1", addr.port())));
 372             t!(stream.write(&[44]));
 373         });
 374
 375         let mut stream = t!(acceptor.accept()).0;
 376         let mut buf = [0];
 377         t!(stream.read(&mut buf));
 378         assert!(buf[0] == 44);
 379     }
 380
 381     #[test]
 382     fn connect_ip6_loopback() {
 383         let addr = next_test_ip6();
 384         let acceptor = t!(TcpListener::bind(&addr));
 385
 386         let _t = thread::spawn(move|| {
 387             let mut stream = t!(TcpStream::connect(&("::1", addr.port())));
 388             t!(stream.write(&[66]));
 389         });
 390
 391         let mut stream = t!(acceptor.accept()).0;
 392         let mut buf = [0];
 393         t!(stream.read(&mut buf));
 394         assert!(buf[0] == 66);
 395     }
 396
 397     #[test]
 398     fn smoke_test_ip6() {
 399         each_ip(&mut |addr| {
 400             let acceptor = t!(TcpListener::bind(&addr));
 401
 402             let (tx, rx) = channel();
 403             let _t = thread::spawn(move|| {
 404                 let mut stream = t!(TcpStream::connect(&addr));
 405                 t!(stream.write(&[99]));
 406                 tx.send(t!(stream.local_addr())).unwrap();
 407             });
 408
 409             let (mut stream, addr) = t!(acceptor.accept());
 410             let mut buf = [0];
 411             t!(stream.read(&mut buf));
 412             assert!(buf[0] == 99);
 413             assert_eq!(addr, t!(rx.recv()));
 414         })
 415     }
 416
 417     #[test]
 418     fn read_eof_ip4() {
 419         each_ip(&mut |addr| {
 420             let acceptor = t!(TcpListener::bind(&addr));
 421
 422             let _t = thread::spawn(move|| {
 423                 let _stream = t!(TcpStream::connect(&addr));
 424                 // Close
 425             });
 426
 427             let mut stream = t!(acceptor.accept()).0;
 428             let mut buf = [0];
 429             let nread = t!(stream.read(&mut buf));
 430             assert_eq!(nread, 0);
 431             let nread = t!(stream.read(&mut buf));
 432             assert_eq!(nread, 0);
 433         })
 434     }
 435
 436     #[test]
 437     fn write_close() {
 438         each_ip(&mut |addr| {
 439             let acceptor = t!(TcpListener::bind(&addr));
 440
 441             let (tx, rx) = channel();
 442             let _t = thread::spawn(move|| {
 443                 drop(t!(TcpStream::connect(&addr)));
 444                 tx.send(()).unwrap();
 445             });
 446
 447             let mut stream = t!(acceptor.accept()).0;
 448             rx.recv().unwrap();
 449             let buf = [0];
 450             match stream.write(&buf) {
 451                 Ok(..) => {}
 452                 Err(e) => {
 453                     assert!(e.kind() == ErrorKind::ConnectionReset ||
 454                             e.kind() == ErrorKind::BrokenPipe ||
 455                             e.kind() == ErrorKind::ConnectionAborted,
 456                             "unknown error: {}", e);
 457                 }
 458             }
 459         })
 460     }
 461
 462     #[test]
 463     fn multiple_connect_serial_ip4() {
 464         each_ip(&mut |addr| {
 465             let max = 10;
 466             let acceptor = t!(TcpListener::bind(&addr));
 467
 468             let _t = thread::spawn(move|| {
 469                 for _ in 0..max {
 470                     let mut stream = t!(TcpStream::connect(&addr));
 471                     t!(stream.write(&[99]));
 472                 }
 473             });
 474
 475             for stream in acceptor.incoming().take(max) {
 476                 let mut stream = t!(stream);
 477                 let mut buf = [0];
 478                 t!(stream.read(&mut buf));
 479                 assert_eq!(buf[0], 99);
 480             }
 481         })
 482     }
 483
 484     #[test]
 485     fn multiple_connect_interleaved_greedy_schedule() {
 486         const MAX: usize = 10;
 487         each_ip(&mut |addr| {
 488             let acceptor = t!(TcpListener::bind(&addr));
 489
 490             let _t = thread::spawn(move|| {
 491                 let acceptor = acceptor;
 492                 for (i, stream) in acceptor.incoming().enumerate().take(MAX) {
 493                     // Start another thread to handle the connection
 494                     let _t = thread::spawn(move|| {
 495                         let mut stream = t!(stream);
 496                         let mut buf = [0];
 497                         t!(stream.read(&mut buf));
 498                         assert!(buf[0] == i as u8);
 499                     });
 500                 }
 501             });
 502
 503             connect(0, addr);
 504         });
 505
 506         fn connect(i: usize, addr: SocketAddr) {
 507             if i == MAX { return }
 508
 509             let t = thread::spawn(move|| {
 510                 let mut stream = t!(TcpStream::connect(&addr));
 511                 // Connect again before writing
 512                 connect(i + 1, addr);
 513                 t!(stream.write(&[i as u8]));
 514             });
 515             t.join().ok().unwrap();
 516         }
 517     }
 518
 519     #[test]
 520     fn multiple_connect_interleaved_lazy_schedule_ip4() {
 521         const MAX: usize = 10;
 522         each_ip(&mut |addr| {
 523             let acceptor = t!(TcpListener::bind(&addr));
 524
 525             let _t = thread::spawn(move|| {
 526                 for stream in acceptor.incoming().take(MAX) {
 527                     // Start another thread to handle the connection
 528                     let _t = thread::spawn(move|| {
 529                         let mut stream = t!(stream);
 530                         let mut buf = [0];
 531                         t!(stream.read(&mut buf));
 532                         assert!(buf[0] == 99);
 533                     });
 534                 }
 535             });
 536
 537             connect(0, addr);
 538         });
 539
 540         fn connect(i: usize, addr: SocketAddr) {
 541             if i == MAX { return }
 542
 543             let t = thread::spawn(move|| {
 544                 let mut stream = t!(TcpStream::connect(&addr));
 545                 connect(i + 1, addr);
 546                 t!(stream.write(&[99]));
 547             });
 548             t.join().ok().unwrap();
 549         }
 550     }
 551
 552     #[test]
 553     fn socket_and_peer_name_ip4() {
 554         each_ip(&mut |addr| {
 555             let listener = t!(TcpListener::bind(&addr));
 556             let so_name = t!(listener.local_addr());
 557             assert_eq!(addr, so_name);
 558             let _t = thread::spawn(move|| {
 559                 t!(listener.accept());
 560             });
 561
 562             let stream = t!(TcpStream::connect(&addr));
 563             assert_eq!(addr, t!(stream.peer_addr()));
 564         })
 565     }
 566
 567     #[test]
 568     fn partial_read() {
 569         each_ip(&mut |addr| {
 570             let (tx, rx) = channel();
 571             let srv = t!(TcpListener::bind(&addr));
 572             let _t = thread::spawn(move|| {
 573                 let mut cl = t!(srv.accept()).0;
 574                 cl.write(&[10]).unwrap();
 575                 let mut b = [0];
 576                 t!(cl.read(&mut b));
 577                 tx.send(()).unwrap();
 578             });
 579
 580             let mut c = t!(TcpStream::connect(&addr));
 581             let mut b = [0; 10];
 582             assert_eq!(c.read(&mut b).unwrap(), 1);
 583             t!(c.write(&[1]));
 584             rx.recv().unwrap();
 585         })
 586     }
 587
 588     #[test]
 589     fn double_bind() {
 590         each_ip(&mut |addr| {
 591             let _listener = t!(TcpListener::bind(&addr));
 592             match TcpListener::bind(&addr) {
 593                 Ok(..) => panic!(),
 594                 Err(e) => {
 595                     assert!(e.kind() == ErrorKind::ConnectionRefused ||
 596                             e.kind() == ErrorKind::Other ||
 597                             e.kind() == ErrorKind::AddrInUse,
 598                             "unknown error: {} {:?}", e, e.kind());
 599                 }
 600             }
 601         })
 602     }
 603
 604     #[test]
 605     fn fast_rebind() {
 606         each_ip(&mut |addr| {
 607             let acceptor = t!(TcpListener::bind(&addr));
 608
 609             let _t = thread::spawn(move|| {
 610                 t!(TcpStream::connect(&addr));
 611             });
 612
 613             t!(acceptor.accept());
 614             drop(acceptor);
 615             t!(TcpListener::bind(&addr));
 616         });
 617     }
 618
 619     #[test]
 620     fn tcp_clone_smoke() {
 621         each_ip(&mut |addr| {
 622             let acceptor = t!(TcpListener::bind(&addr));
 623
 624             let _t = thread::spawn(move|| {
 625                 let mut s = t!(TcpStream::connect(&addr));
 626                 let mut buf = [0, 0];
 627                 assert_eq!(s.read(&mut buf).unwrap(), 1);
 628                 assert_eq!(buf[0], 1);
 629                 t!(s.write(&[2]));
 630             });
 631
 632             let mut s1 = t!(acceptor.accept()).0;
 633             let s2 = t!(s1.try_clone());
 634
 635             let (tx1, rx1) = channel();
 636             let (tx2, rx2) = channel();
 637             let _t = thread::spawn(move|| {
 638                 let mut s2 = s2;
 639                 rx1.recv().unwrap();
 640                 t!(s2.write(&[1]));
 641                 tx2.send(()).unwrap();
 642             });
 643             tx1.send(()).unwrap();
 644             let mut buf = [0, 0];
 645             assert_eq!(s1.read(&mut buf).unwrap(), 1);
 646             rx2.recv().unwrap();
 647         })
 648     }
 649
 650     #[test]
 651     fn tcp_clone_two_read() {
 652         each_ip(&mut |addr| {
 653             let acceptor = t!(TcpListener::bind(&addr));
 654             let (tx1, rx) = channel();
 655             let tx2 = tx1.clone();
 656
 657             let _t = thread::spawn(move|| {
 658                 let mut s = t!(TcpStream::connect(&addr));
 659                 t!(s.write(&[1]));
 660                 rx.recv().unwrap();
 661                 t!(s.write(&[2]));
 662                 rx.recv().unwrap();
 663             });
 664
 665             let mut s1 = t!(acceptor.accept()).0;
 666             let s2 = t!(s1.try_clone());
 667
 668             let (done, rx) = channel();
 669             let _t = thread::spawn(move|| {
 670                 let mut s2 = s2;
 671                 let mut buf = [0, 0];
 672                 t!(s2.read(&mut buf));
 673                 tx2.send(()).unwrap();
 674                 done.send(()).unwrap();
 675             });
 676             let mut buf = [0, 0];
 677             t!(s1.read(&mut buf));
 678             tx1.send(()).unwrap();
 679
 680             rx.recv().unwrap();
 681         })
 682     }
 683
 684     #[test]
 685     fn tcp_clone_two_write() {
 686         each_ip(&mut |addr| {
 687             let acceptor = t!(TcpListener::bind(&addr));
 688
 689             let _t = thread::spawn(move|| {
 690                 let mut s = t!(TcpStream::connect(&addr));
 691                 let mut buf = [0, 1];
 692                 t!(s.read(&mut buf));
 693                 t!(s.read(&mut buf));
 694             });
 695
 696             let mut s1 = t!(acceptor.accept()).0;
 697             let s2 = t!(s1.try_clone());
 698
 699             let (done, rx) = channel();
 700             let _t = thread::spawn(move|| {
 701                 let mut s2 = s2;
 702                 t!(s2.write(&[1]));
 703                 done.send(()).unwrap();
 704             });
 705             t!(s1.write(&[2]));
 706
 707             rx.recv().unwrap();
 708         })
 709     }
 710
 711     #[test]
 712     fn shutdown_smoke() {
 713         each_ip(&mut |addr| {
 714             let a = t!(TcpListener::bind(&addr));
 715             let _t = thread::spawn(move|| {
 716                 let mut c = t!(a.accept()).0;
 717                 let mut b = [0];
 718                 assert_eq!(c.read(&mut b).unwrap(), 0);
 719                 t!(c.write(&[1]));
 720             });
 721
 722             let mut s = t!(TcpStream::connect(&addr));
 723             t!(s.shutdown(Shutdown::Write));
 724             assert!(s.write(&[1]).is_err());
 725             let mut b = [0, 0];
 726             assert_eq!(t!(s.read(&mut b)), 1);
 727             assert_eq!(b[0], 1);
 728         })
 729     }
 730
 731     #[test]
 732     fn close_readwrite_smoke() {
 733         each_ip(&mut |addr| {
 734             let a = t!(TcpListener::bind(&addr));
 735             let (tx, rx) = channel::<()>();
 736             let _t = thread::spawn(move|| {
 737                 let _s = t!(a.accept());
 738                 let _ = rx.recv();
 739             });
 740
 741             let mut b = [0];
 742             let mut s = t!(TcpStream::connect(&addr));
 743             let mut s2 = t!(s.try_clone());
 744
 745             // closing should prevent reads/writes
 746             t!(s.shutdown(Shutdown::Write));
 747             assert!(s.write(&[0]).is_err());
 748             t!(s.shutdown(Shutdown::Read));
 749             assert_eq!(s.read(&mut b).unwrap(), 0);
 750
 751             // closing should affect previous handles
 752             assert!(s2.write(&[0]).is_err());
 753             assert_eq!(s2.read(&mut b).unwrap(), 0);
 754
 755             // closing should affect new handles
 756             let mut s3 = t!(s.try_clone());
 757             assert!(s3.write(&[0]).is_err());
 758             assert_eq!(s3.read(&mut b).unwrap(), 0);
 759
 760             // make sure these don't die
 761             let _ = s2.shutdown(Shutdown::Read);
 762             let _ = s2.shutdown(Shutdown::Write);
 763             let _ = s3.shutdown(Shutdown::Read);
 764             let _ = s3.shutdown(Shutdown::Write);
 765             drop(tx);
 766         })
 767     }
 768
 769     #[test]
 770     fn close_read_wakes_up() {
 771         each_ip(&mut |addr| {
 772             let a = t!(TcpListener::bind(&addr));
 773             let (tx1, rx) = channel::<()>();
 774             let _t = thread::spawn(move|| {
 775                 let _s = t!(a.accept());
 776                 let _ = rx.recv();
 777             });
 778
 779             let s = t!(TcpStream::connect(&addr));
 780             let s2 = t!(s.try_clone());
 781             let (tx, rx) = channel();
 782             let _t = thread::spawn(move|| {
 783                 let mut s2 = s2;
 784                 assert_eq!(t!(s2.read(&mut [0])), 0);
 785                 tx.send(()).unwrap();
 786             });
 787             // this should wake up the child thread
 788             t!(s.shutdown(Shutdown::Read));
 789
 790             // this test will never finish if the child doesn't wake up
 791             rx.recv().unwrap();
 792             drop(tx1);
 793         })
 794     }
 795
 796     #[test]
 797     fn clone_while_reading() {
 798         each_ip(&mut |addr| {
 799             let accept = t!(TcpListener::bind(&addr));
 800
 801             // Enqueue a thread to write to a socket
 802             let (tx, rx) = channel();
 803             let (txdone, rxdone) = channel();
 804             let txdone2 = txdone.clone();
 805             let _t = thread::spawn(move|| {
 806                 let mut tcp = t!(TcpStream::connect(&addr));
 807                 rx.recv().unwrap();
 808                 t!(tcp.write(&[0]));
 809                 txdone2.send(()).unwrap();
 810             });
 811
 812             // Spawn off a reading clone
 813             let tcp = t!(accept.accept()).0;
 814             let tcp2 = t!(tcp.try_clone());
 815             let txdone3 = txdone.clone();
 816             let _t = thread::spawn(move|| {
 817                 let mut tcp2 = tcp2;
 818                 t!(tcp2.read(&mut [0]));
 819                 txdone3.send(()).unwrap();
 820             });
 821
 822             // Try to ensure that the reading clone is indeed reading
 823             for _ in 0..50 {
 824                 thread::yield_now();
 825             }
 826
 827             // clone the handle again while it's reading, then let it finish the
 828             // read.
 829             let _ = t!(tcp.try_clone());
 830             tx.send(()).unwrap();
 831             rxdone.recv().unwrap();
 832             rxdone.recv().unwrap();
 833         })
 834     }
 835
 836     #[test]
 837     fn clone_accept_smoke() {
 838         each_ip(&mut |addr| {
 839             let a = t!(TcpListener::bind(&addr));
 840             let a2 = t!(a.try_clone());
 841
 842             let _t = thread::spawn(move|| {
 843                 let _ = TcpStream::connect(&addr);
 844             });
 845             let _t = thread::spawn(move|| {
 846                 let _ = TcpStream::connect(&addr);
 847             });
 848
 849             t!(a.accept());
 850             t!(a2.accept());
 851         })
 852     }
 853
 854     #[test]
 855     fn clone_accept_concurrent() {
 856         each_ip(&mut |addr| {
 857             let a = t!(TcpListener::bind(&addr));
 858             let a2 = t!(a.try_clone());
 859
 860             let (tx, rx) = channel();
 861             let tx2 = tx.clone();
 862
 863             let _t = thread::spawn(move|| {
 864                 tx.send(t!(a.accept())).unwrap();
 865             });
 866             let _t = thread::spawn(move|| {
 867                 tx2.send(t!(a2.accept())).unwrap();
 868             });
 869
 870             let _t = thread::spawn(move|| {
 871                 let _ = TcpStream::connect(&addr);
 872             });
 873             let _t = thread::spawn(move|| {
 874                 let _ = TcpStream::connect(&addr);
 875             });
 876
 877             rx.recv().unwrap();
 878             rx.recv().unwrap();
 879         })
 880     }
 881
 882     #[test]
 883     fn debug() {
 884         let name = if cfg!(windows) {"socket"} else {"fd"};
 885         let socket_addr = next_test_ip4();
 886
 887         let listener = t!(TcpListener::bind(&socket_addr));
 888         let listener_inner = listener.0.socket().as_inner();
 889         let compare = format!("TcpListener {{ addr: {:?}, {}: {:?} }}",
 890                               socket_addr, name, listener_inner);
 891         assert_eq!(format!("{:?}", listener), compare);
 892
 893         let stream = t!(TcpStream::connect(&("localhost",
 894                                                  socket_addr.port())));
 895         let stream_inner = stream.0.socket().as_inner();
 896         let compare = format!("TcpStream {{ addr: {:?}, \
 897                               peer: {:?}, {}: {:?} }}",
 898                               stream.local_addr().unwrap(),
 899                               stream.peer_addr().unwrap(),
 900                               name,
 901                               stream_inner);
 902         assert_eq!(format!("{:?}", stream), compare);
 903     }
 904
 905     // FIXME: re-enabled bitrig/openbsd tests once their socket timeout code
 906     //        no longer has rounding errors.
 907     #[cfg_attr(any(target_os = "bitrig", target_os = "openbsd"), ignore)]
 908     #[test]
 909     fn timeouts() {
 910         let addr = next_test_ip4();
 911         let listener = t!(TcpListener::bind(&addr));
 912
 913         let stream = t!(TcpStream::connect(&("localhost", addr.port())));
 914         let dur = Duration::new(15410, 0);
 915
 916         assert_eq!(None, t!(stream.read_timeout()));
 917
 918         t!(stream.set_read_timeout(Some(dur)));
 919         assert_eq!(Some(dur), t!(stream.read_timeout()));
 920
 921         assert_eq!(None, t!(stream.write_timeout()));
 922
 923         t!(stream.set_write_timeout(Some(dur)));
 924         assert_eq!(Some(dur), t!(stream.write_timeout()));
 925
 926         t!(stream.set_read_timeout(None));
 927         assert_eq!(None, t!(stream.read_timeout()));
 928
 929         t!(stream.set_write_timeout(None));
 930         assert_eq!(None, t!(stream.write_timeout()));
 931     }
 932
 933     #[test]
 934     fn test_read_timeout() {
 935         let addr = next_test_ip4();
 936         let listener = t!(TcpListener::bind(&addr));
 937
 938         let mut stream = t!(TcpStream::connect(&("localhost", addr.port())));
 939         t!(stream.set_read_timeout(Some(Duration::from_millis(1000))));
 940
 941         let mut buf = [0; 10];
 942         let wait = Duration::span(|| {
 943             let kind = stream.read(&mut buf).err().expect("expected error").kind();
 944             assert!(kind == ErrorKind::WouldBlock || kind == ErrorKind::TimedOut);
 945         });
 946         assert!(wait > Duration::from_millis(400));
 947         assert!(wait < Duration::from_millis(1600));
 948     }
 949
 950     #[test]
 951     fn test_read_with_timeout() {
 952         let addr = next_test_ip4();
 953         let listener = t!(TcpListener::bind(&addr));
 954
 955         let mut stream = t!(TcpStream::connect(&("localhost", addr.port())));
 956         t!(stream.set_read_timeout(Some(Duration::from_millis(1000))));
 957
 958         let mut other_end = t!(listener.accept()).0;
 959         t!(other_end.write_all(b"hello world"));
 960
 961         let mut buf = [0; 11];
 962         t!(stream.read(&mut buf));
 963         assert_eq!(b"hello world", &buf[..]);
 964
 965         let wait = Duration::span(|| {
 966             let kind = stream.read(&mut buf).err().expect("expected error").kind();
 967             assert!(kind == ErrorKind::WouldBlock || kind == ErrorKind::TimedOut);
 968         });
 969         assert!(wait > Duration::from_millis(400));
 970         assert!(wait < Duration::from_millis(1600));
 971     }
 972 }