Merge pull request #6 from lsk569937453/dev

Add the report!

Cargo.toml

@@ -21,3 +21,4 @@ futures = "0.3.29"
 mlua = { version = "0.9.1", features = ["lua54", "vendored","async","macros"] }
 hyper-rustls = "0.26.0"
 rustls = { version = "0.22.1"  ,features = [ "logging" ]}
+itertools = "0.12.1"

src/main.rs

@@ -1,40 +1,54 @@
 use futures::{stream, StreamExt};
+use http_body_util::BodyExt;
 use http_body_util::Empty;
+use hyper::body::Incoming;
 use hyper::Request;
 use hyper_util::client::legacy::{connect::HttpConnector, Client};
+use output::report::ResponseStatistic;
+use output::report::StatisticList;
 use std::str::FromStr;
 use std::sync::atomic::AtomicI32;
 use std::sync::Arc;
 use tokio::signal::ctrl_c;
 use tokio::sync::Mutex;
+mod output;
 #[macro_use]
 extern crate anyhow;
 use clap::Parser;
 use http_body_util::Full;
 use hyper::body::Bytes;
+use hyper::header::HeaderValue;
+use hyper::header::CONTENT_LENGTH;
+use hyper::Response;
 use hyper_rustls::ConfigBuilderExt;
 use hyper_rustls::HttpsConnector;
 use hyper_util::rt::TokioExecutor;
 use rustls::RootCertStore;
 use std::env;
 use tokio::sync::mpsc;
 use tokio::task::JoinSet;
+use tokio::time::Instant;
 use tokio::time::{sleep, Duration};
-
 #[derive(Parser)]
 #[command(author, version, about, long_about)]
 struct Cli {
     /// The request url,like http://www.google.com
     url: String,
-    /// The thread count.
-    #[arg(short = 't', long, value_name = "Threads count", default_value_t = 20)]
+    /// Number of workers to run concurrently. Total number of requests cannot
+    ///  be smaller than the concurrency level. Default is 50..
+    #[arg(
+        short = 'c',
+        long,
+        value_name = "Number of workers",
+        default_value_t = 50
+    )]
     threads: u16,
-    /// The running seconds for the testing tools.
+    /// Duration of application to send requests. When duration is reached,application stops and exits.
     #[arg(
-        short = 's',
+        short = 'z',
         long,
-        value_name = "The running seconds",
-        default_value_t = 3
+        value_name = "Duration of application to send requests",
+        default_value_t = 5
     )]
     sleep_seconds: u64,
 }
