src/test/run-pass/tcp-accept-stress.rs

   1 // Copyright 2014 The Rust Project Developers. See the COPYRIGHT
   2 // file at the top-level directory of this distribution and at
   3 // http://rust-lang.org/COPYRIGHT.
   4 //
   5 // Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
   6 // http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
   7 // <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
   8 // option. This file may not be copied, modified, or distributed
   9 // except according to those terms.
  10
  11 // ignore-macos osx really doesn't like cycling through large numbers of
  12 //              sockets as calls to connect() will start returning EADDRNOTAVAIL
  13 //              quite quickly and it takes a few seconds for the sockets to get
  14 //              recycled.
  15
  16 use std::io::{TcpListener, Listener, Acceptor, EndOfFile, TcpStream};
  17 use std::sync::Arc;
  18 use std::sync::atomic::{AtomicUint, Ordering};
  19 use std::sync::mpsc::channel;
  20 use std::thread::Thread;
  21
  22 static N: uint = 8;
  23 static M: uint = 20;
  24
  25 fn main() {
  26     test();
  27 }
  28
  29 fn test() {
  30     let mut l = TcpListener::bind("127.0.0.1:0").unwrap();
  31     let addr = l.socket_name().unwrap();
  32     let mut a = l.listen().unwrap();
  33     let cnt = Arc::new(AtomicUint::new(0));
  34
  35     let (srv_tx, srv_rx) = channel();
  36     let (cli_tx, cli_rx) = channel();
  37     for _ in range(0, N) {
  38         let a = a.clone();
  39         let cnt = cnt.clone();
  40         let srv_tx = srv_tx.clone();
  41         Thread::spawn(move|| {
  42             let mut a = a;
  43             loop {
  44                 match a.accept() {
  45                     Ok(..) => {
  46                         if cnt.fetch_add(1, Ordering::SeqCst) == N * M - 1 {
  47                             break
  48                         }
  49                     }
  50                     Err(ref e) if e.kind == EndOfFile => break,
  51                     Err(e) => panic!("{}", e),
  52                 }
  53             }
  54             srv_tx.send(());
  55         }).detach();
  56     }
  57
  58     for _ in range(0, N) {
  59         let cli_tx = cli_tx.clone();
  60         Thread::spawn(move|| {
  61             for _ in range(0, M) {
  62                 let _s = TcpStream::connect(addr).unwrap();
  63             }
  64             cli_tx.send(());
  65         }).detach();
  66     }
  67     drop((cli_tx, srv_tx));
  68
  69     // wait for senders
  70     if cli_rx.iter().take(N).count() != N {
  71         a.close_accept().unwrap();
  72         panic!("clients panicked");
  73     }
  74
  75     // wait for one acceptor to die
  76     let _ = srv_rx.recv();
  77
  78     // Notify other receivers should die
  79     a.close_accept().unwrap();
  80
  81     // wait for receivers
  82     assert_eq!(srv_rx.iter().take(N - 1).count(), N - 1);
  83
  84     // Everything should have been accepted.
  85     assert_eq!(cnt.load(Ordering::SeqCst), N * M);
  86 }