Merge pull request #1335 from nats-io/replace-timer

Replace Timer with scheduled tasks

Author: scottf

CHANGELOG.md

@@ -20,6 +20,18 @@
 ### General
 * Update repository info and cleanup readme. #1318 @scottf
 
+```
+┌─────────────────────┬───────────────────┬─────────────────┬──────────────────────────┬──────────────────┐
+│                     │             count │            time │                 msgs/sec │        bytes/sec │
+├─────────────────────┼───────────────────┼─────────────────┼──────────────────────────┼──────────────────┤
+│ PubAsync            │   50,000,000 msgs │       28:02.729 │      29,713.638 msgs/sec │      7.08 mb/sec │
+├─────────────────────┼───────────────────┼─────────────────┼──────────────────────────┼──────────────────┤
+│ SubFetch            │   49,999,991 msgs │       36:03.206 │      23,113.837 msgs/sec │      5.51 mb/sec │
+├─────────────────────┼───────────────────┼─────────────────┼──────────────────────────┼──────────────────┤
+│ SubIterate          │   50,000,000 msgs │       20:02.740 │      41,571.745 msgs/sec │      9.91 mb/sec │
+└─────────────────────┴───────────────────┴─────────────────┴──────────────────────────┴──────────────────┘
+```
+
 ## 2.21.1
 ### KV
 * KV TTL (stream max_age) versus stream duplicate_window #1301 @scottf
@@ -30,6 +42,18 @@
 ### Misc
 * Remove STAN references #1300 @scottf
 
+```
+┌─────────────────────┬───────────────────┬─────────────────┬──────────────────────────┬──────────────────┐
+│                     │             count │            time │                 msgs/sec │        bytes/sec │
+├─────────────────────┼───────────────────┼─────────────────┼──────────────────────────┼──────────────────┤
+│ PubAsync            │   50,000,000 msgs │       28:01.156 │      29,741.440 msgs/sec │      7.09 mb/sec │
+├─────────────────────┼───────────────────┼─────────────────┼──────────────────────────┼──────────────────┤
+│ SubFetch            │   50,000,000 msgs │       38:01.867 │      21,911.882 msgs/sec │      5.22 mb/sec │
+├─────────────────────┼───────────────────┼─────────────────┼──────────────────────────┼──────────────────┤
+│ SubIterate          │   50,000,000 msgs │       20:27.331 │      40,738.806 msgs/sec │      9.71 mb/sec │
+└─────────────────────┴───────────────────┴─────────────────┴──────────────────────────┴──────────────────┘
+```
+
 ## 2.21.0
 ### Core
 * Handle Server 2.10.26 returns No Responders instead of timeouts. #1292 @scottf
@@ -44,6 +68,18 @@ Main 2 11 merge safe #1294 is actually a compilation of PRs related to 2.11 feat
 * Add Message TTL Stream Configuration #1280
 * Per Message TTL Support for 2.11 #1295
 
+```
+┌─────────────────────┬───────────────────┬─────────────────┬──────────────────────────┬──────────────────┐
+│                     │             count │            time │                 msgs/sec │        bytes/sec │
+├─────────────────────┼───────────────────┼─────────────────┼──────────────────────────┼──────────────────┤
+│ PubAsync            │   50,000,000 msgs │       29:10.069 │      28,570.302 msgs/sec │      6.81 mb/sec │
+├─────────────────────┼───────────────────┼─────────────────┼──────────────────────────┼──────────────────┤
+│ SubFetch            │   49,999,993 msgs │       36:15.122 │      22,987.213 msgs/sec │      5.48 mb/sec │
+├─────────────────────┼───────────────────┼─────────────────┼──────────────────────────┼──────────────────┤
+│ SubIterate          │   50,000,000 msgs │       21:02.180 │      39,614.001 msgs/sec │      9.44 mb/sec │
+└─────────────────────┴───────────────────┴─────────────────┴──────────────────────────┴──────────────────┘
+```
+
 ## 2.20.6
 ### Core
 * Reader Listener #1265 @scottf

build.gradle

