Merge pull request #228 from rabbitmq/rabbitmq-perf-test-222-reason-for-stopping

Print out reason(s) for stopping

Author: lukebakken

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

@@ -40,6 +40,7 @@ public class Consumer extends AgentBase implements Runnable {
 
     private static final AckNackOperation NACK_OPERATION =
             (ch, envelope, multiple) -> ch.basicNack(envelope.getDeliveryTag(), multiple, true);
+    static final String STOP_REASON_CONSUMER_REACHED_MESSAGE_LIMIT = "Consumer reached message limit";
 
     private volatile ConsumerImpl       q;
     private final Channel               channel;
@@ -250,7 +251,7 @@ void handleMessage(Envelope envelope, BasicProperties properties, byte[] body, C
                 }
             }
             if (msgLimit != 0 && currentMessageCount >= msgLimit) { // NB: not quite the inverse of above
-                countDown();
+                countDown(STOP_REASON_CONSUMER_REACHED_MESSAGE_LIMIT);
             }
         }
 
@@ -280,7 +281,7 @@ public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig
             );
             if (!recoveryProcess.isEnabled()) {
                 LOGGER.debug("Counting down for consumer");
-                countDown();
+                countDown("Consumer shut down");
             }
         }
 
@@ -297,9 +298,9 @@ public void handleCancel(String consumerTag) throws IOException {
         }
     }
 
