Take StateMonitor snapshot every 60s

Using Executors simplifies the code and fixes the following 2 small problems:
 - A tiny drift of the 1m interval: It starts the next snapshot 1m + the time
   before wait() return + the time until currentTimeMillis().
 - May potentially (but unlikely) invoke wait() with negative (throws
   exception) or 0 argument (waits forever):  There is no test on the returned
   long from currentTimeMillis().

1 files changed, 19 insertions, 33 deletions

diff --git a/container-core/src/main/java/com/yahoo/container/jdisc/state/StateMonitor.java b/container-core/src/main/java/com/yahoo/container/jdisc/state/StateMonitor.java
index 0018dd22dd9..066118294a0 100644
--- a/container-core/src/main/java/com/yahoo/container/jdisc/state/StateMonitor.java
+++ b/container-core/src/main/java/com/yahoo/container/jdisc/state/StateMonitor.java
@@ -7,14 +7,15 @@ import com.yahoo.container.jdisc.config.HealthMonitorConfig;
 import com.yahoo.jdisc.Timer;
 import com.yahoo.jdisc.application.MetricConsumer;
 import com.yahoo.jdisc.core.SystemTimer;
-import java.util.logging.Level;
 
 import java.util.Map;
 import java.util.TreeSet;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.logging.Level;
 import java.util.logging.Logger;
 
 /**
@@ -31,14 +32,13 @@ public class StateMonitor extends AbstractComponent {
     public enum Status {up, down, initializing}
 
     private final CopyOnWriteArrayList<StateMetricConsumer> consumers = new CopyOnWriteArrayList<>();
-    private final Thread thread;
+    private final ScheduledExecutorService executor;
     private final Timer timer;
     private final long snapshotIntervalMs;
     private volatile long lastSnapshotTimeMs;
     private volatile MetricSnapshot snapshot;
     private volatile Status status;
     private final TreeSet<String> valueNames = new TreeSet<>();
-    private final AtomicBoolean stopped = new AtomicBoolean(false);
 
     /** For testing */
     public StateMonitor() {
@@ -59,17 +59,22 @@ public class StateMonitor extends AbstractComponent {
     StateMonitor(HealthMonitorConfig config, Timer timer, ThreadFactory threadFactory) {
         this((long)(config.snapshot_interval() * TimeUnit.SECONDS.toMillis(1)),
              Status.valueOf(config.initialStatus()),
-             timer, threadFactory);
+             timer, threadFactory, true);
     }
 
     /* Public for testing only */
-    public StateMonitor(long snapshotIntervalMS, Status status, Timer timer, ThreadFactory threadFactory) {
+    public StateMonitor(long snapshotIntervalMS, Status status, Timer timer, ThreadFactory threadFactory,
+                        boolean start) {
         this.timer = timer;
         this.snapshotIntervalMs = snapshotIntervalMS;
         this.lastSnapshotTimeMs = timer.currentTimeMillis();
         this.status = status;
-        thread = threadFactory.newThread(this::run);
-        thread.start();
+        this.executor = Executors.newSingleThreadScheduledExecutor(threadFactory);
+
+        if (start) {
+            executor.scheduleAtFixedRate(this::updateSnapshot, snapshotIntervalMS,
+                    snapshotIntervalMS, TimeUnit.MILLISECONDS);
+        }
     }
 
     /** Returns a metric consumer for jDisc which will write metrics back to this */
@@ -96,28 +101,11 @@ public class StateMonitor extends AbstractComponent {
     /** Returns the interval between each metrics snapshot used by this */
     public long getSnapshotIntervalMillis() { return snapshotIntervalMs; }
 
-    /** NOTE: For unit testing only. May lead to undefined behaviour if StateMonitor thread is running simultaneously **/
-    boolean checkTime() {
+    /** NOTE: Non-private for unit testing only. **/
+    void updateSnapshot() {
         long now = timer.currentTimeMillis();
-        if (now < lastSnapshotTimeMs + snapshotIntervalMs) {
-            return false;
-        }
         snapshot = createSnapshot(lastSnapshotTimeMs, now);
         lastSnapshotTimeMs = now;
-        return true;
-    }
-
-    private void run() {
-        log.finest("StateMonitor started.");
-        try {
-            synchronized (stopped) {
-                while (!stopped.get()) {
-                    checkTime();
-                    stopped.wait((lastSnapshotTimeMs + snapshotIntervalMs) - timer.currentTimeMillis());
-                }
-            }
-        } catch (InterruptedException e) { }
-        log.finest("StateMonitor stopped.");
     }
 
     private MetricSnapshot createSnapshot(long fromMillis, long toMillis) {
@@ -148,14 +136,12 @@ public class StateMonitor extends AbstractComponent {
 
     @Override
     public void deconstruct() {
-        synchronized (stopped) {
-            stopped.set(true);
-            stopped.notifyAll();
-        }
+        executor.shutdown();
         try {
-            thread.join(5000);
+            executor.awaitTermination(5, TimeUnit.SECONDS);
         } catch (InterruptedException e) { }
-        if (thread.isAlive()) {
+
+        if (!executor.isTerminated()) {
             log.warning("StateMonitor failed to terminate within 5 seconds of interrupt signal. Ignoring.");
         }
     }