Repository: skiptests/astraea
Pull request #216 (2022-02-08 13:26:24)

Add new metric for current throughput
同時紀錄 "平均throughput" 和 "現在的每秒throughput"。

另外，把輸出檔案的throughput都改成 "現在的每秒throughput"，因讀取檔案的時候可以靠工具幫忙計算出平均量。console log則需要平均量，所以都保留且印出。

wycccccc: LGTM

Patch:
diff --git a/app/src/main/java/org/astraea/performance/FileWriter.java b/app/src/main/java/org/astraea/performance/FileWriter.java
index f59c00a8..f29e1756 100644
--- a/app/src/main/java/org/astraea/performance/FileWriter.java
+++ b/app/src/main/java/org/astraea/performance/FileWriter.java
@@ -6,6 +6,7 @@ import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.stream.IntStream;
 import org.astraea.concurrent.ThreadPool;
+import org.astraea.utils.DataUnit;
 
 public class FileWriter implements ThreadPool.Executor {
   private final Manager manager;
@@ -35,7 +36,7 @@ public class FileWriter implements ThreadPool.Executor {
     try {
       writer = new BufferedWriter(new java.io.FileWriter(CSVName));
       writer.write(
-          "Time \\ Name, Consumed/Produced, Output throughput (MiB/sec), Input throughput (MiB/sec), "
+          "Time \\ Name, Consumed/Produced, Output throughput (/sec), Input throughput (/sec), "
               + "Publish max latency (ms), Publish min latency (ms), "
               + "End-to-end max latency (ms), End-to-end min latency (ms)");
       IntStream.range(0, tracker.producerResult().bytes.size())
@@ -45,9 +46,9 @@ public class FileWriter implements ThreadPool.Executor {
                   writer.write(
                       ",Producer["
                           + i
-                          + "] average throughput (MB/sec), Producer["
+                          + "] current throughput (/sec), Producer["
                           + i
-                          + "] average publish latency (ms)");
+                          + "] current publish latency (ms)");
                 } catch (IOException ignore) {
                 }
               });
@@ -58,9 +59,9 @@ public class FileWriter implements ThreadPool.Executor {
                   writer.write(
                       ",Consumer["
                           + i
-                          + "] average throughput (MB/sec), Consumer["
+                          + "] current throughput (/sec), Consumer["
                           + i
-                          + "] average ene-to-end latency (ms)");
+                          + "] current ene-to-end latency (ms)");
                 } catch (IOException ignore) {
                 }
               });
@@ -89,16 +90,16 @@ public class FileWriter implements ThreadPool.Executor {
               + duration.toSecondsPart()
               + "s");
       writer.write(String.format(",%.2f%% / %.2f%%", consumerPercentage, producerPercentage));
-      writer.write("," + producerResult.averageBytes(duration));
-      writer.write("," + consumerResult.averageBytes(duration));
+      writer.write("," + DataUnit.Byte.of(producerResult.totalCurrentBytes()));
+      writer.write("," + DataUnit.Byte.of(consumerResult.totalCurrentBytes()));
       writer.write("," + producerResult.maxLatency + "," + producerResult.minLatency);
       writer.write("," + consumerResult.maxLatency + "," + consumerResult.minLatency);
       for (int i = 0; i < producerResult.bytes.size(); ++i) {
-        writer.write("," + Tracker.avg(duration, producerResult.bytes.get(i)));
+        writer.write("," + DataUnit.Byte.of(producerResult.currentBytes.get(i)));
         writer.write("," + producerResult.averageLatencies.get(i));
       }
       for (int i = 0; i < consumerResult.bytes.size(); ++i) {
-        writer.write("," + Tracker.avg(duration, consumerResult.bytes.get(i)));
+        writer.write("," + DataUnit.Byte.of(consumerResult.currentBytes.get(i)));
         writer.write("," + consumerResult.averageLatencies.get(i));
       }
       writer.newLine();
