Repository: skiptests/astraea
Pull request #254 (2022-02-24 09:31:02)

Throttle produce rate
#249 

在`performance.Manager`階段，限制producers的打資料速度。

例如 `--throughput 500KiB` 會限制**每秒**輸出量 "< 500KiB+record.size"。

garyparrot: 我做了一些測量，下面是結果，感覺很棒，讚讚

![image](https://user-images.githubusercontent.com/39105714/155884169-24667497-6869-4882-9107-a2e320e0c80d.png)

![image](https://user-images.githubusercontent.com/39105714/155884227-b211bab6-f531-44b6-82c4-e696f06a8051.png)

![image](https://user-images.githubusercontent.com/39105714/155884267-c267c569-bae3-4733-9fd7-8cc14c0a4c7f.png)

![image](https://user-images.githubusercontent.com/39105714/155884281-9521e9c4-5dc4-4c65-86fb-f56ed2c41b9e.png)

資料量是從 broker 的 log size 撈的
chia7712: @chinghongfang 不好意思，一樣麻煩修復一下衝突

Patch:
diff --git a/README.md b/README.md
index 1e97b1c1..b1913727 100644
--- a/README.md
+++ b/README.md
@@ -252,11 +252,12 @@ Run the benchmark from source
 10. --prop.file: the path to property file.
 11. --partitioner: the partitioner to use in producers
 12. --jmx.servers: the jmx server addresses of the brokers 
-13. --key.distribution: name of the distribution on key. Available distribution names: "uniform", "zipfian", "latest", "fixed". Default: (No key)
-14. --size.distribution: name of the distribution on value size. Available distribution names: "uniform", "zipfian", "latest", "fixed". Default: "uniform"
-15. --specify.broker: list of broker IDs to produce records to. Default: (Do Not Specify)
-16. --report.path: A path to place the report file. Default: (no report)
-17. --report.format: Select output file format. Available format: "csv", "json".
+13--throughput: the produce rate for all producers. e.g. "--throughput 2MiB". Default: 500 GiB (per second)
+14--key.distribution: name of the distribution on key. Available distribution names: "uniform", "zipfian", "latest", "fixed". Default: (No key)
+15--size.distribution: name of the distribution on value size. Available distribution names: "uniform", "zipfian", "latest", "fixed". Default: "uniform"
+16--specify.broker: list of broker IDs to produce records to. Default: (Do Not Specify)
+17--report.path: A path to place the report file. Default: (no report)
+18--report.format: Select output file format. Available format: "csv", "json".
 
 ---
 
diff --git a/app/src/main/java/org/astraea/performance/Manager.java b/app/src/main/java/org/astraea/performance/Manager.java
index d544e56a..c5648f79 100644
--- a/app/src/main/java/org/astraea/performance/Manager.java
+++ b/app/src/main/java/org/astraea/performance/Manager.java
@@ -8,6 +8,7 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
+import org.astraea.Utils;
 import org.astraea.utils.DataSize;
 import org.astraea.utils.DataUnit;
 
@@ -26,6 +27,9 @@ public class Manager {
   private final AtomicLong payloadNum = new AtomicLong(0);
   private final Supplier<Long> keyDistribution;
   private final RandomContent randomContent;
+  private long intervalStart = System.currentTimeMillis();
+  private long payloadBytes = 0L;
+  private final DataSize throughput;
 
   /**
    * Used to manage producing/consuming.
@@ -52,6 +56,7 @@ public class Manager {
     this.producerMetrics = producerMetrics;
     this.consumerMetrics = consumerMetrics;
     this.exeTime = argument.exeTime;
+    this.throughput = argument.throughput;
     this.keyDistribution = argument.keyDistributionType.create(100000);
     this.randomContent =
         new RandomContent(
@@ -61,7 +66,8 @@ public class Manager {
   }
 
   /**
-   * Generate random byte array in random/fixed length.
+   * Generate random byte array in random/fixed length. Warning: This method will block when the
+   * throughput (content generating rate) is higher than the given number (argument.throughput).
    *
    * @return random byte array. When no payload should be generated, Optional.empty() is generated.
    *     Whether the payload should be generated is determined in construct time. "Number of
@@ -72,9 +78,24 @@ public class Manager {
         >= 100D) return Optional.empty();
 
     var payload = randomContent.getContent();
+
+    Utils.waitFor(() -> checkAndAdd(payload.length));
     return Optional.of(payload);
   }
 
+  synchronized boolean checkAndAdd(int payloadLength) {
+    if (System.currentTimeMillis() - intervalStart > 1000) {
+      intervalStart = System.currentTimeMillis();
+      payloadBytes = payloadLength;
+      return true;
+    } else if (payloadBytes < throughput.measurement(DataUnit.Byte).longValue()) {
+      payloadBytes += payloadLength;
+      return true;
+    } else {
+      return false;
+    }
+  }
+
   public long producedRecords() {
     return producerMetrics.stream().mapToLong(Metrics::num).sum();
   }
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index 0a4b7ceb..e99671f8 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -387,6 +387,12 @@ public class Performance {
         validateWith = NonEmptyStringField.class)
     List<Integer> specifyBroker = List.of(-1);
 
+    @Parameter(
+        names = {"--throughput"},
+        description = "dataSize: size output per second. e.g. \"500KiB\"",
+        converter = DataSize.Field.class)
+    DataSize throughput = DataUnit.GiB.of(500);
+
     @Parameter(
         names = {"--report.path"},
         description = "String: A path to place the report. Default: (no report)",
diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties
index 7086e5cb..50f66897 100644
--- a/gradle/wrapper/gradle-wrapper.properties
+++ b/gradle/wrapper/gradle-wrapper.properties
@@ -1,6 +1,6 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 # please update the gradle version in gradlew too
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.4.1-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.4.2-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
diff --git a/gradlew b/gradlew
index 5abce60e..24bdac75 100755
--- a/gradlew
+++ b/gradlew
@@ -116,7 +116,7 @@ esac
 
 # gradle-wrapper.jar is not in git repo anymore, so we have to download it now.
 if [ ! -e "$APP_HOME/gradle/wrapper/gradle-wrapper.jar" ]; then
-  if ! curl -s -S --retry 3 -L -o "$APP_HOME/gradle/wrapper/gradle-wrapper.jar" "https://raw.githubusercontent.com/gradle/gradle/v7.4.1/gradle/wrapper/gradle-wrapper.jar"; then
+  if ! curl -s -S --retry 3 -L -o "$APP_HOME/gradle/wrapper/gradle-wrapper.jar" "https://raw.githubusercontent.com/gradle/gradle/v7.4.2/gradle/wrapper/gradle-wrapper.jar"; then
     rm -f "$APP_HOME/gradle/wrapper/gradle-wrapper.jar"
     echo "failed to download gradle-wrapper.jar"
     exit 1


Test patch:
diff --git a/app/src/test/java/org/astraea/performance/ManagerTest.java b/app/src/test/java/org/astraea/performance/ManagerTest.java
index 050795a1..7cc1eb7e 100644
--- a/app/src/test/java/org/astraea/performance/ManagerTest.java
+++ b/app/src/test/java/org/astraea/performance/ManagerTest.java
@@ -126,4 +126,21 @@ public class ManagerTest {
     manager.producerClosed();
     Assertions.assertTrue(manager.producedDone());
   }
+
+  @Test
+  void testCheckAndAdd() throws InterruptedException {
+    var argument = new Performance.Argument();
+    argument.throughput = DataUnit.KiB.of(3);
+    var recordSize = DataUnit.KiB.of(1).measurement(DataUnit.Byte).intValue();
+    var manager = new Manager(argument, List.of(), List.of());
+
+    Assertions.assertTrue(manager.checkAndAdd(recordSize));
+    Assertions.assertTrue(manager.checkAndAdd(recordSize));
+    Assertions.assertTrue(manager.checkAndAdd(recordSize));
+    Assertions.assertFalse(manager.checkAndAdd(recordSize));
+
+    Thread.sleep(1001);
+
+    Assertions.assertTrue(manager.checkAndAdd(recordSize));
+  }
 }