Replace old NanoTime object with Timer

Author: DomGarguilo

core/src/main/java/org/apache/accumulo/core/clientImpl/TabletServerBatchWriter.java

@@ -77,9 +77,9 @@
 import org.apache.accumulo.core.tabletserver.thrift.NoSuchScanIDException;
 import org.apache.accumulo.core.trace.TraceUtil;
 import org.apache.accumulo.core.util.Retry;
+import org.apache.accumulo.core.util.Timer;
 import org.apache.accumulo.core.util.threads.ThreadPools;
 import org.apache.accumulo.core.util.threads.Threads;
-import org.apache.accumulo.core.util.time.NanoTime;
 import org.apache.thrift.TApplicationException;
 import org.apache.thrift.TException;
 import org.apache.thrift.TServiceClient;
@@ -1096,7 +1096,7 @@ private void cancelSession() throws InterruptedException, ThriftSecurityExceptio
 
         final HostAndPort parsedServer = HostAndPort.fromString(location);
 
-        NanoTime startTime = NanoTime.now();
+        Timer timer = Timer.startNew();
 
         boolean useCloseUpdate = false;
 
@@ -1170,7 +1170,7 @@ private void cancelSession() throws InterruptedException, ThriftSecurityExceptio
           }
 
           // if a timeout is set on the batch writer, then do not retry longer than the timeout