diff --git a/app/src/main/java/org/astraea/performance/Metrics.java b/app/src/main/java/org/astraea/performance/Metrics.java
index ec279c4f..90ebedbc 100644
--- a/app/src/main/java/org/astraea/performance/Metrics.java
+++ b/app/src/main/java/org/astraea/performance/Metrics.java
@@ -9,6 +9,7 @@ public class Metrics implements BiConsumer<Long, Long> {
   private long max;
   private long min;
   private long bytes;
+  private long currentBytes;
 
   public Metrics() {
     avgLatency = 0;
@@ -16,6 +17,7 @@ public class Metrics implements BiConsumer<Long, Long> {
     max = 0;
     min = Long.MAX_VALUE;
     bytes = 0;
+    currentBytes = 0;
   }
 
   /** Simultaneously add latency and bytes. */
@@ -33,6 +35,7 @@ public class Metrics implements BiConsumer<Long, Long> {
   }
   /** Add a new value to bytes. */
   private synchronized void addBytes(long bytes) {
+    this.currentBytes += bytes;
     this.bytes += bytes;
   }
 
@@ -57,4 +60,10 @@ public class Metrics implements BiConsumer<Long, Long> {
   public synchronized long bytes() {
     return bytes;
   }
+
+  public synchronized long clearAndGetCurrentBytes() {
+    var ans = currentBytes;
+    currentBytes = 0;
+    return ans;
+  }
 }
diff --git a/app/src/main/java/org/astraea/performance/Tracker.java b/app/src/main/java/org/astraea/performance/Tracker.java
index fbd25def..ab12bc1a 100644
--- a/app/src/main/java/org/astraea/performance/Tracker.java
+++ b/app/src/main/java/org/astraea/performance/Tracker.java
@@ -3,6 +3,7 @@ package org.astraea.performance;
 import java.time.Duration;
 import java.util.*;
 import org.astraea.concurrent.ThreadPool;
+import org.astraea.utils.DataUnit;
 
 /** Print out the given metrics. */
 public class Tracker implements ThreadPool.Executor {
@@ -60,12 +61,14 @@ public class Tracker implements ThreadPool.Executor {
             + duration.toSecondsPart()
             + "sec");
     System.out.printf("producers completion rate: %.2f%%%n", percentage);
-    System.out.printf("  Throughput: %.3fMB/second%n", result.averageBytes(duration));
+    System.out.printf("  average Throughput: %.3f MB/second%n", result.averageBytes(duration));
+    System.out.printf(
+        "  current throughput: %s/second%n", DataUnit.Byte.of(result.totalCurrentBytes()));
     System.out.println("  publish max latency: " + result.maxLatency + "ms");
     System.out.println("  publish mim latency: " + result.minLatency + "ms");
     for (int i = 0; i < result.bytes.size(); ++i) {
       System.out.printf(
-          "  producer[%d] average throughput: %.3fMB%n", i, avg(duration, result.bytes.get(i)));
+          "  producer[%d] average throughput: %.3f MB%n", i, avg(duration, result.bytes.get(i)));
       System.out.printf(
           "  producer[%d] average publish latency: %.3fms%n", i, result.averageLatencies.get(i));
     }
@@ -82,12 +85,14 @@ public class Tracker implements ThreadPool.Executor {
     // Print out percentage of (consumed records) and (produced records)
     var percentage = result.completedRecords * 100D / manager.producedRecords();
     System.out.printf("consumer completion rate: %.2f%%%n", percentage);
-    System.out.printf("  throughput: %.3fMB/second%n", result.averageBytes(duration));
+    System.out.printf("  average throughput: %.3f MB/second%n", result.averageBytes(duration));
+    System.out.printf(
+        "  current throughput: %s/second%n", DataUnit.Byte.of(result.totalCurrentBytes()));
     System.out.println("  end-to-end max latency: " + result.maxLatency + "ms");
     System.out.println("  end-to-end mim latency: " + result.minLatency + "ms");
     for (int i = 0; i < result.bytes.size(); ++i) {
       System.out.printf(
-          "  consumer[%d] average throughput: %.3fMB%n", i, avg(duration, result.bytes.get(i)));
+          "  consumer[%d] average throughput: %.3f MB%n", i, avg(duration, result.bytes.get(i)));
       System.out.printf(
           "  consumer[%d] average ene-to-end latency: %.3fms%n", i, result.averageLatencies.get(i));
     }
@@ -107,12 +112,14 @@ public class Tracker implements ThreadPool.Executor {
   private static Result result(List<Metrics> metrics) {
     var completed = 0;
     var bytes = new ArrayList<Long>();
+    var currentBytes = new ArrayList<Long>();
     var averageLatencies = new ArrayList<Double>();
     var max = 0L;
     var min = Long.MAX_VALUE;
     for (Metrics data : metrics) {
       completed += data.num();
       bytes.add(data.bytes());
+      currentBytes.add(data.clearAndGetCurrentBytes());
       averageLatencies.add(data.avgLatency());
       max = Math.max(max, data.max());
       min = Math.min(min, data.min());
@@ -120,6 +127,7 @@ public class Tracker implements ThreadPool.Executor {
     return new Result(
         completed,
         Collections.unmodifiableList(bytes),
+        Collections.unmodifiableList(currentBytes),
         Collections.unmodifiableList(averageLatencies),
         min,
         max);
@@ -128,6 +136,7 @@ public class Tracker implements ThreadPool.Executor {
   static class Result {
     public final long completedRecords;
     public final List<Long> bytes;
+    public final List<Long> currentBytes;
     public final List<Double> averageLatencies;
     public final long minLatency;
     public final long maxLatency;
@@ -135,11 +144,13 @@ public class Tracker implements ThreadPool.Executor {
     Result(
         long completedRecords,
         List<Long> bytes,
+        List<Long> currentBytes,
         List<Double> averageLatencies,
         long minLatency,
         long maxLatency) {
       this.completedRecords = completedRecords;
       this.bytes = bytes;
+      this.currentBytes = currentBytes;
       this.averageLatencies = averageLatencies;
       this.minLatency = minLatency;
       this.maxLatency = maxLatency;
@@ -152,5 +163,9 @@ public class Tracker implements ThreadPool.Executor {
     long totalBytes() {
       return bytes.stream().mapToLong(i -> i).sum();
     }
+
+    long totalCurrentBytes() {
+      return currentBytes.stream().mapToLong(i -> i).sum();
+    }
   }
 }


Test patch:
diff --git a/app/src/test/java/org/astraea/performance/MetricsTest.java b/app/src/test/java/org/astraea/performance/MetricsTest.java
index ac9799da..e2c23a78 100644
--- a/app/src/test/java/org/astraea/performance/MetricsTest.java
+++ b/app/src/test/java/org/astraea/performance/MetricsTest.java
@@ -32,4 +32,15 @@ public class MetricsTest {
     metrics.accept(0L, 1000L);
     Assertions.assertEquals(1000, metrics.bytes());
   }
+
+  @Test
+  void testCurrentBytes() {
+    var metrics = new Metrics();
+
+    Assertions.assertEquals(0, metrics.clearAndGetCurrentBytes());
+    metrics.accept(0L, 100L);
+    metrics.accept(0L, 101L);
+    Assertions.assertEquals(201, metrics.clearAndGetCurrentBytes());
+    Assertions.assertEquals(0, metrics.clearAndGetCurrentBytes());
+  }
 }