Prestart connection pool async work manager when running wait-queue-timeout.json, wait-queue-timeouts.json (#804)

JAVA-4318

Author: stIncMale

driver-core/src/main/com/mongodb/connection/ConnectionPoolSettings.java

@@ -398,15 +398,15 @@ public int hashCode() {
     @Override
     public String toString() {
         return "ConnectionPoolSettings{"
-               + "maxSize=" + maxSize
-               + ", minSize=" + minSize
-               + ", maxWaitTimeMS=" + maxWaitTimeMS
-               + ", maxConnectionLifeTimeMS=" + maxConnectionLifeTimeMS
-               + ", maxConnectionIdleTimeMS=" + maxConnectionIdleTimeMS
-               + ", maintenanceInitialDelayMS=" + maintenanceInitialDelayMS
-               + ", maintenanceFrequencyMS=" + maintenanceFrequencyMS
-               + ", connectionPoolListeners=" + connectionPoolListeners
-               + '}';
+                + "maxSize=" + maxSize
+                + ", minSize=" + minSize
+                + ", maxWaitTimeMS=" + maxWaitTimeMS
+                + ", maxConnectionLifeTimeMS=" + maxConnectionLifeTimeMS
+                + ", maxConnectionIdleTimeMS=" + maxConnectionIdleTimeMS
+                + ", maintenanceInitialDelayMS=" + maintenanceInitialDelayMS
+                + ", maintenanceFrequencyMS=" + maintenanceFrequencyMS
+                + ", connectionPoolListeners=" + connectionPoolListeners
+                + '}';
     }
 
     ConnectionPoolSettings(final Builder builder) {

driver-core/src/main/com/mongodb/internal/connection/DefaultClusterFactory.java

@@ -44,6 +44,7 @@ public final class DefaultClusterFactory {
      * @param clusterSettings        the cluster settings
      * @param serverSettings         the server settings
      * @param connectionPoolSettings the connection pool settings
+     * @param internalConnectionPoolSettings the internal connection pool settings
      * @param streamFactory          the stream factory
      * @param heartbeatStreamFactory the heartbeat stream factory
      * @param credential             the credential, which may be null
@@ -57,8 +58,10 @@ public final class DefaultClusterFactory {
      * @since 3.6
      */
     public Cluster createCluster(final ClusterSettings clusterSettings, final ServerSettings serverSettings,
-                                 final ConnectionPoolSettings connectionPoolSettings, final StreamFactory streamFactory,
-                                 final StreamFactory heartbeatStreamFactory, final @Nullable MongoCredential credential,
+                                 final ConnectionPoolSettings connectionPoolSettings,
+                                 final InternalConnectionPoolSettings internalConnectionPoolSettings,
+                                 final StreamFactory streamFactory, final StreamFactory heartbeatStreamFactory,
+                                 final @Nullable MongoCredential credential,
                                  final CommandListener commandListener, final String applicationName,
                                  final MongoDriverInformation mongoDriverInformation,
                                  final List<MongoCompressor> compressorList, final @Nullable ServerApi serverApi) {
@@ -69,13 +72,14 @@ public Cluster createCluster(final ClusterSettings clusterSettings, final Server
 
         if (clusterSettings.getMode() == ClusterConnectionMode.LOAD_BALANCED) {
             ClusterableServerFactory serverFactory = new LoadBalancedClusterableServerFactory(clusterId, serverSettings,
-                    connectionPoolSettings, streamFactory, credential, commandListener, applicationName,
+                    connectionPoolSettings, internalConnectionPoolSettings, streamFactory, credential, commandListener, applicationName,
                     mongoDriverInformation != null ? mongoDriverInformation : MongoDriverInformation.builder().build(), compressorList,
                     serverApi);
             return new LoadBalancedCluster(clusterId, clusterSettings, serverFactory, dnsSrvRecordMonitorFactory);
         } else {
             ClusterableServerFactory serverFactory = new DefaultClusterableServerFactory(clusterId, clusterSettings, serverSettings,
-                    connectionPoolSettings, streamFactory, heartbeatStreamFactory, credential, commandListener, applicationName,
+                    connectionPoolSettings, internalConnectionPoolSettings,
+                    streamFactory, heartbeatStreamFactory, credential, commandListener, applicationName,
                     mongoDriverInformation != null ? mongoDriverInformation : MongoDriverInformation.builder().build(), compressorList,
                     serverApi);
 

driver-core/src/main/com/mongodb/internal/connection/DefaultClusterableServerFactory.java

@@ -41,6 +41,7 @@ public class DefaultClusterableServerFactory implements ClusterableServerFactory
     private final ClusterSettings clusterSettings;
     private final ServerSettings serverSettings;
     private final ConnectionPoolSettings connectionPoolSettings;
+    private final InternalConnectionPoolSettings internalConnectionPoolSettings;
     private final StreamFactory streamFactory;
     private final MongoCredentialWithCache credential;
     private final StreamFactory heartbeatStreamFactory;
@@ -52,15 +53,17 @@ public class DefaultClusterableServerFactory implements ClusterableServerFactory
     private final ServerApi serverApi;
 
     public DefaultClusterableServerFactory(final ClusterId clusterId, final ClusterSettings clusterSettings,
-                                           final ServerSettings serverSettings, final ConnectionPoolSettings connectionPoolSettings,
-                                           final StreamFactory streamFactory, final StreamFactory heartbeatStreamFactory,
-                                           final MongoCredential credential, final CommandListener commandListener,
-                                           final String applicationName, final MongoDriverInformation mongoDriverInformation,
-                                           final List<MongoCompressor> compressorList, final @Nullable ServerApi serverApi) {
+            final ServerSettings serverSettings, final ConnectionPoolSettings connectionPoolSettings,
+            final InternalConnectionPoolSettings internalConnectionPoolSettings,
+            final StreamFactory streamFactory, final StreamFactory heartbeatStreamFactory,
+            final MongoCredential credential, final CommandListener commandListener,
+            final String applicationName, final MongoDriverInformation mongoDriverInformation,
+            final List<MongoCompressor> compressorList, final @Nullable ServerApi serverApi) {
         this.clusterId = clusterId;
         this.clusterSettings = clusterSettings;
         this.serverSettings = serverSettings;
         this.connectionPoolSettings = connectionPoolSettings;
+        this.internalConnectionPoolSettings = internalConnectionPoolSettings;
         this.streamFactory = streamFactory;
         this.credential = credential == null ? null : new MongoCredentialWithCache(credential);
         this.heartbeatStreamFactory = heartbeatStreamFactory;
@@ -86,7 +89,7 @@ mongoDriverInformation, emptyList(), null, serverApi),
         ConnectionPool connectionPool = new DefaultConnectionPool(serverId,
                 new InternalStreamConnectionFactory(clusterSettings.getMode(), streamFactory, credential, applicationName,
                         mongoDriverInformation, compressorList, commandListener, serverApi),
-                connectionPoolSettings, sdamProvider);
+                connectionPoolSettings, internalConnectionPoolSettings, sdamProvider);
         SdamServerDescriptionManager sdam = new DefaultSdamServerDescriptionManager(serverId, serverDescriptionChangedListener,
                 serverListener, serverMonitor, connectionPool, clusterSettings.getMode());
         sdamProvider.initialize(sdam);

driver-core/src/main/com/mongodb/internal/connection/DefaultConnectionPool.java

@@ -118,6 +118,11 @@ class DefaultConnectionPool implements ConnectionPool {
     private final StateAndGeneration stateAndGeneration;
     private final OptionalProvider<SdamServerDescriptionManager> sdamProvider;
 
+    DefaultConnectionPool(final ServerId serverId, final InternalConnectionFactory internalConnectionFactory,
+            final ConnectionPoolSettings settings, final OptionalProvider<SdamServerDescriptionManager> sdamProvider) {
+        this(serverId, internalConnectionFactory, settings, InternalConnectionPoolSettings.builder().build(), sdamProvider);
+    }
+
     /**
      * @param sdamProvider For handling exceptions via the
      *                     <a href="https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst">
@@ -128,7 +133,8 @@ class DefaultConnectionPool implements ConnectionPool {
      *                     otherwise must provide a non-empty {@link Optional}.
      */
     DefaultConnectionPool(final ServerId serverId, final InternalConnectionFactory internalConnectionFactory,
-                          final ConnectionPoolSettings settings, final OptionalProvider<SdamServerDescriptionManager> sdamProvider) {
+            final ConnectionPoolSettings settings, final InternalConnectionPoolSettings internalSettings,
+            final OptionalProvider<SdamServerDescriptionManager> sdamProvider) {
         this.serverId = notNull("serverId", serverId);
         this.settings = notNull("settings", settings);
         UsageTrackingInternalConnectionItemFactory connectionItemFactory =
@@ -140,7 +146,7 @@ class DefaultConnectionPool implements ConnectionPool {
         backgroundMaintenance = new BackgroundMaintenanceManager();
         connectionPoolCreated(connectionPoolListener, serverId, settings);
         openConcurrencyLimiter = new OpenConcurrencyLimiter(MAX_CONNECTING);
-        asyncWorkManager = new AsyncWorkManager();
+        asyncWorkManager = new AsyncWorkManager(internalSettings.isPrestartAsyncWorkManager());
         stateAndGeneration = new StateAndGeneration();
         connectionGenerationSupplier = new ConnectionGenerationSupplier() {
             @Override
@@ -1252,10 +1258,13 @@ private static class AsyncWorkManager implements AutoCloseable {
         @Nullable
         private ExecutorService worker;
 
-        AsyncWorkManager() {
+        AsyncWorkManager(final boolean prestart) {
             state = State.NEW;
             tasks = new LinkedBlockingQueue<>();
             lock = new StampedLock().asWriteLock();
+            if (prestart) {
+                assertTrue(initUnlessClosed());
+            }
         }
 
         void enqueue(final Task task) {
@@ -1272,7 +1281,7 @@ void enqueue(final Task task) {
         }
 
         /**
-         * Invocations of this method must be guarded by {@link #lock}.
+         * Invocations of this method must be guarded by {@link #lock}, unless done from the constructor.
          *
          * @return {@code false} iff the {@link #state} is {@link State#CLOSED}.
          */

driver-core/src/main/com/mongodb/internal/connection/InternalConnectionPoolSettings.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.mongodb.internal.connection;
+
+import com.mongodb.annotations.Immutable;
+import com.mongodb.annotations.NotThreadSafe;
+import java.util.Objects;
+
+@Immutable
+public final class InternalConnectionPoolSettings {
+    private final boolean prestartAsyncWorkManager;
+
+    private InternalConnectionPoolSettings(final Builder builder) {
+        prestartAsyncWorkManager = builder.prestartAsyncWorkManager;
+    }
+
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    /**
+     * Specifies whether to pre-start the asynchronous work manager of the pool.
+     * <p>
+     * Default is {@code false}.
+     *
+     * @return {@code true} iff pool's asynchronous work manager must be pre-started.
+     * @see Builder#prestartAsyncWorkManager(boolean)
+     */
+    public boolean isPrestartAsyncWorkManager() {
+        return prestartAsyncWorkManager;
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        final InternalConnectionPoolSettings that = (InternalConnectionPoolSettings) o;
+        return prestartAsyncWorkManager == that.prestartAsyncWorkManager;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(prestartAsyncWorkManager);
+    }
+
+    @Override
+    public String toString() {
+        return "InternalConnectionPoolSettings{"
+                + "prestartAsyncWorkManager=" + prestartAsyncWorkManager
+                + '}';
+    }
+
+    @NotThreadSafe
+    public static final class Builder {
+        private boolean prestartAsyncWorkManager = false;
+
+        private Builder() {
+        }
+
+        /**
+         * Allows to pre-start the asynchronous work manager of the pool.
+         *
+         * @param prestart {@code true} iff pool's asynchronous work manager must be pre-started.
+         * @return {@code this}.
+         * @see InternalConnectionPoolSettings#isPrestartAsyncWorkManager()
+         */
+        public Builder prestartAsyncWorkManager(final boolean prestart) {
+            prestartAsyncWorkManager = prestart;
+            return this;
+        }
+
+        public InternalConnectionPoolSettings build() {
+            return new InternalConnectionPoolSettings(this);
+        }
+    }
+}

driver-core/src/main/com/mongodb/internal/connection/LoadBalancedClusterableServerFactory.java

@@ -39,6 +39,7 @@ public class LoadBalancedClusterableServerFactory implements ClusterableServerFa
     private final ClusterId clusterId;
     private final ServerSettings serverSettings;
     private final ConnectionPoolSettings connectionPoolSettings;
+    private final InternalConnectionPoolSettings internalConnectionPoolSettings;
     private final StreamFactory streamFactory;
     private final MongoCredentialWithCache credential;
     private final CommandListener commandListener;
@@ -48,13 +49,16 @@ public class LoadBalancedClusterableServerFactory implements ClusterableServerFa
     private final ServerApi serverApi;
 
     public LoadBalancedClusterableServerFactory(final ClusterId clusterId, final ServerSettings serverSettings,
-                                                final ConnectionPoolSettings connectionPoolSettings, final StreamFactory streamFactory,
-                                                final MongoCredential credential, final CommandListener commandListener,
+                                                final ConnectionPoolSettings connectionPoolSettings,
+                                                final InternalConnectionPoolSettings internalConnectionPoolSettings,
+                                                final StreamFactory streamFactory, final MongoCredential credential,
+                                                final CommandListener commandListener,
                                                 final String applicationName, final MongoDriverInformation mongoDriverInformation,
                                                 final List<MongoCompressor> compressorList, final ServerApi serverApi) {
         this.clusterId = clusterId;
         this.serverSettings = serverSettings;
         this.connectionPoolSettings = connectionPoolSettings;
+        this.internalConnectionPoolSettings = internalConnectionPoolSettings;
         this.streamFactory = streamFactory;
         this.credential = credential == null ? null : new MongoCredentialWithCache(credential);
         this.commandListener = commandListener;
@@ -70,8 +74,8 @@ public ClusterableServer create(final ServerAddress serverAddress,
                                     final ServerListener serverListener, final ClusterClock clusterClock) {
         ConnectionPool connectionPool = new DefaultConnectionPool(new ServerId(clusterId, serverAddress),
                 new InternalStreamConnectionFactory(ClusterConnectionMode.LOAD_BALANCED, streamFactory, credential, applicationName,
-                        mongoDriverInformation, compressorList, commandListener, serverApi), connectionPoolSettings,
-                EmptyProvider.instance());
+                        mongoDriverInformation, compressorList, commandListener, serverApi),
+                connectionPoolSettings, internalConnectionPoolSettings, EmptyProvider.instance());
         connectionPool.ready();
 
         return new LoadBalancedServer(new ServerId(clusterId, serverAddress), connectionPool, new DefaultConnectionFactory(),

driver-core/src/test/functional/com/mongodb/ClusterFixture.java

@@ -50,6 +50,7 @@
 import com.mongodb.internal.connection.AsyncConnection;
 import com.mongodb.internal.connection.Cluster;
 import com.mongodb.internal.connection.DefaultClusterFactory;
+import com.mongodb.internal.connection.InternalConnectionPoolSettings;
 import com.mongodb.internal.connection.MongoCredentialWithCache;
 import com.mongodb.internal.operation.AsyncReadOperation;
 import com.mongodb.internal.operation.AsyncWriteOperation;
@@ -380,7 +381,7 @@ public static Cluster createAsyncCluster(final MongoCredential credential) {
     private static Cluster createCluster(final MongoCredential credential, final StreamFactory streamFactory) {
         return new DefaultClusterFactory().createCluster(ClusterSettings.builder().hosts(asList(getPrimary())).build(),
                 ServerSettings.builder().build(),
-                ConnectionPoolSettings.builder().maxSize(1).build(),
+                ConnectionPoolSettings.builder().maxSize(1).build(), InternalConnectionPoolSettings.builder().build(),
                 streamFactory, streamFactory, credential, null, null, null,
                 Collections.<MongoCompressor>emptyList(), getServerApi());
     }
@@ -389,6 +390,7 @@ private static Cluster createCluster(final ConnectionString connectionString, fi
         return new DefaultClusterFactory().createCluster(ClusterSettings.builder().applyConnectionString(connectionString).build(),
                 ServerSettings.builder().build(),
                 ConnectionPoolSettings.builder().applyConnectionString(connectionString).build(),
+                InternalConnectionPoolSettings.builder().build(),
                 streamFactory,
                 new SocketStreamFactory(SocketSettings.builder().readTimeout(5, SECONDS).build(), getSslSettings(connectionString)),
                 connectionString.getCredential(),

driver-core/src/test/functional/com/mongodb/internal/connection/SingleServerClusterTest.java

@@ -69,7 +69,7 @@ private void setUpCluster(final ServerAddress serverAddress) {
         cluster = new SingleServerCluster(clusterId,
                 clusterSettings,
                 new DefaultClusterableServerFactory(clusterId, clusterSettings, ServerSettings.builder().build(),
-                        ConnectionPoolSettings.builder().maxSize(1).build(),
+                        ConnectionPoolSettings.builder().maxSize(1).build(), InternalConnectionPoolSettings.builder().build(),
                         streamFactory, streamFactory, getCredential(),
 
                         null, null, null,