@@ -59,59 +73,73 @@ async fn do_request(
         .enable_http1()
         .build();
 
-    let timer = tokio::time::Instant::now();
     let client = Client::builder(hyper_util::rt::TokioExecutor::new()).build(https.clone());
 
-    let counter = Arc::new(AtomicI32::new(0));
     let mut task_list = vec![];
+    let shared_list: Arc<Mutex<StatisticList>> = Arc::new(Mutex::new(StatisticList {
+        response_list: vec![],
+    }));
+    let now = Instant::now();
     for _ in 0..connections {
+        let cloned_list = shared_list.clone();
         let clone_url = url.clone();
-        let clone_counter = counter.clone();
         let clone_client = client.clone();
         let task =
-            tokio::spawn(async move { submit_task(clone_counter, clone_client, clone_url).await });
+            tokio::spawn(
+                async move { submit_task(cloned_list.clone(), clone_client, clone_url).await },
+            );
         task_list.push(task);
     }
     drop(client);
     let _ = sleep(Duration::from_secs(sleep_seconds)).await;
-
+    let total_cost = now.elapsed().as_millis();
     task_list.iter().for_each(|item| item.abort());
-    let success_count = counter.load(std::sync::atomic::Ordering::Relaxed).clone();
-
-    let time_cost: u128 = timer.elapsed().as_millis();
-
-    let base: i32 = 10;
-
-    let rps = base.pow(3) * success_count / (time_cost as i32);
-
-    println!(
-        "Actual time {:.2} million second, RPS {}/s,count is {}",
-        time_cost, rps, success_count
-    );
+    let list = shared_list.lock().await;
+    list.print(total_cost);
     Ok(())
 }
 async fn submit_task(
-    counter: Arc<AtomicI32>,
+    shared_list: Arc<Mutex<StatisticList>>,
     client: Client<HttpsConnector<HttpConnector>, Full<Bytes>>,
     url: String,
 ) {
     let clone_client = client.clone();
     let clone_url: String = url.clone();
     loop {
+        let now = Instant::now();
+
         let cloned_client1 = clone_client.clone();
         let clone_url1 = clone_url.parse::<hyper::Uri>().unwrap();
         let result = cloned_client1
             .get(clone_url1)
             .await
             .map_err(|e| anyhow!("Terst!"));
-
-        if let Ok(response) = result {
-            if response.status().is_success() {
-                tokio::spawn(statistic(counter.clone()));
-            }
+        let elapsed = now.elapsed().as_millis();
+        if let Ok(r) = result {
+            tokio::spawn(statistic(shared_list.clone(), elapsed, r));
         }
     }
 }
-async fn statistic(counter: Arc<AtomicI32>) {
-    counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+async fn statistic(
+    shared_list: Arc<Mutex<StatisticList>>,
+    time_cost: u128,
+    res: Response<Incoming>,
+) {
+    let default_content_length = HeaderValue::from_static("0");
+    let content_len_header = res
+        .headers()
+        .get(CONTENT_LENGTH)
+        .unwrap_or(&default_content_length);
+    let content_len = content_len_header
+        .to_str()
+        .unwrap_or("0")
+        .parse::<u64>()
+        .unwrap_or(0);
+    let mut list = shared_list.lock().await;
+    let response_statistic = ResponseStatistic {
+        time_cost: time_cost,
+        staus_code: res.status().as_u16(),
+        content_length: content_len,
+    };
+    list.response_list.push(response_statistic);
 }

src/output/mod.rs

@@ -0,0 +1 @@
+pub mod report;

src/output/report.rs

@@ -0,0 +1,105 @@
+use core::time;
+use itertools::Itertools;
+use std::collections::HashMap;
+
+pub struct StatisticList {
+    pub response_list: Vec<ResponseStatistic>,
+}
+pub struct ResponseStatistic {
+    pub time_cost: u128,
+    pub staus_code: u16,
+    pub content_length: u64,
+}
+impl StatisticList {
+    pub fn print(&self, total: u128) {
+        let mut slow = 0;
+        let mut fast = 10000000;
+        let mut average = 0;
+        let mut rps = 0;
+        let mut total_data = 0;
+        let mut size_per_request = 0;
+
+        let mut hashmap = HashMap::new();
+
+        let mut total_time_cost = 0;
+        for item in &self.response_list {
+            let time_cost = item.time_cost;
+            let status_code = item.staus_code;
+            let content_len = item.content_length;
+            if time_cost > slow {
+                slow = time_cost;
+            }
+            if time_cost < fast {
+                fast = time_cost;
+            }
+            total_time_cost += time_cost;
+            total_data += content_len;
+            size_per_request = content_len;
+            hashmap
+                .entry(status_code)
+                .and_modify(|counter| *counter += 1)
+                .or_insert(1);
+        }
+        let mapdata = hashmap
+            .iter()
+            .map(|(k, v)| format!("[{}] {} responses", k, v))
+            .join(", ");
+
+        average = total_time_cost / self.response_list.len() as u128;
+        rps = self.response_list.len() as u128 / (total / 1000);
+
+        let format_str = format!(
+            r#"
+Summary:
+    Total:          {total} millisecond 
+    Slowest:        {slow} millisecond 
+    Fastest:        {fast} millisecond 
+    Average:        {average} millisecond 
+    Requests/sec:   {rps}
+    Total data:     {total_data} bytes
+    Size/request:   {size_per_request} bytes
+        
+Status code distribution:
+    {mapdata}
+"#
+        );
+        println!("{}", format_str);
+    }
+}
+pub fn print() {
+    let x = 42;
+    let y = 123;
+
+    let s = format!(
+        r#"
+        Summary:
+          Total:	{{ formatNumber .Total.Seconds }} secs
+          Slowest:	{{ formatNumber .Slowest }} secs
+          Fastest:	{{ formatNumber .Fastest }} secs
+          Average:	{{ formatNumber .Average }} secs
+          Requests/sec:	{{ formatNumber .Rps }}
+          {{ if gt .SizeTotal 0 }}
+          Total data:	{{ .SizeTotal }} bytes
+          Size/request:	{{ .SizeReq }} bytes{{ end }}
+        
+        Response time histogram:
+        {{ histogram .Histogram }}
+        
+        Latency distribution:{{ range .LatencyDistribution }}
+          {{ .Percentage }}%% in {{ formatNumber .Latency }} secs{{ end }}
+        
+        Details (average, fastest, slowest):
+          DNS+dialup:	{{ formatNumber .AvgConn }} secs, {{ formatNumber .ConnMax }} secs, {{ formatNumber .ConnMin }} secs
+          DNS-lookup:	{{ formatNumber .AvgDNS }} secs, {{ formatNumber .DnsMax }} secs, {{ formatNumber .DnsMin }} secs
+          req write:	{{ formatNumber .AvgReq }} secs, {{ formatNumber .ReqMax }} secs, {{ formatNumber .ReqMin }} secs
+          resp wait:	{{ formatNumber .AvgDelay }} secs, {{ formatNumber .DelayMax }} secs, {{ formatNumber .DelayMin }} secs
+          resp read:	{{ formatNumber .AvgRes }} secs, {{ formatNumber .ResMax }} secs, {{ formatNumber .ResMin }} secs
+        
+        Status code distribution:{{ range $code, $num := .StatusCodeDist }}
+          [{{ $code }}]	{{ $num }} responses{{ end }}
+        
+        {{ if gt (len .ErrorDist) 0 }}Error distribution:{{ range $err, $num := .ErrorDist }}
+          [{{ $num }}]	{{ $err }}{{ end }}{{ end }}
+        "#
+    );
+}

src/report.rs

@@ -0,0 +1 @@
+