Merge pull request #110 from rabbitmq/rabbitmq-perf-test-106-topology-recovery

Handle connection and topology recovery

Author: michaelklishin

.travis.yml

@@ -14,21 +14,19 @@ addons:
     sources:
       - sourceline: deb https://packages.erlang-solutions.com/ubuntu trusty contrib
         key_url: https://packages.erlang-solutions.com/ubuntu/erlang_solutions.asc
-      - sourceline: deb https://dl.bintray.com/rabbitmq/debian trusty main
-        key_url: https://dl.bintray.com/rabbitmq/Keys/rabbitmq-release-signing-key.asc
     packages:
-      - esl-erlang=1:20.1
+      - esl-erlang=1:20.3
 # because of https://github.com/travis-ci/travis-ci/issues/8906
 before_install:
   - sudo mv /opt/jdk_switcher/jdk_switcher.sh /tmp
-  - sudo apt-get install rabbitmq-server=3.7.0-1
   - sudo mv /tmp/jdk_switcher.sh /opt/jdk_switcher/
-services:
-  - rabbitmq
 branches:
   only:
     - master
-script: ./mvnw clean verify -Dtravis=true
+before_script:
+  - ./bin/before_build.sh
+
+script: ./mvnw clean verify -Dtravis=true -Drabbitmqctl.bin='rabbitmq/sbin/rabbitmqctl'
 cache:
   directories:
     - $HOME/.m2

bin/before_build.sh

@@ -0,0 +1,11 @@
+#!/bin/sh
+
+wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.7.7/rabbitmq-server-generic-unix-3.7.7.tar.xz
+tar xf rabbitmq-server-generic-unix-3.7.7.tar.xz
+mv rabbitmq_server-3.7.7 rabbitmq
+
+rabbitmq/sbin/rabbitmq-server -detached
+
+sleep 3
+
+true

pom.xml

@@ -82,6 +82,7 @@
     <maven.jar.plugin.version>3.0.2</maven.jar.plugin.version>
     <buildnumber.plugin.version>1.4</buildnumber.plugin.version>
     <maven.surefire.plugin.version>2.21.0</maven.surefire.plugin.version>
+    <maven.failsafe.plugin.version>2.21.0</maven.failsafe.plugin.version>
 
     <!-- because of https://issues.apache.org/jira/browse/MRESOURCES-99 -->
     <build.timestamp>${maven.build.timestamp}</build.timestamp>
@@ -250,6 +251,11 @@
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-surefire-plugin</artifactId>
         <version>${maven.surefire.plugin.version}</version>
+        <configuration>
+          <includes>
+            <include>**/*Test.java</include>
+          </includes>
+        </configuration>
         <dependencies>
           <dependency>
             <groupId>org.junit.platform</groupId>
@@ -259,6 +265,33 @@
         </dependencies>
       </plugin>
 
+      <plugin>
+        <artifactId>maven-failsafe-plugin</artifactId>
+        <version>${maven.failsafe.plugin.version}</version>
+        <configuration>
+          <includes>
+            <include>**/*IT.java</include>
+          </includes>
+          <encoding>UTF-8</encoding>
+        </configuration>
+        <dependencies>
+          <dependency>
+            <groupId>org.junit.platform</groupId>
+            <artifactId>junit-platform-surefire-provider</artifactId>
+            <version>${junit.platform.version}</version>
+          </dependency>
+        </dependencies>
+
+        <executions>
+          <execution>
+            <goals>
+              <goal>integration-test</goal>
+              <goal>verify</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+
       <plugin>
         <groupId>org.codehaus.mojo</groupId>
         <artifactId>versions-maven-plugin</artifactId>

src/docs/asciidoc/usage-advanced.adoc

@@ -188,13 +188,17 @@ to handle the I/O. That's what the
    --producer-scheduler-threads 10
    --nio-threads 10
 
-PerfTest will use 2 extra threads for internal bookkeeping needs, so the total will be 12 threads
-for I/O over all the connections. With the default blocking I/O mode, each producer (or consumer)
+This way PerfTest will use  12 threads for I/O over all the connections.
+With the default blocking I/O mode, each producer (or consumer)
 uses a thread for the I/O loop, that is 2000 threads to simulate 1000 producers and
