ws load test: provide stats on number of active connections

aquatic_ws_load_test/src/common.rs

@@ -13,6 +13,7 @@ pub struct Statistics {
     pub responses_answer: AtomicUsize,
     pub responses_scrape: AtomicUsize,
     pub responses_error: AtomicUsize,
+    pub connections: AtomicUsize,
 }
 
 #[derive(Clone)]

aquatic_ws_load_test/src/main.rs

@@ -146,6 +146,8 @@ fn monitor_statistics(state: LoadTestState, config: &Config) {
 
         let responses_announce_per_second = responses_announce / interval_f64;
 
+        let connections = statistics.connections.load(Ordering::Relaxed);
+
         let responses_per_second = responses_announce_per_second
             + responses_offer_per_second
             + responses_answer_per_second
@@ -165,6 +167,7 @@ fn monitor_statistics(state: LoadTestState, config: &Config) {
         println!("  - Answer responses:   {:.2}", responses_answer_per_second);
         println!("  - Scrape responses:   {:.2}", responses_scrape_per_second);
         println!("  - Error responses:   {:.2}", responses_error_per_second);
+        println!("Active connections: {}", connections);
 
         let time_elapsed = start_time.elapsed();
         let duration = Duration::from_secs(config.duration as u64);

aquatic_ws_load_test/src/network.rs

@@ -92,6 +92,8 @@ impl Connection {
         );
         let (stream, _) = client_async(request, stream).await?;
 
+        let statistics = load_test_state.statistics.clone();
+
         let mut connection = Connection {
             config,
             load_test_state,
@@ -103,12 +105,14 @@ impl Connection {
         };
 
         *num_active_connections.borrow_mut() += 1;
+        statistics.connections.fetch_add(1, Ordering::Relaxed);
 
         if let Err(err) = connection.run_connection_loop().await {
             ::log::info!("connection error: {:#}", err);
         }
 
         *num_active_connections.borrow_mut() -= 1;
+        statistics.connections.fetch_sub(1, Ordering::Relaxed);
 
         Ok(())
     }