Repository: skiptests/astraea
Pull request #92 (2021-11-01 16:45:18)

Correct the average bytes
原本的average bytes算法是“這一秒取得的資料量“，但對於吞吐量測試中，我們其實並不在意”某一秒”的狀況，反之，從開始到現在的平均量才是我們在意的。因此這隻PR調整了平均量的算法，此外也包含以下幾個修正：

1. 簡化`max`和`min`的算法
2. 簡化一部分的測試


Patch:
diff --git a/app/src/main/java/org/astraea/performance/Metrics.java b/app/src/main/java/org/astraea/performance/Metrics.java
index 1d091d1e..74e76f20 100644
--- a/app/src/main/java/org/astraea/performance/Metrics.java
+++ b/app/src/main/java/org/astraea/performance/Metrics.java
@@ -1,7 +1,11 @@
 package org.astraea.performance;
 
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+
 /** Used to record statistics. This is thread safe. */
 public class Metrics {
+  private final long startTime = System.currentTimeMillis();
   private double avgLatency;
   private long num;
   private long max;
@@ -18,50 +22,48 @@ public class Metrics {
 
   /** Simultaneously add latency and bytes. */
   public synchronized void put(long latency, long bytes) {
+    ++num;
     putLatency(latency);
     addBytes(bytes);
   }
   /** Add a new value to latency metric. */
-  public synchronized void putLatency(long latency) {
-    if (min > latency) min = latency;
-    if (max < latency) max = latency;
-    ++num;
+  private synchronized void putLatency(long latency) {
+    min = Math.min(min, latency);
+    max = Math.max(max, latency);
     avgLatency += (((double) latency) - avgLatency) / (double) num;
   }
   /** Add a new value to bytes. */
-  public synchronized void addBytes(long bytes) {
+  private synchronized void addBytes(long bytes) {
     this.bytes += bytes;
   }
 
-  /** Get the number of latency put. */
+  /** @return Get the number of latency put. */
   public synchronized long num() {
     return num;
   }
-  /** Get the maximum of latency put. */
+  /** @return Get the maximum of latency put. */
   public synchronized long max() {
     return max;
   }
-  /** Get the minimum of latency put. */
+  /** @return Get the minimum of latency put. */
   public synchronized long min() {
     return min;
   }
-  /** Get the average latency. */
+  /** @return Get the average latency. */
   public synchronized double avgLatency() {
     return avgLatency;
   }
-  /** Reset to 0 and returns the old value of bytes */
-  public synchronized long bytesThenReset() {
-    long tmp = this.bytes;
-    this.bytes = 0;
-    return tmp;
+
+  /** @return the average bytes (in second) */
+  public synchronized double avgBytes() {
+    var value = new BigDecimal(bytes);
+    var scale = new BigDecimal(1024 * 1024);
+    var time = BigDecimal.valueOf((double) (System.currentTimeMillis() - startTime) / 1000);
+    return value.divide(scale, 3, RoundingMode.UP).divide(time, 3, RoundingMode.UP).doubleValue();
   }
-  /** Set all attributes to default value */
-  public synchronized void reset() {
-    avgLatency = 0;
-    num = 0;
-    max = 0;
-    // 初始為最大的integer值
-    min = Long.MAX_VALUE;
-    bytes = 0;
+
+  /** @return total send/received bytes */
+  public synchronized long bytes() {
+    return bytes;
   }
 }
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index 340c4bb3..e0373445 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -156,6 +156,7 @@ public class Performance {
             .sender()
             .topic(param.topic)
             .value(payload)
+            .timestamp(start)
             .run()
             .whenComplete(
                 (m, e) -> metrics.put(System.currentTimeMillis() - start, payload.length));
diff --git a/app/src/main/java/org/astraea/performance/Tracker.java b/app/src/main/java/org/astraea/performance/Tracker.java
index 963451c5..8d98263d 100644
--- a/app/src/main/java/org/astraea/performance/Tracker.java
+++ b/app/src/main/java/org/astraea/performance/Tracker.java
@@ -18,26 +18,28 @@ public class Tracker implements ThreadPool.Executor {
   @Override
   public State execute() throws InterruptedException {
     long completed = 0;
-    long bytes = 0L;
+    double bytes = 0;
     long max = 0;
     long min = Long.MAX_VALUE;
 
     /* producer */
     for (Metrics data : producerData) {
       completed += data.num();
-      bytes += data.bytesThenReset();
-      if (max < data.max()) max = data.max();
-      if (min > data.min()) min = data.min();
+      bytes += data.avgBytes();
+      max = Math.max(max, data.max());
+      min = Math.min(min, data.min());
     }
     if (completed == 0) {
       Thread.sleep(1000);
       return State.RUNNING;
     }
     System.out.printf("producers完成度: %.2f%%%n", ((double) completed * 100.0 / (double) records));
-    System.out.printf("  輸出%.3fMB/second%n", ((double) bytes / (1 << 20)));
+    System.out.printf("  輸出%.3fMB/second%n", bytes);
     System.out.println("  發送max latency: " + max + "ms");
     System.out.println("  發送mim latency: " + min + "ms");
     for (int i = 0; i < producerData.size(); ++i) {
+      System.out.printf(
+          "  producer[%d]的發送average bytes: %.3fMB%n", i, producerData.get(i).avgBytes());
       System.out.printf(
           "  producer[%d]的發送average latency: %.3fms%n", i, producerData.get(i).avgLatency());
     }
@@ -48,15 +50,17 @@ public class Tracker implements ThreadPool.Executor {
     min = Long.MAX_VALUE;
     for (Metrics data : consumerData) {
       completed += data.num();
-      bytes += data.bytesThenReset();
-      if (max < data.max()) max = data.max();
-      if (min > data.min()) min = data.min();
+      bytes += data.avgBytes();
+      max = Math.max(max, data.max());
+      min = Math.min(min, data.min());
     }
     System.out.printf("consumer完成度: %.2f%%%n", ((double) completed * 100.0 / (double) records));
-    System.out.printf("  輸入%.3fMB/second%n", ((double) bytes / (1 << 20)));
+    System.out.printf("  輸入%.3fMB/second%n", bytes);
     System.out.println("  端到端max latency: " + max + "ms");
     System.out.println("  端到端mim latency: " + min + "ms");
     for (int i = 0; i < consumerData.size(); ++i) {
+      System.out.printf(
+          "  consumer[%d]的端到端average bytes: %.3fMB%n", i, consumerData.get(i).avgBytes());
       System.out.printf(
           "  consumer[%d]的端到端average latency: %.3fms%n", i, consumerData.get(i).avgLatency());
     }


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 940b4164..103f1769 100644
--- a/app/src/test/java/org/astraea/performance/MetricsTest.java
+++ b/app/src/test/java/org/astraea/performance/MetricsTest.java
@@ -1,13 +1,11 @@
 package org.astraea.performance;
 
 import java.util.Random;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.LongAdder;
-import org.astraea.concurrent.ThreadPool;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 public class MetricsTest {
+
   @Test
   void testAverage() {
     Random rand = new Random();
@@ -20,55 +18,18 @@ public class MetricsTest {
     for (int i = 0; i < num; ++i) {
       long next = rand.nextInt();
       avg += ((double) next - avg) / (i + 1);
-      metrics.putLatency(next);
+      metrics.put(next, 0);
     }
 
     Assertions.assertEquals(avg, metrics.avgLatency());
   }
 
-  // Simultaneously add and get
   @Test
   void testBytes() {
-    final Metrics metrics = new Metrics();
-    final LongAdder longAdder = new LongAdder();
-    final AtomicInteger adderCount = new AtomicInteger(0);
-    final AtomicInteger getterCount = new AtomicInteger(0);
-    final long input = 100;
-    final int loopCount = 10000;
-
-    try (ThreadPool threadPool =
-        ThreadPool.builder()
-            .executor(
-                () -> {
-                  metrics.addBytes(input);
-                  if (adderCount.incrementAndGet() < loopCount)
-                    return ThreadPool.Executor.State.RUNNING;
-                  else return ThreadPool.Executor.State.DONE;
-                })
-            .executor(
-                () -> {
-                  longAdder.add(metrics.bytesThenReset());
-                  if (getterCount.incrementAndGet() < loopCount)
-                    return ThreadPool.Executor.State.RUNNING;
-                  else return ThreadPool.Executor.State.DONE;
-                })
-            .build()) {
-      threadPool.waitAll();
-    }
-    longAdder.add(metrics.bytesThenReset());
-
-    Assertions.assertEquals(loopCount * input, longAdder.sum());
-  }
+    var metrics = new Metrics();
 
-  @Test
-  public void testReset() {
-    final Metrics metrics = new Metrics();
-    metrics.addBytes(10);
-    metrics.putLatency(11);
-
-    metrics.reset();
-
-    Assertions.assertEquals(0, metrics.bytesThenReset());
-    Assertions.assertEquals(0, metrics.avgLatency());
+    Assertions.assertEquals(0, metrics.bytes());
+    metrics.put(0, 1000);
+    Assertions.assertEquals(1000, metrics.bytes());
   }
 }
diff --git a/app/src/test/java/org/astraea/performance/PerformanceTest.java b/app/src/test/java/org/astraea/performance/PerformanceTest.java
index 8514dc9f..cc684cad 100644
--- a/app/src/test/java/org/astraea/performance/PerformanceTest.java
+++ b/app/src/test/java/org/astraea/performance/PerformanceTest.java
@@ -27,7 +27,7 @@ public class PerformanceTest extends RequireBrokerCluster {
       executor.execute();
 
       Utils.waitFor(() -> metrics.num() == 1);
-      Assertions.assertEquals(1024, metrics.bytesThenReset());
+      Assertions.assertEquals(1024, metrics.bytes());
     }
   }
 
@@ -43,7 +43,7 @@ public class PerformanceTest extends RequireBrokerCluster {
       executor.execute();
 
       Assertions.assertEquals(0, metrics.num());
-      Assertions.assertEquals(0, metrics.bytesThenReset());
+      Assertions.assertEquals(0, metrics.bytes());
 
       try (var producer = Producer.builder().brokers(bootstrapServers()).build()) {
         producer.sender().topic(topicName).value(new byte[1024]).run().toCompletableFuture().get();
@@ -51,7 +51,7 @@ public class PerformanceTest extends RequireBrokerCluster {
       executor.execute();
 
       Assertions.assertEquals(1, metrics.num());
-      Assertions.assertNotEquals(1024, metrics.bytesThenReset());
+      Assertions.assertNotEquals(1024, metrics.bytes());
     }
   }
 }