ZooStore deferral time to use System.nanoTime()

- ZooStore() now uses System.nanoTime() instead of System.currentTimeMillis() - Added TimeUnit param to unreserve() - Renamed 'defered' -> 'deferred'
kevinrr888 · Jan 3, 2024 · 1b885d8 · 1b885d8
1 parent d973179
commit 1b885d8
Show file tree

Hide file tree

Showing 9 changed files with 50 additions and 38 deletions.
diff --git a/core/src/main/java/org/apache/accumulo/core/fate/AdminUtil.java b/core/src/main/java/org/apache/accumulo/core/fate/AdminUtil.java
@@ -32,6 +32,7 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.accumulo.core.fate.ReadOnlyTStore.TStatus;
 import org.apache.accumulo.core.fate.zookeeper.FateLock;
@@ -367,7 +368,7 @@ private FateStatus getTransactionStatus(ReadOnlyTStore<T> zs, Set<Long> filterTx
 
       long timeCreated = zs.timeCreated(tid);
 
-      zs.unreserve(tid, 0);
+      zs.unreserve(tid, 0, TimeUnit.MILLISECONDS);
 
       if (includeByStatus(status, filterStatus) && includeByTxid(tid, filterTxid)) {
         statuses.add(new TransactionStatus(tid, status, txName, hlocks, wlocks, top, timeCreated));
@@ -450,7 +451,7 @@ public boolean prepDelete(TStore<T> zs, ZooReaderWriter zk, ServiceLockPath path
         break;
     }
 
-    zs.unreserve(txid, 0);
+    zs.unreserve(txid, 0, TimeUnit.MILLISECONDS);
     return state;
   }
 
@@ -494,7 +495,7 @@ public boolean prepFail(TStore<T> zs, ZooReaderWriter zk, ServiceLockPath zLockM
         break;
     }
 
-    zs.unreserve(txid, 0);
+    zs.unreserve(txid, 0, TimeUnit.MILLISECONDS);
     return state;
   }
 

diff --git a/core/src/main/java/org/apache/accumulo/core/fate/AgeOffStore.java b/core/src/main/java/org/apache/accumulo/core/fate/AgeOffStore.java
@@ -25,6 +25,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.concurrent.TimeUnit;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -107,7 +108,7 @@ public void ageOff() {
           }
 
         } finally {
-          store.unreserve(txid, 0);
+          store.unreserve(txid, 0, TimeUnit.MILLISECONDS);
         }
       } catch (Exception e) {
         log.warn("Failed to age off FATE tx " + FateTxId.formatTid(txid), e);
@@ -137,7 +138,7 @@ public AgeOffStore(ZooStore<T> store, long ageOffTime, TimeSource timeSource) {
             break;
         }
       } finally {
-        store.unreserve(txid, 0);
+        store.unreserve(txid, 0, TimeUnit.MILLISECONDS);
       }
     }
   }
@@ -165,8 +166,8 @@ public boolean tryReserve(long tid) {
   }
 
   @Override
-  public void unreserve(long tid, long deferTime) {
-    store.unreserve(tid, deferTime);
+  public void unreserve(long tid, long deferTime, TimeUnit deferTimeUnit) {
+    store.unreserve(tid, deferTime, deferTimeUnit);
   }
 
   @Override

diff --git a/core/src/main/java/org/apache/accumulo/core/fate/Fate.java b/core/src/main/java/org/apache/accumulo/core/fate/Fate.java
@@ -34,6 +34,7 @@
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 
@@ -131,7 +132,7 @@ public void run() {
           runnerLog.error("Uncaught exception in FATE runner thread.", e);
         } finally {
           if (tid != null) {
-            store.unreserve(tid, deferTime);
+            store.unreserve(tid, deferTime, TimeUnit.MILLISECONDS);
           }
         }
       }
@@ -295,7 +296,7 @@ public void seedTransaction(String txName, long tid, Repo<T> repo, boolean autoC
         store.setStatus(tid, SUBMITTED);
       }
     } finally {
-      store.unreserve(tid, 0);
+      store.unreserve(tid, 0, TimeUnit.MILLISECONDS);
     }
 
   }
@@ -331,7 +332,7 @@ public boolean cancel(long tid) {
             return false;
           }
         } finally {
-          store.unreserve(tid, 0);
+          store.unreserve(tid, 0, TimeUnit.MILLISECONDS);
         }
       } else {
         // reserved, lets retry.
@@ -362,7 +363,7 @@ public void delete(long tid) {
           break;
       }
     } finally {
-      store.unreserve(tid, 0);
+      store.unreserve(tid, 0, TimeUnit.MILLISECONDS);
     }
   }
 
@@ -375,7 +376,7 @@ public String getReturn(long tid) {
       }
       return (String) store.getTransactionInfo(tid, TxInfo.RETURN_VALUE);
     } finally {
-      store.unreserve(tid, 0);
+      store.unreserve(tid, 0, TimeUnit.MILLISECONDS);
     }
   }
 