@@ -85,7 +85,11 @@ test {
         showStandardStreams = true
     }
     if (isLocal) {
-        failFast = true
+        retry {
+            failOnPassedAfterRetry = false
+            maxFailures = 2
+            maxRetries = 2
+        }
     }
     else {
         retry {

src/main/java/io/nats/client/NatsSystemClock.java

@@ -13,17 +13,29 @@
 
 package io.nats.client;
 
-public class NatsSystemClock {
+public final class NatsSystemClock {
     private static NatsSystemClockProvider PROVIDER = new NatsSystemClockProvider() {};
 
+    /**
+     * Set the provider. Null will reset to system default
+     * @param provider the provider
+     */
     public static void setProvider(final NatsSystemClockProvider provider) {
         PROVIDER = provider == null ? new NatsSystemClockProvider() {} : provider;
     }
 
+    /**
+     * Get the current milliseconds from the provider
+     * @return the milliseconds
+     */
     public static long currentTimeMillis() {
         return PROVIDER.currentTimeMillis();
     }
 
+    /**
+     * Get the current nano time from the provider
+     * @return the nano time
+     */
     public static long nanoTime() {
         return PROVIDER.nanoTime();
     }

src/main/java/io/nats/client/Options.java

@@ -504,6 +504,11 @@ public class Options {
      * {@link Builder#executor(ExecutorService) executor}.
      */
     public static final String PROP_EXECUTOR_SERVICE_CLASS = "executor.service.class";
+    /**
+     * Property used to set class name for the Executor Service (executor) class
+     * {@link Builder#executor(ExecutorService) executor}.
+     */
+    public static final String PROP_SCHEDULED_EXECUTOR_SERVICE_CLASS = "scheduled.executor.service.class";
     /**
      * Property used to set class name for the Connect Thread Factory
      * {@link Builder#connectThreadFactory(ThreadFactory) connectThreadFactory}.
@@ -660,14 +665,15 @@ public class Options {
     private final ErrorListener errorListener;
     private final TimeTraceLogger timeTraceLogger;
     private final ConnectionListener connectionListener;
-    private ReadListener readListener;
+    private final ReadListener readListener;
     private final StatisticsCollector statisticsCollector;
     private final String dataPortType;
 
     private final boolean trackAdvancedStats;
     private final boolean traceConnection;
 
     private final ExecutorService executor;
+    private final ScheduledExecutorService scheduledExecutor;
     private final ThreadFactory connectThreadFactory;
     private final ThreadFactory callbackThreadFactory;
     private final ServerPool serverPool;
@@ -808,6 +814,7 @@ public static class Builder {
         private StatisticsCollector statisticsCollector = null;
         private String dataPortType = DEFAULT_DATA_PORT_TYPE;
         private ExecutorService executor;
+        private ScheduledExecutorService scheduledExecutor;
         private ThreadFactory connectThreadFactory;
         private ThreadFactory callbackThreadFactory;
         private List<java.util.function.Consumer<HttpRequest>> httpRequestInterceptors;
@@ -939,6 +946,7 @@ public Builder properties(Properties props) {
             classnameProperty(props, PROP_SERVERS_POOL_IMPLEMENTATION_CLASS, o -> this.serverPool = (ServerPool) o);
             classnameProperty(props, PROP_DISPATCHER_FACTORY_CLASS, o -> this.dispatcherFactory = (DispatcherFactory) o);
             classnameProperty(props, PROP_EXECUTOR_SERVICE_CLASS, o -> this.executor = (ExecutorService) o);
+            classnameProperty(props, PROP_SCHEDULED_EXECUTOR_SERVICE_CLASS, o -> this.scheduledExecutor = (ScheduledExecutorService) o);
             classnameProperty(props, PROP_CONNECT_THREAD_FACTORY_CLASS, o -> this.connectThreadFactory = (ThreadFactory) o);
             classnameProperty(props, PROP_CALLBACK_THREAD_FACTORY_CLASS, o -> this.callbackThreadFactory = (ThreadFactory) o);
             return this;
@@ -1630,6 +1638,19 @@ public Builder executor(ExecutorService executor) {
             return this;
         }
 
+        /**
+         * Set the {@link ScheduledExecutorService ScheduledExecutorService} used to run scheduled task like
+         * heartbeat timers
+         * The default is a ScheduledThreadPoolExecutor that does not
+         *  execute delayed tasks after shutdown and removes tasks on cancel;
+         * @param scheduledExecutor The ScheduledExecutorService to use for timer tasks
+         * @return the Builder for chaining
+         */
+        public Builder scheduledExecutor(ScheduledExecutorService scheduledExecutor) {
+            this.scheduledExecutor = scheduledExecutor;
+            return this;
+        }
+
         /**
          * Sets custom thread factory for the executor service
          *
@@ -1910,6 +1931,17 @@ else if (useDefaultTls) {
                     new DefaultThreadFactory(threadPrefix));
             }
 
+            if (this.scheduledExecutor == null) {
+                String threadPrefix = nullOrEmpty(this.connectionName) ? DEFAULT_THREAD_NAME_PREFIX : this.connectionName;
+                // the core pool size of 3 is chosen considering where we know the scheduler is used.
+                // 1. Ping timer, 2. cleanup timer, 3. SocketDataPortWithWriteTimeout
+                // Pull message managers also use a scheduler, but we don't even know if this will be consuming
+                ScheduledThreadPoolExecutor stpe = new ScheduledThreadPoolExecutor(3, new DefaultThreadFactory(threadPrefix));
+                stpe.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
+                stpe.setRemoveOnCancelPolicy(true);
+                this.scheduledExecutor = stpe;
+            }
+
             if (socketReadTimeoutMillis > 0) {
                 long srtMin = pingInterval.toMillis() + MINIMUM_SOCKET_WRITE_TIMEOUT_GT_CONNECTION_TIMEOUT;
                 if (socketReadTimeoutMillis < srtMin) {
@@ -2014,6 +2046,7 @@ public Builder(Options o) {
             this.dataPortType = o.dataPortType;
             this.trackAdvancedStats = o.trackAdvancedStats;
             this.executor = o.executor;
+            this.scheduledExecutor = o.scheduledExecutor;
             this.callbackThreadFactory = o.callbackThreadFactory;
             this.connectThreadFactory = o.connectThreadFactory;
             this.httpRequestInterceptors = o.httpRequestInterceptors;
@@ -2082,6 +2115,7 @@ private Options(Builder b) {
         this.dataPortType = b.dataPortType;
         this.trackAdvancedStats = b.trackAdvancedStats;
         this.executor = b.executor;
+        this.scheduledExecutor = b.scheduledExecutor;
         this.callbackThreadFactory = b.callbackThreadFactory;
         this.connectThreadFactory = b.connectThreadFactory;
         this.httpRequestInterceptors = b.httpRequestInterceptors;
@@ -2107,6 +2141,13 @@ public ExecutorService getExecutor() {
         return this.executor;
     }
 
+    /**
+     * @return the ScheduledExecutorService, see {@link Builder#scheduledExecutor(ScheduledExecutorService) scheduledExecutor()} in the builder doc
+     */
+    public ScheduledExecutorService getScheduledExecutor() {
+        return scheduledExecutor;
+    }
+
     /**
      * @return the callback executor, see {@link Builder#callbackThreadFactory(ThreadFactory) callbackThreadFactory()} in the builder doc
      */

src/main/java/io/nats/client/impl/MessageManager.java

@@ -18,11 +18,10 @@
 import io.nats.client.PullRequestOptions;
 import io.nats.client.SubscribeOptions;
 import io.nats.client.support.NatsConstants;
+import io.nats.client.support.ScheduledTask;
 
 import java.time.Duration;
-import java.util.Timer;
-import java.util.TimerTask;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantLock;
 
@@ -46,8 +45,8 @@ public enum ManageResult {MESSAGE, STATUS_HANDLED, STATUS_TERMINUS, STATUS_ERROR
     protected boolean hb;
     protected long idleHeartbeatSetting;
     protected long alarmPeriodSettingNanos;
-    protected MmTimerTask heartbeatTimerTask;
-    protected Timer heartbeatTimer;
+    protected ScheduledTask heartbeatTask;
+    protected final AtomicLong currentAlarmPeriodNanos;
 
     protected MessageManager(NatsConnection conn, SubscribeOptions so, boolean syncMode) {
         stateChangeLock = new ReentrantLock();
@@ -62,6 +61,7 @@ protected MessageManager(NatsConnection conn, SubscribeOptions so, boolean syncM
         idleHeartbeatSetting = 0;
         alarmPeriodSettingNanos = 0;
         lastMsgReceivedNanoTime = new AtomicLong(NatsSystemClock.nanoTime());
+        currentAlarmPeriodNanos = new AtomicLong();
     }
 
     protected boolean isSyncMode()              { return syncMode; }
@@ -134,62 +134,29 @@ protected void updateLastMessageReceived() {
         lastMsgReceivedNanoTime.set(NatsSystemClock.nanoTime());
     }
 
-    class MmTimerTask extends TimerTask {
-        long alarmPeriodNanos;
-        final AtomicBoolean alive;
-
-        public MmTimerTask(long alarmPeriodNanos) {
-            this.alarmPeriodNanos = alarmPeriodNanos;
-            alive = new AtomicBoolean(true);
-        }
-
-        public void reuse() {
-            alive.getAndSet(true);
-        }
-
-        public void shutdown() {
-            alive.getAndSet(false);
-        }
-
-        @Override
-        public void run() {
-            if (alive.get() && !Thread.interrupted()) {
-                long sinceLast = NatsSystemClock.nanoTime() - lastMsgReceivedNanoTime.get();
-                if (alive.get() && sinceLast > alarmPeriodNanos) {
-                    handleHeartbeatError();
-                }
-            }
-        }
-
-        @Override
-        public String toString() {
-            long sinceLastMillis = (NatsSystemClock.nanoTime() - lastMsgReceivedNanoTime.get()) / NatsConstants.NANOS_PER_MILLI;
-            return "MmTimerTask{" +
-                ", alarmPeriod=" + (alarmPeriodNanos / NatsConstants.NANOS_PER_MILLI) +
-                "ms, alive=" + alive.get() +
-                ", sinceLast=" + sinceLastMillis + "ms}";
-        }
-    }
-
     protected void initOrResetHeartbeatTimer() {
         stateChangeLock.lock();
         try {
-            if (heartbeatTimer != null) {
+            if (heartbeatTask != null) {
                 // Same settings, just reuse the existing timer
-                if (heartbeatTimerTask.alarmPeriodNanos == alarmPeriodSettingNanos) {
-                    heartbeatTimerTask.reuse();
+                if (currentAlarmPeriodNanos.get() == alarmPeriodSettingNanos) {
                     updateLastMessageReceived();
                     return;
                 }
 
                 // Replace timer since settings have changed
                 shutdownHeartbeatTimer();
             }
+
             // replacement or new comes here
-            heartbeatTimer = new Timer();
-            heartbeatTimerTask = new MmTimerTask(alarmPeriodSettingNanos);
-            long alarmPeriodSettingMillis = alarmPeriodSettingNanos / NatsConstants.NANOS_PER_MILLI;
-            heartbeatTimer.schedule(heartbeatTimerTask, alarmPeriodSettingMillis, alarmPeriodSettingMillis);
+            this.currentAlarmPeriodNanos.set(alarmPeriodSettingNanos);
+            heartbeatTask = new ScheduledTask(conn.getScheduledExecutor(), alarmPeriodSettingNanos, TimeUnit.NANOSECONDS,
+                () -> {
+                    long sinceLast = NatsSystemClock.nanoTime() - lastMsgReceivedNanoTime.get();
+                    if (sinceLast > currentAlarmPeriodNanos.get()) {
+                        handleHeartbeatError();
+                    }
+                });
             updateLastMessageReceived();
         }
         finally {
@@ -200,12 +167,11 @@ protected void initOrResetHeartbeatTimer() {
     protected void shutdownHeartbeatTimer() {
         stateChangeLock.lock();
         try {
-            if (heartbeatTimer != null) {
-                heartbeatTimerTask.shutdown();
-                heartbeatTimerTask = null;
-                heartbeatTimer.cancel();
-                heartbeatTimer = null;
+            if (heartbeatTask != null) {
+                heartbeatTask.shutdown();
+                heartbeatTask = null;
             }
+            currentAlarmPeriodNanos.set(0);
         }
         finally {
             stateChangeLock.unlock();