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 #![feature(phase)]
  17
  18 #[phase(plugin)]
  19 extern crate green;
  20 extern crate native;
  21
  22 use std::io::{TcpListener, Listener, Acceptor, EndOfFile, TcpStream};
  23 use std::sync::{atomic, Arc};
  24 use std::task::TaskBuilder;
  25 use native::NativeTaskBuilder;
  26
  27 static N: uint = 8;
  28 static M: uint = 20;
  29
  30 green_start!(main)
  31
  32 fn main() {
  33     test();
  34
  35     let (tx, rx) = channel();
  36     TaskBuilder::new().native().spawn(proc() {
  37         tx.send(test());
  38     });
  39     rx.recv();
  40 }
  41
  42 fn test() {
  43     let mut l = TcpListener::bind("127.0.0.1", 0).unwrap();
  44     let addr = l.socket_name().unwrap();
  45     let mut a = l.listen().unwrap();
  46     let cnt = Arc::new(atomic::AtomicUint::new(0));
  47
  48     let (srv_tx, srv_rx) = channel();
  49     let (cli_tx, cli_rx) = channel();
  50     for _ in range(0, N) {
  51         let a = a.clone();
  52         let cnt = cnt.clone();
  53         let srv_tx = srv_tx.clone();
  54         spawn(proc() {
  55             let mut a = a;
  56             loop {
  57                 match a.accept() {
  58                     Ok(..) => {
  59                         if cnt.fetch_add(1, atomic::SeqCst) == N * M - 1 {
  60                             break
  61                         }
  62                     }
  63                     Err(ref e) if e.kind == EndOfFile => break,
  64                     Err(e) => fail!("{}", e),
  65                 }
  66             }
  67             srv_tx.send(());
  68         });
  69     }
  70
  71     for _ in range(0, N) {
  72         let cli_tx = cli_tx.clone();
  73         spawn(proc() {
  74             for _ in range(0, M) {
  75                 let _s = TcpStream::connect(addr.ip.to_string().as_slice(),
  76                                             addr.port).unwrap();
  77             }
  78             cli_tx.send(());
  79         });
  80     }
  81     drop((cli_tx, srv_tx));
  82
  83     // wait for senders
  84     if cli_rx.iter().take(N).count() != N {
  85         a.close_accept().unwrap();
  86         fail!("clients failed");
  87     }
  88
  89     // wait for one acceptor to die
  90     let _ = srv_rx.recv();
  91
  92     // Notify other receivers should die
  93     a.close_accept().unwrap();
  94
  95     // wait for receivers
  96     assert_eq!(srv_rx.iter().take(N - 1).count(), N - 1);
  97
  98     // Everything should have been accepted.
  99     assert_eq!(cnt.load(atomic::SeqCst), N * M);
 100 }
 101