@@ -389,7 +390,7 @@ public Exception getException(long tid) {
       }
       return (Exception) store.getTransactionInfo(tid, TxInfo.EXCEPTION);
     } finally {
-      store.unreserve(tid, 0);
+      store.unreserve(tid, 0, TimeUnit.MILLISECONDS);
     }
   }
 

diff --git a/core/src/main/java/org/apache/accumulo/core/fate/ReadOnlyTStore.java b/core/src/main/java/org/apache/accumulo/core/fate/ReadOnlyTStore.java
@@ -21,6 +21,7 @@
 import java.io.Serializable;
 import java.util.EnumSet;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Read only access to a Transaction Store.
@@ -76,10 +77,11 @@ enum TStatus {
    * longer interact with it.
    *
    * @param tid transaction id, previously reserved.
-   * @param deferTime time in millis to keep this transaction out of the pool used in the
-   *        {@link #reserve() reserve} method. must be non-negative.
+   * @param deferTime time to keep this transaction out of the pool used in the {@link #reserve()
+   *        reserve} method. must be non-negative.
+   * @param deferTimeUnit the time unit of deferTime
    */
-  void unreserve(long tid, long deferTime);
+  void unreserve(long tid, long deferTime, TimeUnit deferTimeUnit);
 
   /**
    * Get the current operation for the given transaction id.

diff --git a/core/src/main/java/org/apache/accumulo/core/fate/ZooStore.java b/core/src/main/java/org/apache/accumulo/core/fate/ZooStore.java
@@ -37,6 +37,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
 import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeExistsPolicy;
@@ -61,7 +62,7 @@ public class ZooStore<T> implements TStore<T> {
   private ZooReaderWriter zk;
   private String lastReserved = "";
   private Set<Long> reserved;
-  private Map<Long,Long> defered;
+  private Map<Long,Long> deferred;
   private static final SecureRandom random = new SecureRandom();
   private long statusChangeEvents = 0;
   private int reservationsWaiting = 0;
@@ -106,7 +107,7 @@ public ZooStore(String path, ZooReaderWriter zk) throws KeeperException, Interru
     this.path = path;
     this.zk = zk;
     this.reserved = new HashSet<>();
-    this.defered = new HashMap<>();
+    this.deferred = new HashMap<>();
 
     zk.putPersistentData(path, new byte[0], NodeExistsPolicy.SKIP);
   }
@@ -163,9 +164,9 @@ public long reserve() {
               continue;
             }
 
-            if (defered.containsKey(tid)) {
-              if (defered.get(tid) < System.currentTimeMillis()) {
-                defered.remove(tid);
+            if (deferred.containsKey(tid)) {
+              if ((deferred.get(tid) - System.nanoTime()) < 0) {
+                deferred.remove(tid);
               } else {
                 continue;
               }
@@ -200,11 +201,12 @@ public long reserve() {
         synchronized (this) {
           // suppress lgtm alert - synchronized variable is not always true
           if (events == statusChangeEvents) { // lgtm [java/constant-comparison]
-            if (defered.isEmpty()) {
+            if (deferred.isEmpty()) {
               this.wait(5000);
             } else {
-              Long minTime = Collections.min(defered.values());
-              long waitTime = minTime - System.currentTimeMillis();
+              Long minTime = Collections.min(deferred.values());
+              long waitTime = minTime - System.nanoTime();
+              waitTime = TimeUnit.MILLISECONDS.convert(waitTime, TimeUnit.NANOSECONDS);
               if (waitTime > 0) {
                 this.wait(Math.min(waitTime, 5000));
               }
@@ -271,7 +273,8 @@ private void unreserve(long tid) {
   }
 
   @Override
-  public void unreserve(long tid, long deferTime) {
+  public void unreserve(long tid, long deferTime, TimeUnit deferTimeUnit) {
+    deferTime = TimeUnit.NANOSECONDS.convert(deferTime, deferTimeUnit);
 
     if (deferTime < 0) {
       throw new IllegalArgumentException("deferTime < 0 : " + deferTime);
@@ -284,7 +287,7 @@ public void unreserve(long tid, long deferTime) {
       }
 
       if (deferTime > 0) {
-        defered.put(tid, System.currentTimeMillis() + deferTime);
+        deferred.put(tid, System.nanoTime() + deferTime);
       }
 
       this.notifyAll();

diff --git a/core/src/main/java/org/apache/accumulo/core/logging/FateLogger.java b/core/src/main/java/org/apache/accumulo/core/logging/FateLogger.java
@@ -23,6 +23,7 @@
 import java.io.Serializable;
 import java.util.EnumSet;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 
 import org.apache.accumulo.core.fate.Fate;
@@ -61,8 +62,8 @@ public boolean tryReserve(long tid) {
       }
 
       @Override
-      public void unreserve(long tid, long deferTime) {
-        store.unreserve(tid, deferTime);
+      public void unreserve(long tid, long deferTime, TimeUnit deferTimeUnit) {
+        store.unreserve(tid, deferTime, deferTimeUnit);
       }
 
       @Override

diff --git a/core/src/test/java/org/apache/accumulo/core/fate/AgeOffStoreTest.java b/core/src/test/java/org/apache/accumulo/core/fate/AgeOffStoreTest.java
@@ -22,6 +22,7 @@
 
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.accumulo.core.fate.AgeOffStore.TimeSource;
 import org.apache.accumulo.core.fate.ReadOnlyTStore.TStatus;
@@ -52,23 +53,23 @@ public void testBasic() throws InterruptedException, KeeperException {
     long txid1 = aoStore.create();
     aoStore.reserve(txid1);
     aoStore.setStatus(txid1, TStatus.IN_PROGRESS);
-    aoStore.unreserve(txid1, 0);
+    aoStore.unreserve(txid1, 0, TimeUnit.MILLISECONDS);
 
     aoStore.ageOff();
 
     long txid2 = aoStore.create();
     aoStore.reserve(txid2);
     aoStore.setStatus(txid2, TStatus.IN_PROGRESS);
     aoStore.setStatus(txid2, TStatus.FAILED);
-    aoStore.unreserve(txid2, 0);
+    aoStore.unreserve(txid2, 0, TimeUnit.MILLISECONDS);
 
     tts.time = 6;
 
     long txid3 = aoStore.create();
     aoStore.reserve(txid3);
     aoStore.setStatus(txid3, TStatus.IN_PROGRESS);
     aoStore.setStatus(txid3, TStatus.SUCCESSFUL);
-    aoStore.unreserve(txid3, 0);
+    aoStore.unreserve(txid3, 0, TimeUnit.MILLISECONDS);
 
     Long txid4 = aoStore.create();
 
@@ -101,19 +102,19 @@ public void testNonEmpty() throws InterruptedException, KeeperException {
     long txid1 = testStore.create();
     testStore.reserve(txid1);
     testStore.setStatus(txid1, TStatus.IN_PROGRESS);
-    testStore.unreserve(txid1, 0);
+    testStore.unreserve(txid1, 0, TimeUnit.MILLISECONDS);
 
     long txid2 = testStore.create();
     testStore.reserve(txid2);
     testStore.setStatus(txid2, TStatus.IN_PROGRESS);
     testStore.setStatus(txid2, TStatus.FAILED);
-    testStore.unreserve(txid2, 0);
+    testStore.unreserve(txid2, 0, TimeUnit.MILLISECONDS);
 
     long txid3 = testStore.create();
     testStore.reserve(txid3);
     testStore.setStatus(txid3, TStatus.IN_PROGRESS);
     testStore.setStatus(txid3, TStatus.SUCCESSFUL);
-    testStore.unreserve(txid3, 0);
+    testStore.unreserve(txid3, 0, TimeUnit.MILLISECONDS);
 
     Long txid4 = testStore.create();
 
@@ -136,7 +137,7 @@ public void testNonEmpty() throws InterruptedException, KeeperException {
 
     aoStore.reserve(txid1);
     aoStore.setStatus(txid1, TStatus.FAILED_IN_PROGRESS);
-    aoStore.unreserve(txid1, 0);
+    aoStore.unreserve(txid1, 0, TimeUnit.MILLISECONDS);
 
     tts.time = 30;
 
@@ -147,7 +148,7 @@ public void testNonEmpty() throws InterruptedException, KeeperException {
 
     aoStore.reserve(txid1);
     aoStore.setStatus(txid1, TStatus.FAILED);
-    aoStore.unreserve(txid1, 0);
+    aoStore.unreserve(txid1, 0, TimeUnit.MILLISECONDS);
 
     aoStore.ageOff();
 

diff --git a/core/src/test/java/org/apache/accumulo/core/fate/TestStore.java b/core/src/test/java/org/apache/accumulo/core/fate/TestStore.java
@@ -24,6 +24,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Transient in memory store for transactions.
@@ -61,7 +62,7 @@ public boolean tryReserve(long tid) {
   }
 
   @Override
-  public void unreserve(long tid, long deferTime) {
+  public void unreserve(long tid, long deferTime, TimeUnit deferTimeUnit) {
     if (!reserved.remove(tid)) {
       throw new IllegalStateException();
     }

diff --git a/shell/src/test/java/org/apache/accumulo/shell/commands/FateCommandTest.java b/shell/src/test/java/org/apache/accumulo/shell/commands/FateCommandTest.java
@@ -35,6 +35,7 @@
 import java.io.PrintStream;
 import java.nio.file.Files;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.accumulo.core.client.AccumuloClient;
 import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
@@ -144,7 +145,7 @@ public void testFailTx() throws Exception {
     expectLastCall().once();
     zs.setStatus(tid, ReadOnlyTStore.TStatus.FAILED_IN_PROGRESS);
     expectLastCall().once();
-    zs.unreserve(tid, 0);
+    zs.unreserve(tid, 0, TimeUnit.MILLISECONDS);
     expectLastCall().once();
 
     TestHelper helper = new TestHelper(true);