Start metrics service before consumers and publishers

This way it outputs results even if consumers and publishers
take time to start.

Author: acogoluegnes

src/main/java/com/rabbitmq/perf/MulticastSet.java

@@ -236,6 +236,8 @@ public void run(boolean announceStartup)
 
             createProducers(announceStartup, producerStates, producerConnections);
 
+            this.performanceMetrics.start();
+
             startConsumers(consumerRunnables);
             startProducers(producerStates);
 
@@ -300,7 +302,6 @@ public void run(boolean announceStartup)
                 shutdownSequence = () -> { };
             }
 
-            this.performanceMetrics.start();
             this.completionHandler.waitForCompletion();
 
             try {
@@ -432,7 +433,9 @@ private void startConsumers(Runnable[] consumerRunnables) throws InterruptedExce
         if (this.params.getConsumerStartDelay().getSeconds() <= 0) {
             this.consumerLatencyIndicator.start();
             for (Runnable runnable : consumerRunnables) {
+                LOGGER.debug("Starting consumer runnable...");
                 runnable.run();
+                LOGGER.debug("Consumer runnable started");
                 if (params.getConsumerSlowStart()) {
                     System.out.println("Delaying start by 1 second because -S/--slow-start was requested");
                     Thread.sleep(1000);

src/main/java/com/rabbitmq/perf/metrics/DefaultPerformanceMetrics.java

@@ -73,7 +73,7 @@ public final class DefaultPerformanceMetrics implements PerformanceMetrics, Auto
   private final Duration interval;
   private final TimeUnit latencyCollectionTimeUnit;
   private final AtomicBoolean firstReport = new AtomicBoolean(false);
-  private final AtomicBoolean closed = new AtomicBoolean(false);
+  private final AtomicBoolean started = new AtomicBoolean(false);
   private final AtomicReference<Histogram> consumedLatency, confirmedLatency;
   private final MetricsFormatter formatter;
 
@@ -162,27 +162,29 @@ private static Runnable wrapInCatch(Runnable runnable) {
 
   @Override
   public void start() {
-    startTime.set(System.nanoTime());
-    lastTick.set(startTime.get());
-    startTimeForTotal.set(startTime.get());
-
-    scheduledExecutorService.scheduleAtFixedRate(wrapInCatch(() -> {
-      if (this.closed.get()) {
-        return;
-      }
-      if (noActivity()) {
-        this.publishedRate.accumulate(0);
-        this.confirmedRate.accumulate(0);
-        this.nackedRate.accumulate(0);
-        this.returnedRate.accumulate(0);
-        this.receivedRate.accumulate(0);
-        this.confirmedLatency.set(histogram());
-        this.consumedLatency.set(histogram());
-      } else {
-        metrics(System.nanoTime());
-      }
-
-    }), interval.getSeconds(), interval.getSeconds(), TimeUnit.SECONDS);
+    if (this.started.compareAndSet(false, true)) {
+      startTime.set(System.nanoTime());
+      lastTick.set(startTime.get());
+      startTimeForTotal.set(startTime.get());
+
+      scheduledExecutorService.scheduleAtFixedRate(wrapInCatch(() -> {
+        if (!this.started.get()) {
+          return;
+        }
+        if (noActivity()) {
+          this.publishedRate.accumulate(0);
+          this.confirmedRate.accumulate(0);
+          this.nackedRate.accumulate(0);
+          this.returnedRate.accumulate(0);
+          this.receivedRate.accumulate(0);
+          this.confirmedLatency.set(histogram());
+          this.consumedLatency.set(histogram());
+        } else {
+          metrics(System.nanoTime());
+        }
+
+      }), interval.getSeconds(), interval.getSeconds(), TimeUnit.SECONDS);
+    }
   }
 
   void metrics(long currentTime) {
@@ -211,7 +213,7 @@ void metrics(long currentTime) {
     this.confirmedLatency.set(histogram());
     this.consumedLatency.set(histogram());
 
-    if (!this.closed.get()) {
+    if (this.started.get()) {
       if (this.firstReport.compareAndSet(false, true)) {
         this.formatter.header();
       }
@@ -320,9 +322,15 @@ public void resetGlobals() {
 
   @Override
   public void close() {
-    if (this.closed.compareAndSet(false, true)) {
+    if (this.started.compareAndSet(false, true)) {
       this.scheduledExecutorService.shutdownNow();
       printFinal();
     }
   }
+
+  // for testing
+  void started(boolean value) {
+    this.started.set(value);
+  }
+
 }

src/test/java/com/rabbitmq/perf/metrics/DefaultPerformanceMetricsTest.java

@@ -166,6 +166,7 @@ public void stats(TestConfiguration testConfiguration) {
 
   void execute(Configurator configurator) {
     DefaultPerformanceMetrics metrics = metrics(configurator);
+    metrics.started(true);
     metrics.metrics(System.nanoTime());
     this.defaultOutput = defaultConsoleOut.toString();
     this.compactOutput = compactConsoleOut.toString();