Merge 1514505 into e4f78cf

Cargo.toml

@@ -15,7 +15,7 @@ Gauging performance of network services. Snapshot or continuous, supports Promet
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-clap = "2.33"
+clap = { version = "3.1.6", features = ["derive"] }
 base64 = "0.13"
 derive_builder = "0.9"
 log = "0.4"

src/bench_run.rs

@@ -9,9 +9,12 @@ use crate::rate_limiter::RateLimiter;
 /// except according to those terms.
 use async_trait::async_trait;
 use log::error;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::{Duration, Instant};
 use tokio::sync::mpsc::Sender;
 
+static STOP_ON_FATAL: AtomicBool = AtomicBool::new(false);
+
 #[derive(Clone, Debug)]
 pub struct BenchRun {
     pub index: usize,
@@ -89,20 +92,26 @@ impl BenchRun {
     ) -> Result<(), String> {
         let client = bench_protocol_adapter.build_client()?;
 
-        while self.has_more_work() {
+        while self.has_more_work() && !STOP_ON_FATAL.load(Ordering::Relaxed) {
             self.rate_limiter
                 .acquire_one()
                 .await
                 .expect("Unexpected LeakyBucket.acquire error");
 
             let request_stats = bench_protocol_adapter.send_request(&client).await;
+            let fatal_error = request_stats.fatal_error;
 
             metrics_channel
                 .try_send(request_stats)
                 .map_err(|e| {
                     error!("Error sending metrics: {}", e);
                 })
                 .unwrap_or_default();
+
+            if fatal_error {
+                STOP_ON_FATAL.store(true, Ordering::Relaxed);
+                break;
+            }
         }
 
         Ok(())