-          if (startTime.elapsed().toMillis() > timeout) {
+          if (timer.hasElapsed(timeout, MILLISECONDS)) {
             log.debug("Giving up on canceling session {} {} and timing out.", location, usid);
             throw new TimedOutException(Set.of(location));
           }

core/src/main/java/org/apache/accumulo/core/fate/zookeeper/DistributedReadWriteLock.java

@@ -22,7 +22,6 @@
 import static java.util.concurrent.TimeUnit.DAYS;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
-import java.time.Duration;
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.Map.Entry;
@@ -31,8 +30,8 @@
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
 
+import org.apache.accumulo.core.util.Timer;
 import org.apache.accumulo.core.util.UtilWaitThread;
-import org.apache.accumulo.core.util.time.NanoTime;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -182,9 +181,8 @@ public boolean tryLock() {
 
     @Override
     public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
-      Duration returnTime = Duration.of(time, unit.toChronoUnit());
-      NanoTime start = NanoTime.now();
-      while (start.elapsed().compareTo(returnTime) < 0) {
+      Timer timer = Timer.startNew();
+      while (timer.hasElapsed(time, unit)) {
         if (tryLock()) {
           return true;
         }

core/src/main/java/org/apache/accumulo/core/util/Retry.java

@@ -22,7 +22,6 @@
 
 import java.time.Duration;
 
-import org.apache.accumulo.core.util.time.NanoTime;
 import org.slf4j.Logger;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -42,9 +41,8 @@ public class Retry {
   private Duration currentWait;
   private Duration initialWait;
 
-  private boolean hasNeverLogged;
   private boolean hasLoggedWarn = false;
-  private NanoTime lastRetryLog;
+  private Timer lastRetryLogTimer;
   private double currentBackOffFactor;
   private boolean doTimeJitter = true;
 
@@ -64,8 +62,7 @@ private Retry(long maxRetries, Duration startWait, Duration waitIncrement, Durat
     this.currentWait = startWait;
     this.initialWait = startWait;
     this.logInterval = logInterval;
-    this.hasNeverLogged = true;
-    this.lastRetryLog = null;
+    this.lastRetryLogTimer = null;
     this.backOffFactor = backOffFactor;
     this.currentBackOffFactor = this.backOffFactor;
 
@@ -202,16 +199,14 @@ protected void sleep(Duration wait) throws InterruptedException {
 
   public void logRetry(Logger log, String message, Throwable t) {
     // log the first time as debug, and then after every logInterval as a warning
-    NanoTime now = NanoTime.now();
-    if (hasNeverLogged) {
+    if (lastRetryLogTimer == null) {
       if (log.isDebugEnabled()) {
         log.debug(getMessage(message, t));
       }
-      hasNeverLogged = false;
-      lastRetryLog = now;
-    } else if (now.subtract(lastRetryLog).compareTo(logInterval) > 0) {
+      lastRetryLogTimer = Timer.startNew();
+    } else if (lastRetryLogTimer.hasElapsed(logInterval)) {
       log.warn(getMessage(message), t);
-      lastRetryLog = now;
+      lastRetryLogTimer.restart();
       hasLoggedWarn = true;
     } else {
       if (log.isTraceEnabled()) {
@@ -222,16 +217,14 @@ public void logRetry(Logger log, String message, Throwable t) {
 
   public void logRetry(Logger log, String message) {
     // log the first time as debug, and then after every logInterval as a warning
-    NanoTime now = NanoTime.now();
-    if (hasNeverLogged) {
+    if (lastRetryLogTimer == null) {
       if (log.isDebugEnabled()) {
         log.debug(getMessage(message));
       }
-      hasNeverLogged = false;
-      lastRetryLog = now;
-    } else if (now.subtract(lastRetryLog).compareTo(logInterval) > 0) {
+      lastRetryLogTimer = Timer.startNew();
+    } else if (lastRetryLogTimer.hasElapsed(logInterval)) {
       log.warn(getMessage(message));
-      lastRetryLog = now;
+      lastRetryLogTimer.restart();
       hasLoggedWarn = true;
     } else {
       if (log.isTraceEnabled()) {
@@ -251,14 +244,15 @@ private String getMessage(String message, Throwable t) {
   }
 
   public void logCompletion(Logger log, String operationDescription) {
-    if (!hasNeverLogged) {
-      var message = operationDescription + " completed after " + (retriesDone + 1)
-          + " retries and is no longer retrying.";
-      if (hasLoggedWarn) {
-        log.info(message);
-      } else {
-        log.debug(message);
-      }
+    if (lastRetryLogTimer == null) { // have never logged a retry
+      return;
+    }
+    var message = operationDescription + " completed after " + (retriesDone + 1)
+        + " retries and is no longer retrying.";
+    if (hasLoggedWarn) {
+      log.info(message);
+    } else {
+      log.debug(message);
     }
   }
 

server/base/src/main/java/org/apache/accumulo/server/mem/LowMemoryDetector.java

@@ -18,8 +18,11 @@
  */
 package org.apache.accumulo.server.mem;
 
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
 import java.lang.management.GarbageCollectorMXBean;
 import java.lang.management.ManagementFactory;
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.List;
 import java.util.concurrent.locks.Lock;
@@ -29,7 +32,7 @@
 import org.apache.accumulo.core.conf.AccumuloConfiguration;
 import org.apache.accumulo.core.conf.Property;
 import org.apache.accumulo.core.util.Halt;
-import org.apache.accumulo.core.util.time.NanoTime;
+import org.apache.accumulo.core.util.Timer;
 import org.apache.accumulo.server.ServerContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -51,7 +54,7 @@ public enum DetectionScope {
 
   private long lastMemorySize = 0;
   private int lowMemCount = 0;
-  private NanoTime lastMemoryCheckTime = null;
+  private Timer lastMemoryCheckTime = null;
   private final Lock memCheckTimeLock = new ReentrantLock();
   private volatile boolean runningLowOnMemory = false;
 
@@ -104,7 +107,7 @@ public void logGCInfo(AccumuloConfiguration conf) {
 
     memCheckTimeLock.lock();
     try {
-      final NanoTime now = NanoTime.now();
+      final Timer currentTimer = Timer.startNew();
 
       List<GarbageCollectorMXBean> gcmBeans = ManagementFactory.getGarbageCollectorMXBeans();
 
@@ -173,25 +176,24 @@ public void logGCInfo(AccumuloConfiguration conf) {
         LOG.debug(sb.toString());
       }
 
-      final long keepAliveTimeout = conf.getTimeInMillis(Property.INSTANCE_ZK_TIMEOUT);
-      if (lastMemoryCheckTime != null && lastMemoryCheckTime.compareTo(now) < 0) {
-        final long diff = now.subtract(lastMemoryCheckTime).toMillis();
-        if (diff > keepAliveTimeout + 1000) {
+      final Duration keepAliveTimeout = conf.getDuration(Property.INSTANCE_ZK_TIMEOUT);
+      if (lastMemoryCheckTime != null && lastMemoryCheckTime.hasElapsed(keepAliveTimeout)) {
+        if (currentTimer.hasElapsed(keepAliveTimeout.plus(Duration.ofSeconds(1)))) {
           LOG.warn(String.format(
               "GC pause checker not called in a timely"
                   + " fashion. Expected every %.1f seconds but was %.1f seconds since last check",
-              keepAliveTimeout / 1000., diff / 1000.));
+              keepAliveTimeout.toMillis() / 1000., currentTimer.elapsed(MILLISECONDS) / 1000.));
         }
-        lastMemoryCheckTime = now;
+        lastMemoryCheckTime = currentTimer;
         return;
       }
 
-      if (maxIncreaseInCollectionTime > keepAliveTimeout) {
+      if (maxIncreaseInCollectionTime > keepAliveTimeout.toMillis()) {
         Halt.halt("Garbage collection may be interfering with lock keep-alive. Halting.", -1);
       }
 
       lastMemorySize = freeMemory;
-      lastMemoryCheckTime = now;
+      lastMemoryCheckTime = currentTimer;
     } finally {
       memCheckTimeLock.unlock();
     }

server/base/src/main/java/org/apache/accumulo/server/rpc/TimedProcessor.java

@@ -18,8 +18,10 @@
  */
 package org.apache.accumulo.server.rpc;
 
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
 import org.apache.accumulo.core.metrics.MetricsInfo;
-import org.apache.accumulo.core.util.time.NanoTime;
+import org.apache.accumulo.core.util.Timer;
 import org.apache.accumulo.server.metrics.ThriftMetrics;
 import org.apache.thrift.TException;
 import org.apache.thrift.TProcessor;
@@ -32,25 +34,25 @@ public class TimedProcessor implements TProcessor {
 
   private final TProcessor other;
   private final ThriftMetrics thriftMetrics;
-  private NanoTime idleStart;
+  private final Timer idleTimer;
 
   public TimedProcessor(final TProcessor next, final MetricsInfo metricsInfo) {
     this.other = next;
     thriftMetrics = new ThriftMetrics();
     metricsInfo.addMetricsProducers(thriftMetrics);
-    idleStart = NanoTime.now();
+    idleTimer = Timer.startNew();
   }
 
   @Override
   public void process(TProtocol in, TProtocol out) throws TException {
-    NanoTime processStart = NanoTime.now();
-    thriftMetrics.addIdle(processStart.subtract(idleStart).toMillis());
+    thriftMetrics.addIdle(idleTimer.elapsed(MILLISECONDS));
+    Timer processTimer = Timer.startNew();
     try {
       other.process(in, out);
     } finally {
-      // set idle to now, calc time in process
-      idleStart = NanoTime.now();
-      thriftMetrics.addExecute(idleStart.subtract(processStart).toMillis());
+      // calc time in process, restart idle timer
+      thriftMetrics.addExecute(processTimer.elapsed(MILLISECONDS));
+      idleTimer.restart();
     }
   }
 }

server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java

@@ -20,6 +20,8 @@
 
 import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.concurrent.TimeUnit.MINUTES;
+import static java.util.concurrent.TimeUnit.SECONDS;
 import static java.util.stream.Collectors.toList;
 
 import java.io.ByteArrayInputStream;
@@ -92,7 +94,7 @@
 import org.apache.accumulo.core.tabletserver.thrift.TabletStats;
 import org.apache.accumulo.core.trace.TraceUtil;
 import org.apache.accumulo.core.util.Pair;
-import org.apache.accumulo.core.util.time.NanoTime;
+import org.apache.accumulo.core.util.Timer;
 import org.apache.accumulo.core.volume.Volume;
 import org.apache.accumulo.server.ServerContext;
 import org.apache.accumulo.server.compaction.CompactionStats;
@@ -1032,7 +1034,7 @@ synchronized void completeClose(boolean saveState, boolean completeClose) throws
       return currentlyUnreserved;
     });
 
-    long lastLogTime = System.nanoTime();
+    Timer lastLogTimer = Timer.startNew();
 
     // wait for reads and writes to complete
     while (writesInProgress > 0 || !runningScans.isEmpty()) {
@@ -1044,12 +1046,12 @@ synchronized void completeClose(boolean saveState, boolean completeClose) throws
         return currentlyUnreserved;
       });
 
-      if (log.isDebugEnabled() && System.nanoTime() - lastLogTime > TimeUnit.SECONDS.toNanos(60)) {
+      if (log.isDebugEnabled() && lastLogTimer.hasElapsed(1, MINUTES)) {
         for (ScanDataSource activeScan : runningScans) {
           log.debug("Waiting on scan in completeClose {} {}", extent, activeScan);
         }
 
-        lastLogTime = System.nanoTime();
+        lastLogTimer.restart();
       }
 
       try {
@@ -1770,7 +1772,7 @@ public void importDataFiles(long tid, Map<ReferencedTabletFile,DataFileInfo> fil
 
     // Clients timeout and will think that this operation failed.
     // Don't do it if we spent too long waiting for the lock
-    NanoTime now = NanoTime.now();
+    Timer lockWaitTimer = Timer.startNew();
     synchronized (this) {
       if (isClosed()) {
         throw new IOException("tablet " + extent + " is closed");
@@ -1789,9 +1791,8 @@ public void importDataFiles(long tid, Map<ReferencedTabletFile,DataFileInfo> fil
           throw new IllegalStateException(e);
         }
 
-        Duration lockWait = now.elapsed();
-        if (lockWait.compareTo(rpcTimeout) > 0) {
-          throw new IOException("Timeout waiting " + lockWait.toSeconds()
+        if (lockWaitTimer.hasElapsed(rpcTimeout)) {
+          throw new IOException("Timeout waiting " + lockWaitTimer.elapsed(SECONDS)
               + " seconds to get tablet lock for " + extent + " " + tid);
         }
       }
@@ -1801,9 +1802,8 @@ public void importDataFiles(long tid, Map<ReferencedTabletFile,DataFileInfo> fil
         throw new IOException("tablet " + extent + " is closed");
       }
 
-      Duration lockWait = now.elapsed();
-      if (lockWait.compareTo(rpcTimeout) > 0) {
-        throw new IOException("Timeout waiting " + lockWait.toSeconds()
+      if (lockWaitTimer.hasElapsed(rpcTimeout)) {
+        throw new IOException("Timeout waiting " + lockWaitTimer.elapsed(SECONDS)
             + " seconds to get tablet lock for " + extent + " " + tid);
       }