JAVA-1202: Dedicate a single thread scheduled executor to each server being monitored, and , when changing the heartbeat frequency, ensure that the ServerStateNotifier is scheduled at most once by cancelling the existing schedule before creating the new one.

jyemin · jyemin · commit 7d59cc8217eb · 2014-05-04T10:46:40.000-04:00
diff --git a/src/main/com/mongodb/BaseCluster.java b/src/main/com/mongodb/BaseCluster.java
@@ -138,7 +138,6 @@ public ClusterSettings getSettings() {
     public void close() {
         if (!isClosed()) {
             isClosed = true;
-            serverFactory.close();
             phase.get().countDown();
             clusterListener.clusterClosed(new ClusterEvent(clusterId));
         }
diff --git a/src/main/com/mongodb/ClusterableServerFactory.java b/src/main/com/mongodb/ClusterableServerFactory.java
@@ -18,6 +18,4 @@
 
 interface ClusterableServerFactory {
     ClusterableServer create(ServerAddress serverAddress);
-
-    void close();
 }
diff --git a/src/main/com/mongodb/Clusters.java b/src/main/com/mongodb/Clusters.java
@@ -16,8 +16,6 @@
 
 package com.mongodb;
 
-import java.util.concurrent.ScheduledExecutorService;
-
 /**
  * A factory for cluster implementations.
  *
@@ -27,11 +25,8 @@ private Clusters() {
     }
 
     public static Cluster create(final String clusterId, final ClusterSettings settings, final ServerSettings serverSettings,
-                                 final ScheduledExecutorService scheduledExecutorService,
-                                 final ClusterListener clusterListener,
-                                 final Mongo mongo) {
-        ClusterableServerFactory serverFactory = new DefaultClusterableServerFactory(clusterId, serverSettings, scheduledExecutorService,
-                                                                                     mongo);
+                                 final ClusterListener clusterListener, final Mongo mongo) {
+        ClusterableServerFactory serverFactory = new DefaultClusterableServerFactory(clusterId, serverSettings, mongo);
 
         if (settings.getMode() == ClusterConnectionMode.Single) {
             return new SingleServerCluster(clusterId, settings, serverFactory,
diff --git a/src/main/com/mongodb/DBTCPConnector.java b/src/main/com/mongodb/DBTCPConnector.java
@@ -21,9 +21,6 @@
 import java.net.SocketTimeoutException;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static com.mongodb.ClusterConnectionMode.Multiple;
@@ -47,7 +44,6 @@ public class DBTCPConnector implements DBConnector {
 
     private final Mongo _mongo;
 
-    private ScheduledExecutorService scheduledExecutorService;
     private Cluster cluster;
 
     private final MyPort _myPort = new MyPort();
@@ -73,10 +69,6 @@ public void start() {
         MongoOptions options = _mongo.getMongoOptions();
 
         String clusterId = Integer.toString(NEXT_CLUSTER_ID.getAndIncrement());
-        scheduledExecutorService = Executors.newScheduledThreadPool(options.heartbeatThreadCount > 0 ?
-                                                                    options.heartbeatThreadCount :
-                                                                    _mongo.getAuthority().getServerAddresses().size(),
-                                                                    new DefaultThreadFactory(clusterId));
         cluster =
         Clusters.create(clusterId,
                         ClusterSettings.builder()
@@ -94,7 +86,7 @@ public void start() {
                                                                              .socketFactory(_mongo.getMongoOptions().getSocketFactory())
                                                                              .build())
                                       .build(),
-                        scheduledExecutorService, null, _mongo);
+                        null, _mongo);
     }
 
     /**
@@ -603,10 +595,6 @@ public void close(){
             cluster.close();
             cluster = null;
         }
-        if (scheduledExecutorService != null) {
-            scheduledExecutorService.shutdownNow();
-            scheduledExecutorService = null;
-        }
     }
 
     /**
@@ -665,21 +653,4 @@ MyPort getMyPort() {
     private Server getServer(final ServerSelector serverSelector) {
         return cluster.getServer(serverSelector, getClusterWaitTimeMS(), MILLISECONDS);
     }
-
-    // Custom thread factory for scheduled executor service that creates daemon threads.  Otherwise,
-    // applications that neglect to close the MongoClient will not exit.
-    static class DefaultThreadFactory implements ThreadFactory {
-        private final AtomicInteger threadNumber = new AtomicInteger(1);
-        private final String clusterId;
-
-        DefaultThreadFactory(final String clusterId) {
-            this.clusterId = clusterId;
-        }
-
-        public Thread newThread(Runnable runnable) {
-            Thread t = new Thread(runnable, "cluster-" + clusterId + "-thread-" + threadNumber.getAndIncrement());
-            t.setDaemon(true);
-            return t;
-        }
-    }
 }
diff --git a/src/main/com/mongodb/DefaultClusterableServerFactory.java b/src/main/com/mongodb/DefaultClusterableServerFactory.java
@@ -16,22 +16,16 @@
 
 package com.mongodb;
 
-import java.util.concurrent.ScheduledExecutorService;
-
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
 class DefaultClusterableServerFactory implements ClusterableServerFactory {
     private final String clusterId;
     private ServerSettings settings;
-    private final ScheduledExecutorService scheduledExecutorService;
     private final Mongo mongo;
 
-    public DefaultClusterableServerFactory(final String clusterId, final ServerSettings settings,
-                                           final ScheduledExecutorService scheduledExecutorService,
-                                           final Mongo mongo) {
+    public DefaultClusterableServerFactory(final String clusterId, final ServerSettings settings, final Mongo mongo) {
         this.clusterId = clusterId;
         this.settings = settings;
-        this.scheduledExecutorService = scheduledExecutorService;
         this.mongo = mongo;
     }
 
@@ -47,14 +41,9 @@ public ClusterableServer create(final ServerAddress serverAddress) {
                               .maxWaitQueueSize(options.getConnectionsPerHost() * options.getThreadsAllowedToBlockForConnectionMultiplier())
                               .maxWaitTime(options.getMaxWaitTime(), MILLISECONDS)
                               .build();
-        return new DefaultServer(serverAddress, settings,
+        return new DefaultServer(serverAddress, settings, clusterId,
                                  new PooledConnectionProvider(clusterId, serverAddress, new DBPortFactory(options), connectionPoolSettings,
                                                               new JMXConnectionPoolListener(mongo.getMongoOptions().getDescription())),
-                                 scheduledExecutorService, mongo);
-    }
-
-    @Override
-    public void close() {
-        scheduledExecutorService.shutdownNow();
+                                 mongo);
     }
 }
diff --git a/src/main/com/mongodb/DefaultServer.java b/src/main/com/mongodb/DefaultServer.java
@@ -18,8 +18,10 @@
 
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 
 import static com.mongodb.ServerConnectionState.Connecting;
@@ -29,10 +31,30 @@
 import static org.bson.util.Assertions.notNull;
 
 class DefaultServer implements ClusterableServer {
+
+    private final String clusterId;
+
+    private enum HeartbeatFrequency {
+        NORMAL {
+            @Override
+            long getFrequencyMS(final ServerSettings settings) {
+                return settings.getHeartbeatFrequency(MILLISECONDS);
+            }
+        },
+
+        RETRY {
+            @Override
+            long getFrequencyMS(final ServerSettings settings) {
+                return settings.getHeartbeatConnectRetryFrequency(MILLISECONDS);
+            }
+        };
+
+        abstract long getFrequencyMS(final ServerSettings settings);
+    }
+
     private final ScheduledExecutorService scheduledExecutorService;
     private final ServerAddress serverAddress;
     private final ServerStateNotifier stateNotifier;
-    private final ScheduledFuture<?> scheduledFuture;
     private final PooledConnectionProvider connectionProvider;
     private final Map<ChangeListener<ServerDescription>, Boolean> changeListeners =
     new ConcurrentHashMap<ChangeListener<ServerDescription>, Boolean>();
@@ -41,22 +63,22 @@ class DefaultServer implements ClusterableServer {
     private volatile ServerDescription description;
     private volatile boolean isClosed;
 
+    private ScheduledFuture<?> scheduledFuture;
+    private HeartbeatFrequency currentFrequency;
+
     public DefaultServer(final ServerAddress serverAddress,
                          final ServerSettings settings,
-                         final PooledConnectionProvider connectionProvider,
-                         final ScheduledExecutorService scheduledExecutorService,
-                         Mongo mongo) {
+                         final String clusterId, final PooledConnectionProvider connectionProvider,
+                         final Mongo mongo) {
+        this.clusterId = notNull("clusterId", clusterId);
         this.settings = notNull("settings", settings);
-
-        this.scheduledExecutorService = notNull("scheduledExecutorService", scheduledExecutorService);
         this.serverAddress = notNull("serverAddress", serverAddress);
         this.description = ServerDescription.builder().state(Connecting).address(serverAddress).build();
         serverStateListener = new DefaultServerStateListener();
         this.stateNotifier = new ServerStateNotifier(serverAddress, serverStateListener,
                                                      settings.getHeartbeatSocketSettings(), mongo);
-        this.scheduledFuture = scheduledExecutorService.scheduleAtFixedRate(stateNotifier, 0,
-                                                                            settings.getHeartbeatFrequency(MILLISECONDS),
-                                                                            MILLISECONDS);
+        this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory());
+        setHeartbeat(0, HeartbeatFrequency.NORMAL);
         this.connectionProvider = connectionProvider;
     }
 
@@ -87,14 +109,15 @@ public void invalidate() {
         serverStateListener.stateChanged(new ChangeEvent<ServerDescription>(description, ServerDescription.builder()
                                                                                                           .state(Connecting)
                                                                                                           .address(serverAddress).build()));
-        scheduledExecutorService.submit(stateNotifier);
+        setHeartbeat(0, HeartbeatFrequency.RETRY);
         connectionProvider.invalidate();
     }
 
     @Override
     public void close() {
         if (!isClosed()) {
             scheduledFuture.cancel(true);
+            scheduledExecutorService.shutdownNow();
             stateNotifier.close();
             connectionProvider.close();
             isClosed = true;
@@ -106,18 +129,44 @@ public boolean isClosed() {
         return isClosed;
     }
 
+    private void setHeartbeat(final ChangeEvent<ServerDescription> event) {
+        HeartbeatFrequency heartbeatFrequency = event.getNewValue().getState() == Unconnected
+                                                ? HeartbeatFrequency.RETRY
+                                                : HeartbeatFrequency.NORMAL;
+        long initialDelay = heartbeatFrequency.getFrequencyMS(settings);
+        setHeartbeat(initialDelay, heartbeatFrequency);
+    }
+
+    private synchronized void setHeartbeat(final long initialDelay, final HeartbeatFrequency newFrequency) {
+        if (currentFrequency != newFrequency) {
+            currentFrequency = newFrequency;
+            if (scheduledFuture != null) {
+                scheduledFuture.cancel(false);
+            }
+            scheduledFuture = scheduledExecutorService.scheduleAtFixedRate(stateNotifier, initialDelay,
+                                                                           newFrequency.getFrequencyMS(settings),
+                                                                           MILLISECONDS);
+        }
+    }
+
+    // Custom thread factory for scheduled executor service that creates daemon threads.  Otherwise,
+    // applications that neglect to close the MongoClient will not exit.
+    class DefaultThreadFactory implements ThreadFactory {
+        public Thread newThread(final Runnable runnable) {
+            Thread t = new Thread(runnable, "cluster-" + clusterId + "-" + serverAddress);
+            t.setDaemon(true);
+            return t;
+        }
+    }
+
     private final class DefaultServerStateListener implements ChangeListener<ServerDescription> {
         @Override
         public void stateChanged(final ChangeEvent<ServerDescription> event) {
             description = event.getNewValue();
             for (ChangeListener<ServerDescription> listener : changeListeners.keySet()) {
                 listener.stateChanged(event);
             }
-            if (event.getNewValue().getState() == Unconnected) {
-                scheduledExecutorService.schedule(stateNotifier, settings.getHeartbeatConnectRetryFrequency(MILLISECONDS),
-                                                  MILLISECONDS);
-            }
+            setHeartbeat(event);
         }
-
     }
 }
diff --git a/src/main/com/mongodb/ServerStateNotifier.java b/src/main/com/mongodb/ServerStateNotifier.java

Original file line number	Diff line number	Diff line change
`@@ -138,7 +138,6 @@ public ClusterSettings getSettings() {`
`138`	`138`	`public void close() {`
`139`	`139`	`if (!isClosed()) {`
`140`	`140`	`isClosed = true;`
`141`		`- serverFactory.close();`
`142`	`141`	`phase.get().countDown();`
`143`	`142`	`clusterListener.clusterClosed(new ClusterEvent(clusterId));`
`144`	`143`	`}`
Original file line number	Diff line number	Diff line change
`@@ -18,6 +18,4 @@`
`18`	`18`
`19`	`19`	`interface ClusterableServerFactory {`
`20`	`20`	`ClusterableServer create(ServerAddress serverAddress);`
`21`		`-`
`22`		`- void close();`
`23`	`21`	`}`