wip

Author: lcian

stresstest/config/example.yaml

@@ -1,30 +1,15 @@
-remote: http://localhost:18888
+remote: http://localhost:8888
 
-duration: 5s
+duration: 10s
 
 workloads:
-  - name: attachments
-    mode: throughput
-    organizations: 1000
+  - name: snapshots
+    mode: batch
+    concurrency: 1
     file_sizes:
-      p50: 50 KiB
-      p99: 200 KiB
+      p50: 5 KiB
+      p99: 20 KiB
     actions:
-      writes: 10
+      writes: 20000
       reads: 0
       deletes: 0
-  - name: profiling
-    concurrency: 8
-    file_sizes:
-      p50: 15 KiB
-      p99: 100 KiB
-  - name: replays
-    concurrency: 16
-    file_sizes:
-      p50: 25 KiB
-      p99: 400 KiB
-  - name: nodestore
-    concurrency: 32
-    file_sizes:
-      p50: 5 KiB
-      p99: 20 KiB

stresstest/src/stresstest.rs

@@ -126,8 +126,10 @@ impl Stresstest {
         bar.finish_and_clear();
 
         let mut total_metrics = WorkloadMetrics::default();
+        let mut max_elapsed = Duration::ZERO;
         let workloads = finished_tasks.into_iter().map(|task| {
-            let (workload, metrics) = task.unwrap();
+            let (workload, metrics, elapsed) = task.unwrap();
+            max_elapsed = max_elapsed.max(elapsed);
 
             println!();
             println!(
@@ -137,7 +139,7 @@ impl Stresstest {
                 workload.mode,
                 workload.concurrency.bold()
             );
-            print_metrics(&metrics, duration);
+            print_metrics(&metrics, elapsed);
 
             total_metrics.file_sizes.merge(&metrics.file_sizes).unwrap();
             total_metrics.bytes_written += metrics.bytes_written;
@@ -163,7 +165,7 @@ impl Stresstest {
         let workloads: Vec<_> = workloads.collect();
         println!();
         println!("{}", "## TOTALS".bold());
-        print_metrics(&total_metrics, duration);
+        print_metrics(&total_metrics, max_elapsed);
         println!();
 
         if !cleanup {
@@ -230,14 +232,15 @@ async fn run_workload(
     workload: Workload,
     duration: Duration,
     ttl: Option<Duration>,
-) -> (Workload, WorkloadMetrics) {
+) -> (Workload, WorkloadMetrics, Duration) {
     // In throughput mode, allow for a high concurrency value.
     let concurrency = match workload.mode {
         WorkloadMode::Weighted => workload.concurrency,
         WorkloadMode::Throughput => 100,
     };
 
     let semaphore = Arc::new(Semaphore::new(concurrency));
+    let start = Instant::now();
     let deadline = tokio::time::Instant::now() + duration;
 
     let workload = Arc::new(Mutex::new(workload));
@@ -330,21 +333,23 @@ async fn run_workload(
         }
     }
 
+    // by acquiring *all* the semaphores, we essentially wait for all outstanding tasks to finish
+    let _permits = semaphore.acquire_many(concurrency as u32).await;
+
     let metrics: WorkloadMetrics = {
         let mut metrics = metrics.lock().unwrap();
         std::mem::take(&mut metrics)
     };
 
-    // by acquiring *all* the semaphores, we essentially wait for all outstanding tasks to finish
-    let _permits = semaphore.acquire_many(concurrency as u32).await;
-
     let workload = Arc::try_unwrap(workload)
         .map_err(|_| ())
         .unwrap()
         .into_inner()
         .unwrap();
 
-    (workload, metrics)
+    let elapsed = start.elapsed();
+
+    (workload, metrics, elapsed)
 }
 
 fn print_error(message: &str, error: &Error) {