-    private void countDown() {
+    private void countDown(String reason) {
         if (completed.compareAndSet(false, true)) {
-            completionHandler.countDown();
+            completionHandler.countDown(reason);
         }
     }
 

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

@@ -49,6 +49,7 @@ public class MulticastSet {
      */
     private static final int PUBLISHING_INTERVAL_NB_PRODUCERS_PER_THREAD = 50;
     private static final String PRODUCER_THREAD_PREFIX = "perf-test-producer-";
+    static final String STOP_REASON_REACHED_TIME_LIMIT = "Reached time limit";
     private final Stats stats;
     private final ConnectionFactory factory;
     private final MulticastParams params;
@@ -501,7 +502,7 @@ public interface CompletionHandler {
 
         void waitForCompletion() throws InterruptedException;
 
-        void countDown();
+        void countDown(String reason);
     }
 
     static class DefaultThreadingHandler implements ThreadingHandler {
@@ -566,34 +567,49 @@ private static class AgentState {
         private Future<?> task;
     }
 
+    private static void recordReason(Map<String, Integer> reasons, String reason) {
+        reasons.compute(reason, (keyReason, count) -> count == null ? 1 : ++count);
+    }
+
     static class DefaultCompletionHandler implements CompletionHandler {
 
         private final int timeLimit;
         private final CountDownLatch latch;
+        private final ConcurrentMap<String, Integer> reasons;
+        private final AtomicBoolean completed = new AtomicBoolean(false);
 
-        DefaultCompletionHandler(int timeLimit, int countLimit) {
+        DefaultCompletionHandler(int timeLimit, int countLimit, ConcurrentMap<String, Integer> reasons) {
             this.timeLimit = timeLimit;
             this.latch = new CountDownLatch(countLimit <= 0 ? 1 : countLimit);
+            this.reasons = reasons;
         }
 
         @Override
         public void waitForCompletion() throws InterruptedException {
             if (timeLimit <= 0) {
                 this.latch.await();
+                completed.set(true);
             } else {
                 boolean countedDown = this.latch.await(timeLimit, TimeUnit.SECONDS);
+                completed.set(true);
                 if (LOGGER.isDebugEnabled()) {
                     LOGGER.debug("Completed, counted down? {}", countedDown);
                 }
+                if (!countedDown) {
+                    recordReason(reasons, STOP_REASON_REACHED_TIME_LIMIT);
+                }
             }
         }
 
         @Override
-        public void countDown() {
+        public void countDown(String reason) {
             if (LOGGER.isDebugEnabled()) {
-                LOGGER.debug("Counting down");
+                LOGGER.debug("Counting down ({})", reason);
+            }
+            if (!completed.get()) {
+                recordReason(reasons, reason);
+                latch.countDown();
             }
-            latch.countDown();
         }
     }
 
@@ -605,14 +621,20 @@ public void countDown() {
     static class NoLimitCompletionHandler implements CompletionHandler {
 
         private final CountDownLatch latch = new CountDownLatch(1);
+        private final ConcurrentMap<String, Integer> reasons;
+
+        NoLimitCompletionHandler(ConcurrentMap<String, Integer> reasons) {
+            this.reasons = reasons;
+        }
 
         @Override
         public void waitForCompletion() throws InterruptedException {
             latch.await();
         }
 
         @Override
-        public void countDown() {
+        public void countDown(String reason) {
+            recordReason(reasons, reason);
             latch.countDown();
         }
     }

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

@@ -32,10 +32,7 @@
 import java.security.NoSuchAlgorithmException;
 import java.text.SimpleDateFormat;
 import java.util.*;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.SynchronousQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 import java.util.function.Function;
 
 import static com.rabbitmq.perf.OptionsUtils.forEach;
@@ -318,13 +315,16 @@ public static void main(String [] args, PerfTestOptions perfTestOptions) {
             p.setBodyFieldCount(bodyFieldCount);
             p.setBodyCount(bodyCount);
 
-            MulticastSet.CompletionHandler completionHandler = getCompletionHandler(p);
+            ConcurrentMap<String, Integer> completionReasons = new ConcurrentHashMap<>();
+
+            MulticastSet.CompletionHandler completionHandler = getCompletionHandler(p, completionReasons);
 
             factory.setExceptionHandler(perfTestOptions.exceptionHandler);
 
             MulticastSet set = new MulticastSet(stats, factory, p, testID, uris, completionHandler, shutdownService);
             set.run(true);
 
+            System.out.println(stopLine(completionReasons));
             stats.printFinal();
         } catch (ParseException exp) {
             System.err.println("Parsing failed. Reason: " + exp.getMessage());
@@ -340,6 +340,26 @@ public static void main(String [] args, PerfTestOptions perfTestOptions) {
         systemExiter.exit(exitStatus);
     }
 
+    static String stopLine(Map<String, Integer> reasons) {
+        StringBuilder stoppedLine = new StringBuilder("test stopped");
+        if (reasons.size() > 0) {
+            stoppedLine.append(" (");
+            int count = 1;
+            for (Map.Entry<String, Integer> reasonToCount : reasons.entrySet()) {
+                stoppedLine.append(reasonToCount.getKey());
+                if (reasonToCount.getValue() > 1) {
+                    stoppedLine.append(" [").append(reasonToCount.getValue()).append("]");
+                }
+                if (count < reasons.size()) {
+                    stoppedLine.append(", ");
+                }
+                count++;
+            }
+            stoppedLine.append(")");
+        }
+        return stoppedLine.toString();
+    }
+
     private static PrintWriter openCsvFileForWriting(String outputFile, ShutdownService shutdownService) throws IOException {
         PrintWriter output;
         File file = new File(outputFile);
@@ -396,7 +416,7 @@ private static ConnectionFactory configureNioIfRequested(CommandLineProxy cmd, C
         };
     }
 
-    static MulticastSet.CompletionHandler getCompletionHandler(MulticastParams p) {
+    static MulticastSet.CompletionHandler getCompletionHandler(MulticastParams p, ConcurrentMap<String, Integer> reasons) {
         MulticastSet.CompletionHandler completionHandler;
         if (p.hasLimit()) {
             int countLimit = 0;
@@ -406,14 +426,14 @@ static MulticastSet.CompletionHandler getCompletionHandler(MulticastParams p) {
                 countLimit += p.getProducerThreadCount();
             }
             if (p.getConsumerMsgCount() > 0) {
-                countLimit += p.getProducerThreadCount();
+                countLimit += p.getConsumerThreadCount();
             }
             completionHandler = new MulticastSet.DefaultCompletionHandler(
                 p.getTimeLimit(),
-                countLimit
-            );
+                countLimit,
+                reasons);
         } else {
-            completionHandler = new MulticastSet.NoLimitCompletionHandler();
+            completionHandler = new MulticastSet.NoLimitCompletionHandler(reasons);
         }
         return completionHandler;
     }

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

@@ -63,6 +63,9 @@ public class Producer extends AgentBase implements Runnable, ReturnListener,
     public static final String APP_ID_PROPERTY = "appId";
     public static final String CLUSTER_ID_PROPERTY = "clusterId";
     public static final String TIMESTAMP_HEADER = TIMESTAMP_PROPERTY;
+    static final String STOP_REASON_PRODUCER_MESSAGE_LIMIT = "Producer reached message limit";
+    static final String STOP_REASON_PRODUCER_THREAD_INTERRUPTED = "Producer thread interrupted";
+    static final String STOP_REASON_ERROR_IN_PRODUCER = "Error in producer";
     private final Channel channel;
     private final String  exchangeName;
     private final String  id;
@@ -356,12 +359,25 @@ public void run() {
             }
         } catch (RuntimeException e) {
             LOGGER.debug("Error in publisher", e);
+            String reason;
+            if (e.getCause() instanceof InterruptedException && this.rateIndicator.getValue() != 0.0f) {
+                // likely to have been interrupted while sleeping to honor rate limit
+                reason = STOP_REASON_PRODUCER_THREAD_INTERRUPTED;
+            } else {
+                reason = STOP_REASON_ERROR_IN_PRODUCER;
+            }
             // failing, we don't want to block the whole process, so counting down
-            countDown();
+            countDown(reason);
             throw e;
         }
         if (state.getMsgCount() >= msgLimit) {
-            countDown();
+            String reason;
+            if (msgLimit == 0) {
+                reason = STOP_REASON_PRODUCER_THREAD_INTERRUPTED;
+            } else {
+                reason = STOP_REASON_PRODUCER_MESSAGE_LIMIT;
+            }
+            countDown(reason);
         }
     }
 
@@ -406,7 +422,7 @@ public int incrementMessageCount() {
                 maybeHandlePublish(state);
             } catch (RuntimeException e) {
                 // failing, we don't want to block the whole process, so counting down
-                countDown();
+                countDown("Error in scheduled producer");
                 throw e;
             }
         };
@@ -416,7 +432,21 @@ public void maybeHandlePublish(AgentState state) {
         if (keepGoing(state)) {
             handlePublish(state);
         } else {
-            countDown();
+            String reason;
+            if (messageLimitReached(state)) {
+                reason = STOP_REASON_PRODUCER_MESSAGE_LIMIT;
+            } else {
+                reason = STOP_REASON_PRODUCER_THREAD_INTERRUPTED;
+            }
+            countDown(reason);
+        }
+    }
+
+    private boolean messageLimitReached(AgentState state) {
+        if (msgLimit == 0) {
+            return false;
+        } else {
+            return state.getMsgCount() >= msgLimit;
         }
     }
 
@@ -500,9 +530,9 @@ private void publish(MessageBodySource.MessageEnvelope messageEnvelope)
                              messageEnvelope.getBody());
     }
 
-    private void countDown() {
+    private void countDown(String reason) {
         if (completed.compareAndSet(false, true)) {
-            completionHandler.countDown();
+            completionHandler.countDown(reason);
         }
     }
 

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

@@ -21,6 +21,7 @@
 import java.net.URISyntaxException;
 import java.security.KeyManagementException;
 import java.security.NoSuchAlgorithmException;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeoutException;
 
@@ -46,7 +47,7 @@ public void run()
         throws IOException, InterruptedException, TimeoutException, NoSuchAlgorithmException, KeyManagementException, URISyntaxException, ExecutionException {
         this.stats = new SimpleScenarioStats(interval);
         for (MulticastParams p : params) {
-            MulticastSet set = new MulticastSet(stats, factory, p, null, PerfTest.getCompletionHandler(p));
+            MulticastSet set = new MulticastSet(stats, factory, p, null, PerfTest.getCompletionHandler(p, new ConcurrentHashMap<>()));
             stats.setup(p);
             set.run();
         }

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

@@ -19,6 +19,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
 
 public class VaryingScenario implements Scenario {
     private final String name;
@@ -60,7 +61,7 @@ private void run(Variable[] variables, List<VariableValue> values) throws Except
                 for (VariableValue value : values) {
                     value.setup(p);
                 }
-                MulticastSet set = new MulticastSet(stats0, factory, p, null, PerfTest.getCompletionHandler(p));
+                MulticastSet set = new MulticastSet(stats0, factory, p, null, PerfTest.getCompletionHandler(p, new ConcurrentHashMap<>()));
                 stats0.setup(p);
                 set.run();
                 for (VariableValue value : values) {

src/test/java/com/rabbitmq/perf/ConsumerSharingThreadsTest.java

@@ -117,7 +117,7 @@ public void waitForCompletion() {
             }
 
             @Override
-            public void countDown() {
+            public void countDown(String reason) {
             }
         }
         );