-1000 consumers.
+1000 consumers. Using NIO in PerfTest can dramatically reduce the resources used
+to simulate workloads with a large number of connections with appropriate tuning.
+
+Note that in NIO mode the number of threads used can increase temporarily when connections close
+unexpectedly and connection recovery kicks in. This is due to the NIO mode dispatching
+connection closing to non-I/O threads to avoid deadlocks. Connection recovery can be disabled
+with the `--disable-connection-recovery` flag.
 
-Using NIO in PerfTest can dramatically reduce the resources used to simulate workloads with a large
-number of connections with appropriate tuning.
 
 == Running Producers and Consumers on Different Machines
 

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

@@ -15,11 +15,27 @@
 
 package com.rabbitmq.perf;
 
+import com.rabbitmq.client.MissedHeartbeatException;
+import com.rabbitmq.client.ShutdownSignalException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.SocketException;
+import java.util.function.Predicate;
+
 /**
  *
  */
 public abstract class AgentBase {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(AgentBase.class);
+
+    // FIXME this is the condition to start connection recovery
+    // ensure it's the appropriate condition and get it from the Java client code
+    static final Predicate<ShutdownSignalException> CONNECTION_RECOVERY_TRIGGERED =
+        e -> !e.isInitiatedByApplication() || (e.getCause() instanceof MissedHeartbeatException);
+
     protected void delay(long now, AgentState state) {
 
         long elapsed = now - state.getLastStatsTime();
@@ -38,6 +54,51 @@ protected void delay(long now, AgentState state) {
         }
     }
 
+    protected boolean isConnectionRecoveryTriggered(ShutdownSignalException e) {
+        return CONNECTION_RECOVERY_TRIGGERED.test(e);
+    }
+
+    protected void handleShutdownSignalExceptionOnWrite(Recovery.RecoveryProcess recoveryProcess, ShutdownSignalException e) {
+        if (LOGGER.isDebugEnabled()) {
+            LOGGER.debug(
+                "Handling write error, recovery process enabled? {}, condition to trigger connection recovery? {}",
+                recoveryProcess.isEnabled(), isConnectionRecoveryTriggered(e)
+            );
+        }
+        if (shouldStop(recoveryProcess, e)) {
+            throw e;
+        }
+    }
+
+    protected boolean shouldStop(Recovery.RecoveryProcess recoveryProcess, ShutdownSignalException e) {
+        if (recoveryProcess.isEnabled()) {
+            // we stop only if the error isn't likely to trigger connection recovery
+            return !isConnectionRecoveryTriggered(e);
+        } else {
+            return true;
+        }
+    }
+
+    protected void dealWithWriteOperation(WriteOperation writeOperation, Recovery.RecoveryProcess recoveryProcess) throws IOException {
+        try {
+            writeOperation.call();
+        } catch (ShutdownSignalException e) {
+            handleShutdownSignalExceptionOnWrite(recoveryProcess, e);
+        } catch (SocketException e) {
+            if (recoveryProcess.isEnabled()) {
+                if (LOGGER.isDebugEnabled()) {
+                    LOGGER.debug(
+                        "Socket exception in write, recovery process is enabled, ignoring to let connection recovery carry on"
+                    );
+                }
+            } else {
+                throw e;
+            }
+        }
+    }
+
+    public abstract void recover(TopologyRecording topologyRecording);
+
     protected interface AgentState {
 
         float getRateLimit();
@@ -48,4 +109,9 @@ protected interface AgentState {
 
         int incrementMessageCount();
     }
+
+    @FunctionalInterface
+    interface WriteOperation {
+        void call() throws IOException;
+    }
 }

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

@@ -20,6 +20,8 @@
 import com.rabbitmq.client.DefaultConsumer;
 import com.rabbitmq.client.Envelope;
 import com.rabbitmq.client.ShutdownSignalException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.ByteArrayInputStream;
 import java.io.DataInputStream;
@@ -34,7 +36,9 @@
 
 public class Consumer extends AgentBase implements Runnable {
 
-    private ConsumerImpl                q;
+    private static final Logger LOGGER = LoggerFactory.getLogger(Consumer.class);
+
+    private volatile ConsumerImpl       q;
     private final Channel               channel;
     private final String                id;
     private final List<String>          queueNames;
@@ -52,12 +56,15 @@ public class Consumer extends AgentBase implements Runnable {
 
     private final ConsumerState state;
 
+    private final Recovery.RecoveryProcess recoveryProcess;
+
     public Consumer(Channel channel, String id,
                     List<String> queueNames, int txSize, boolean autoAck,
                     int multiAckEvery, Stats stats, float rateLimit, int msgLimit,
                     int consumerLatencyInMicroSeconds,
                     TimestampProvider timestampProvider,
-                    MulticastSet.CompletionHandler completionHandler) {
+                    MulticastSet.CompletionHandler completionHandler,
+                    Recovery.RecoveryProcess recoveryProcess) {
 
         this.channel           = channel;
         this.id                = id;
@@ -97,6 +104,8 @@ public Consumer(Channel channel, String id,
         }
 
         this.state = new ConsumerState(rateLimit);
+        this.recoveryProcess = recoveryProcess;
+        this.recoveryProcess.init(this);
     }
 
     public void run() {
@@ -124,21 +133,22 @@ private ConsumerImpl(Channel channel) {
         @Override
         public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
             int currentMessageCount = state.incrementMessageCount();
-
             if (msgLimit == 0 || currentMessageCount <= msgLimit) {
                 long messageTimestamp = timestampExtractor.apply(properties, body);
                 long nowTimestamp = timestampProvider.getCurrentTime();
 
                 if (!autoAck) {
-                    if (multiAckEvery == 0) {
-                        channel.basicAck(envelope.getDeliveryTag(), false);
-                    } else if (currentMessageCount % multiAckEvery == 0) {
-                        channel.basicAck(envelope.getDeliveryTag(), true);
-                    }
+                    dealWithWriteOperation(() -> {
+                        if (multiAckEvery == 0) {
+                            channel.basicAck(envelope.getDeliveryTag(), false);
+                        } else if (currentMessageCount % multiAckEvery == 0) {
+                            channel.basicAck(envelope.getDeliveryTag(), true);
+                        }
+                    }, recoveryProcess);
                 }
 
                 if (txSize != 0 && currentMessageCount % txSize == 0) {
-                    channel.txCommit();
+                    dealWithWriteOperation(() -> channel.txCommit(), recoveryProcess);
                 }
 
                 long diff_time = timestampProvider.getDifference(nowTimestamp, messageTimestamp);
@@ -157,7 +167,14 @@ public void handleDelivery(String consumerTag, Envelope envelope, BasicPropertie
 
         @Override
         public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
-            countDown();
+            LOGGER.debug(
+                "Consumer received shutdown signal, recovery process enabled? {}, condition to trigger connection recovery? {}",
+                recoveryProcess.isEnabled(), isConnectionRecoveryTriggered(sig)
+            );
+            if (!recoveryProcess.isEnabled()) {
+                LOGGER.debug("Counting down for consumer");
+                countDown();
+            }
         }
 
         @Override
@@ -179,6 +196,21 @@ private void countDown() {
         }
     }
 
+    @Override
+    public void recover(TopologyRecording topologyRecording) {
+        for (Map.Entry<String, String> entry : consumerTagBranchMap.entrySet()) {
+            TopologyRecording.RecordedQueue queue = topologyRecording.queue(entry.getValue());
+            try {
+                channel.basicConsume(queue.name(), autoAck, entry.getKey(), q);
+            } catch (IOException e) {
+                LOGGER.warn(
+                    "Error while recovering consumer {} on queue {} on connection {}",
+                    entry.getKey(), queue.name(), channel.getConnection().getClientProvidedName(), e
+                );
+            }
+        }
+    }
+
     private static class ConsumerState implements AgentState {
 
         private final float rateLimit;