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