Use multiple producer threads to drive the load generation (openmessaging#2)

merlimat · web-flow · commit c5ad4f72c8b3 · 2017-11-30T15:05:54.000-08:00
diff --git a/benchmark-framework/src/main/java/io/openmessaging/benchmark/WorkloadGenerator.java b/benchmark-framework/src/main/java/io/openmessaging/benchmark/WorkloadGenerator.java
@@ -20,6 +20,7 @@
 
 import java.text.DecimalFormat;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Random;
 import java.util.concurrent.CompletableFuture;
@@ -35,6 +36,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.collect.Lists;
 import com.google.common.io.BaseEncoding;
 import com.google.common.util.concurrent.RateLimiter;
 
@@ -355,38 +357,45 @@ private TestResult generateLoad(List<BenchmarkProducer> producers, long testDura
         long startTime = System.nanoTime();
         this.testCompleted = false;
 
-        executor.submit(() -> {
-            try {
-                byte[] payloadData = new byte[workload.messageSize];
-
-                // Send messages on all topics/producers
-                while (!testCompleted) {
-                    for (int i = 0; i < producers.size(); i++) {
-                        BenchmarkProducer producer = producers.get(i);
-                        rateLimiter.acquire();
-
-                        final long sendTime = System.nanoTime();
-
-                        producer.sendAsync(payloadData).thenRun(() -> {
-                            messagesSent.increment();
-                            totalMessagesSent.increment();
-                            bytesSent.add(payloadData.length);
-
-                            long latencyMicros = TimeUnit.NANOSECONDS.toMicros(System.nanoTime() - sendTime);
-                            recorder.recordValue(latencyMicros);
-                            cumulativeRecorder.recordValue(latencyMicros);
-
-                        }).exceptionally(ex -> {
-                            log.warn("Write error on message", ex);
-                            System.exit(-1);
-                            return null;
-                        });
+        int processors = Runtime.getRuntime().availableProcessors();
+        Collections.shuffle(producers);
+
+        final byte[] payloadData = new byte[workload.messageSize];
+
+        // Divide the producers across multiple different threads
+        for (List<BenchmarkProducer> producersPerThread : Lists.partition(producers, processors)) {
+            executor.submit(() -> {
+                try {
+
+                    // Send messages on all topics/producers assigned to this thread
+                    while (!testCompleted) {
+                        for (int i = 0; i < producersPerThread.size(); i++) {
+                            BenchmarkProducer producer = producersPerThread.get(i);
+                            rateLimiter.acquire();
+
+                            final long sendTime = System.nanoTime();
+
+                            producer.sendAsync(payloadData).thenRun(() -> {
+                                messagesSent.increment();
+                                totalMessagesSent.increment();
+                                bytesSent.add(payloadData.length);
+
+                                long latencyMicros = TimeUnit.NANOSECONDS.toMicros(System.nanoTime() - sendTime);
+                                recorder.recordValue(latencyMicros);
+                                cumulativeRecorder.recordValue(latencyMicros);
+
+                            }).exceptionally(ex -> {
+                                log.warn("Write error on message", ex);
+                                System.exit(-1);
+                                return null;
+                            });
+                        }
                     }
+                } catch (Throwable t) {
+                    log.error("Got error", t);
                 }
-            } catch (Throwable t) {
-                log.error("Got error", t);
-            }
-        });
+            });
+        }
 
         // Print report stats
         long oldTime = System.nanoTime();
