Terminate management receive loop when idle

Applications are likely to configure their topology when they start and
modify it sporadically at best during the application lifetime.
Terminating the receive loop saves a thread.

There is now a dedicated, non-configurable thread pool for management
instances.

Author: acogoluegnes

src/main/java/com/rabbitmq/client/amqp/impl/AmqpConnection.java

@@ -29,6 +29,8 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.BiConsumer;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
@@ -69,6 +71,7 @@ final class AmqpConnection extends ResourceBase implements Connection {
   private final ConnectionUtils.AffinityContext affinity;
   private final ConnectionSettings.AffinityStrategy affinityStrategy;
   private final String name;
+  private final Lock instanceLock = new ReentrantLock();
 
   AmqpConnection(AmqpConnectionBuilder builder) {
     super(builder.listeners());
@@ -523,14 +526,17 @@ Session nativeSession(boolean check) {
     }
     Session result = this.nativeSession;
     if (result == null) {
-      synchronized (this) {
+      this.instanceLock.lock();
+      try {
         result = this.nativeSession;
         if (result == null) {
           if (check) {
             checkOpen();
           }
           this.nativeSession = result = this.openSession(this.nativeConnection);
         }
+      } finally {
+        this.instanceLock.unlock();
       }
     }
     return result;
@@ -548,6 +554,10 @@ org.apache.qpid.protonj2.client.Connection nativeConnection() {
     return this.nativeConnection;
   }
 
+  AmqpEnvironment environment() {
+    return this.environment;
+  }
+
   ExecutorService executorService() {
     return this.environment.executorService();
   }

src/main/java/com/rabbitmq/client/amqp/impl/AmqpEnvironment.java

@@ -49,6 +49,7 @@ class AmqpEnvironment implements Environment {
   private final ConnectionUtils.AffinityCache affinityCache = new ConnectionUtils.AffinityCache();
   private final EventLoop recoveryEventLoop;
   private final ExecutorService recoveryEventLoopExecutorService;
+  private final ExecutorService managementExecutorService;
 
   AmqpEnvironment(
       ExecutorService executorService,
@@ -91,6 +92,14 @@ class AmqpEnvironment implements Environment {
             new LinkedBlockingQueue<>(),
             Utils.threadFactory(threadPrefix + "event-loop-"));
     this.recoveryEventLoop = new EventLoop(this.recoveryEventLoopExecutorService);
+    this.managementExecutorService =
+        new ThreadPoolExecutor(
+            0,
+            Integer.MAX_VALUE,
+            30,
+            TimeUnit.SECONDS,
+            new SynchronousQueue<>(),
+            Utils.threadFactory(threadPrefix + "management-loop-"));
   }
 
   DefaultConnectionSettings<?> connectionSettings() {
@@ -117,6 +126,7 @@ public void close() {
       this.client.close();
       this.recoveryEventLoop.close();
       this.recoveryEventLoopExecutorService.shutdownNow();
+      this.managementExecutorService.shutdownNow();
       if (this.internalExecutor) {
         this.executorService.shutdownNow();
       }
@@ -139,6 +149,10 @@ ExecutorService executorService() {
     return this.executorService;
   }
 
+  ExecutorService managementExecutorService() {
+    return this.managementExecutorService;
+  }
+
   ScheduledExecutorService scheduledExecutorService() {
     return this.scheduledExecutorService;
   }

src/main/java/com/rabbitmq/client/amqp/impl/AmqpManagement.java

@@ -28,6 +28,8 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -72,6 +74,8 @@ class AmqpManagement implements Management {
   private final Supplier<String> nameSupplier;
   private final AtomicReference<State> state = new AtomicReference<>(CREATED);
   private final AtomicBoolean initializing = new AtomicBoolean(false);
+  private final Duration receiveLoopIdleTimeout;
+  private final Lock instanceLock = new ReentrantLock();
 
   AmqpManagement(AmqpManagementParameters parameters) {
     this.id = ID_SEQUENCE.getAndIncrement();
@@ -81,6 +85,10 @@ class AmqpManagement implements Management {
             ? TopologyListener.NO_OP
             : parameters.topologyListener();
     this.nameSupplier = parameters.nameSupplier();
+    this.receiveLoopIdleTimeout =
+        parameters.receiveLoopIdleTimeout() == null
+            ? Duration.ofSeconds(20)
+            : parameters.receiveLoopIdleTimeout();
   }
 
   @Override
@@ -190,6 +198,10 @@ void init() {
         LOGGER.debug("Initializing management ({}).", this);
         this.state(UNAVAILABLE);
         try {
+          if (this.receiveLoop != null) {
+            this.receiveLoop.cancel(true);
+            this.receiveLoop = null;
+          }
           LOGGER.debug("Creating management session ({}).", this);
           this.session = this.connection.nativeConnection().openSession();
           String linkPairName = "management-link-pair";
@@ -217,49 +229,6 @@ void init() {
           LOGGER.debug("Management sender created ({}).", this);
           this.receiver.openFuture().get(this.rpcTimeout.toMillis(), MILLISECONDS);
           LOGGER.debug("Management receiver created ({}).", this);
-          Runnable receiveTask =
-              () -> {
-                try {
-                  while (!Thread.currentThread().isInterrupted()) {
-                    Delivery delivery = receiver.receive(100, MILLISECONDS);
-                    if (delivery != null) {
-                      Object correlationId = delivery.message().correlationId();
-                      if (correlationId instanceof UUID) {
-                        OutstandingRequest request = outstandingRequests.remove(correlationId);
-                        if (request != null) {
-                          request.complete(delivery.message());
-                        } else {
-                          LOGGER.info("Could not find outstanding request {}", correlationId);
-                        }
-                      } else {
-                        LOGGER.info("Could not correlate inbound message with management request");
-                      }
-                    }
-                  }
-                } catch (ClientConnectionRemotelyClosedException
-                    | ClientLinkRemotelyClosedException e) {
-                  // receiver is closed
-                } catch (ClientSessionRemotelyClosedException e) {
-                  this.state(UNAVAILABLE);
-                  LOGGER.info(
-                      "Management session closed in receive loop: {} ({})", e.getMessage(), this);
-                  AmqpException exception = ExceptionUtils.convert(e);
-                  this.failRequests(exception);
-                  if (exception instanceof AmqpException.AmqpSecurityException) {
-                    LOGGER.debug(
-                        "Recovering AMQP management because the failure was a security exception ({}).",
-                        this);
-                    this.init();
-                  }
-                } catch (ClientException e) {
-                  java.util.function.Consumer<String> log =
-                      this.closed.get() ? m -> LOGGER.debug(m, e) : m -> LOGGER.info(m, e);
-                  log.accept("Error while polling AMQP receiver");
-                }
-              };
-          LOGGER.debug("Starting management receive loop ({}).", this);
-          this.receiveLoop = this.connection.executorService().submit(receiveTask);
-          LOGGER.debug("Management initialized ({}).", this);
           this.state(OPEN);
           this.initializing.set(false);
         } catch (Exception e) {
@@ -269,6 +238,58 @@ void init() {
     }
   }
 
+  private Runnable receiveTask() {
+    return () -> {
+      try {
+        Duration waitDuration = Duration.ofMillis(100);
+        long idleTime = 0;
+        while (!Thread.currentThread().isInterrupted()) {
+          Delivery delivery = receiver.receive(waitDuration.toMillis(), MILLISECONDS);
+          if (delivery != null) {
+            idleTime = 0;
+            Object correlationId = delivery.message().correlationId();
+            if (correlationId instanceof UUID) {
+              OutstandingRequest request = outstandingRequests.remove(correlationId);
+              if (request != null) {
+                request.complete(delivery.message());
+              } else {
+                LOGGER.info("Could not find outstanding request {}", correlationId);
+              }
+            } else {
+              LOGGER.info("Could not correlate inbound message with management request");
+            }
+          } else {
+            idleTime += waitDuration.toMillis();
+            if (idleTime > receiveLoopIdleTimeout.toMillis()) {
+              LOGGER.debug(
+                  "Management receive loop has been idle for more than {}, finishing it.",
+                  this.receiveLoopIdleTimeout);
+              this.receiveLoop = null;
+              return;
+            }
+          }
+        }
+      } catch (ClientConnectionRemotelyClosedException | ClientLinkRemotelyClosedException e) {
+        // receiver is closed
+      } catch (ClientSessionRemotelyClosedException e) {
+        this.state(UNAVAILABLE);
+        LOGGER.info("Management session closed in receive loop: {} ({})", e.getMessage(), this);
+        AmqpException exception = ExceptionUtils.convert(e);
+        this.failRequests(exception);
+        if (exception instanceof AmqpException.AmqpSecurityException) {
+          LOGGER.debug(
+              "Recovering AMQP management because the failure was a security exception ({}).",
+              this);
+          this.init();
+        }
+      } catch (ClientException e) {
+        java.util.function.Consumer<String> log =
+            this.closed.get() ? m -> LOGGER.debug(m, e) : m -> LOGGER.info(m, e);
+        log.accept("Error while polling AMQP receiver");
+      }
+    };
+  }
+
   private void failRequests(AmqpException exception) {
     Iterator<Map.Entry<UUID, OutstandingRequest>> iterator =
         this.outstandingRequests.entrySet().iterator();
@@ -284,6 +305,7 @@ void releaseResources() {
     this.markUnavailable();
     if (this.receiveLoop != null) {
       this.receiveLoop.cancel(true);
+      this.receiveLoop = null;
     }
   }
 
@@ -338,6 +360,22 @@ OutstandingRequest request(Message<?> request, UUID requestId) throws ClientExce
     this.outstandingRequests.put(requestId, outstandingRequest);
     LOGGER.debug("Sending request {}", requestId);
     this.sender.send(request);
+    Future<?> loop = this.receiveLoop;
+    if (loop == null) {
+      this.instanceLock.lock();
+      try {
+        loop = this.receiveLoop;
+        if (loop == null) {
+          Runnable receiveTask = receiveTask();
+          LOGGER.debug("Starting management receive loop ({}).", this);
+          this.receiveLoop =
+              this.connection.environment().managementExecutorService().submit(receiveTask);
+          LOGGER.debug("Management initialized ({}).", this);
+        }
+      } finally {
+        this.instanceLock.unlock();
+      }
+    }
     return outstandingRequest;
   }
 
@@ -548,6 +586,11 @@ TopologyListener recovery() {
     return this.topologyListener;
   }
 
+  // for testing
+  boolean hasReceiveLoop() {
+    return this.receiveLoop != null;
+  }
+
   private static class DefaultQueueInfo implements QueueInfo {
 
     private final String name;

src/main/java/com/rabbitmq/client/amqp/impl/AmqpManagementParameters.java

@@ -17,13 +17,15 @@
 // [email protected].
 package com.rabbitmq.client.amqp.impl;
 
+import java.time.Duration;
 import java.util.function.Supplier;
 
 class AmqpManagementParameters {
 
   private final AmqpConnection connection;
   private TopologyListener topologyListener;
   private Supplier<String> nameSupplier = Utils.NAME_SUPPLIER;
+  private Duration receiveLoopIdleTimeout;
 
   AmqpManagementParameters(AmqpConnection connection) {
     this.connection = connection;
@@ -39,6 +41,15 @@ AmqpManagementParameters nameSupplier(Supplier<String> nameSupplier) {
     return this;
   }
 
+  public AmqpManagementParameters receiveLoopIdleTimeout(Duration receiveLoopIdleTimeout) {
+    this.receiveLoopIdleTimeout = receiveLoopIdleTimeout;
+    return this;
+  }
+
+  Duration receiveLoopIdleTimeout() {
+    return receiveLoopIdleTimeout;
+  }
+
   AmqpConnection connection() {
     return this.connection;
   }

src/test/java/com/rabbitmq/client/amqp/impl/AmqpConnectionRecoveryTest.java

@@ -165,7 +165,6 @@ void connectionShouldRecoverAfterClosingIt(boolean isolateResources, TestInfo in
       assertThat(receivedMessageIds)
           .hasSameSizeAs(publishedMessageIds)
           .containsAll(publishedMessageIds);
-
     } finally {
       c.management().queueDeletion().delete(q);
       c.close();

src/test/java/com/rabbitmq/client/amqp/impl/ManagementTest.java

@@ -17,11 +17,13 @@
 // [email protected].
 package com.rabbitmq.client.amqp.impl;
 
+import static com.rabbitmq.client.amqp.impl.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import com.rabbitmq.client.amqp.AmqpException;
 import com.rabbitmq.client.amqp.Management;
+import java.time.Duration;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Supplier;
 import org.junit.jupiter.api.*;
@@ -58,7 +60,7 @@ void queueDeclareWithClientNamedQueueShouldBeRetriedIfNameAlreadyExists() {
         new AmqpManagement(new AmqpManagementParameters(connection).nameSupplier(nameSupplier));
     management.init();
     Management.QueueInfo queueInfo = management.queue().exclusive(true).autoDelete(true).declare();
-    Assertions.assertThat(queueInfo).hasName(q);
+    assertThat(queueInfo).hasName(q);
     assertThat(nameSupplierCallCount).hasValue(1);
     queueInfo = management.queue().exclusive(true).autoDelete(false).declare();
     assertThat(queueInfo.name()).isNotEqualTo(q);
@@ -70,4 +72,21 @@ void queueInfoShouldThrowDoesNotExistExceptionWhenQueueDoesNotExist() {
     assertThatThrownBy(() -> connection.management().queueInfo("do not exists"))
         .isInstanceOf(AmqpException.AmqpEntityDoesNotExistException.class);
   }
+
+  @Test
+  void receiveLoopShouldStopAfterBeingIdle() {
+    management =
+        new AmqpManagement(
+            new AmqpManagementParameters(connection)
+                .receiveLoopIdleTimeout(Duration.ofMillis(500)));
+    management.init();
+    assertThat(management.hasReceiveLoop()).isFalse();
+    Management.QueueInfo info1 = management.queue().exclusive(true).autoDelete(true).declare();
+    assertThat(management.hasReceiveLoop()).isTrue();
+    TestUtils.waitAtMost(() -> !management.hasReceiveLoop());
+    Management.QueueInfo info2 = management.queue().exclusive(true).autoDelete(true).declare();
+    assertThat(management.hasReceiveLoop()).isTrue();
+    assertThat(management.queueInfo(info1.name())).hasName(info1.name());
+    assertThat(management.queueInfo(info2.name())).hasName(info2.name());
+  }
 }