HBASE-25913 Introduce EnvironmentEdge.Clock and Clock.currentTimeAdvancing

hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/TestBackupDelete.java

@@ -28,7 +28,7 @@
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.backup.impl.BackupSystemTable;
 import org.apache.hadoop.hbase.testclassification.LargeTests;
-import org.apache.hadoop.hbase.util.EnvironmentEdge;
+import org.apache.hadoop.hbase.util.BaseEnvironmentEdge;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.hadoop.util.ToolRunner;
 import org.junit.Assert;
@@ -112,13 +112,7 @@ public void testBackupDeleteCommand() throws Exception {
   public void testBackupPurgeOldBackupsCommand() throws Exception {
     LOG.info("test backup delete (purge old backups) on a single table with data: command-line");
     List<TableName> tableList = Lists.newArrayList(table1);
-    EnvironmentEdgeManager.injectEdge(new EnvironmentEdge() {
-      // time - 2 days
-      @Override
-      public long currentTime() {
-        return System.currentTimeMillis() - 2 * 24 * 3600 * 1000 ;
-      }
-    });
+    EnvironmentEdgeManager.injectEdge(new TwoDaysBackEnvironmentEdge());
     String backupId = fullTableBackup(tableList);
     assertTrue(checkSucceeded(backupId));
 
@@ -160,4 +154,13 @@ public long currentTime() {
     LOG.info(baos.toString());
     assertTrue(output.indexOf("Deleted 1 backups") >= 0);
   }
+
+  static class TwoDaysBackEnvironmentEdge extends BaseEnvironmentEdge {
+    // time - 2 days
+    @Override
+    public long currentTime() {
+      return System.currentTimeMillis() - 2 * 24 * 3600 * 1000 ;
+    }
+  }
+
 }

hbase-common/src/main/java/org/apache/hadoop/hbase/util/BaseEnvironmentEdge.java

@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.util;
+
+import java.lang.reflect.Constructor;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.hadoop.hbase.util.clock.BoundedIncrementYieldAdvancingClock;
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+ * Base implementation of an environment edge.
+ */
+@InterfaceAudience.Private
+public abstract class BaseEnvironmentEdge implements EnvironmentEdge {
+
+  // TODO: Hardcoded for now. Should this be configurable? BoundedIncrementYieldAdvancingClock
+  // is the best option, as determined by microbenchmarks. See HBASE-25913.
+  private static final Constructor<?> CLOCK_IMPL_CONSTRUCTOR;
+  static {
+    try {
+      CLOCK_IMPL_CONSTRUCTOR =
+        BoundedIncrementYieldAdvancingClock.class.getConstructor(HashedBytes.class);
+    } catch (Exception e) {
+      // If there is a problem this exception will bubble up and ultimately cause a JVM abort,
+      // which is what we want, because it means a developer failed to update here after
+      // changing how clocks are constructed.
+      throw new RuntimeException(e);
+    }
+  }
+
+  /**
+   * A clock instance representing the system time directly, so we introduce no overheads for the
+   * vast majority of users of EnvironmentEdgeManager.currentTime.
+   */
+  private static final Clock SYSTEM_CLOCK = new Clock() {
+    final HashedBytes NAME = new HashedBytes(Bytes.toBytes("DEFAULT"));
+    @Override
+    public HashedBytes getName() {
+      return NAME;
+    }
+    @Override
+    public long currentTime() {
+      return System.currentTimeMillis();
+    }
+    @Override
+    public long currentTimeAdvancing() {
+      throw new UnsupportedOperationException(
+        "Default clock does not implement currentTimeAdvancing()");
+    }
+    @Override
+    public void get() {
+      throw new UnsupportedOperationException("Default clock does not implement get()");
+    }
+    @Override
+    public boolean remove() {
+      throw new UnsupportedOperationException("Default clock does not implement remove()");
+    }
+  };
+
+  @Override
+  public long currentTime() {
+    return SYSTEM_CLOCK.currentTime();
+  }
+
+  private static final ConcurrentHashMap<HashedBytes, Clock> clockMap = new ConcurrentHashMap<>();
+
+  @Override
+  public Clock getClock(final HashedBytes name) {
+    Clock clock = clockMap.computeIfAbsent(name, k -> {
+      try {
+        return (Clock)CLOCK_IMPL_CONSTRUCTOR.newInstance(k);
+      } catch (Exception e) {
+        throw new RuntimeException(e);
+      }
+    });
+    clock.get(); // increment reference count
+    return clock;
+  }
+
+  @Override
+  public boolean removeClock(Clock clock) {
+    if (clock.remove()) { // only remove when refcount drops to zero
+      if (clockMap.remove(clock.getName()) != null) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+}

hbase-common/src/main/java/org/apache/hadoop/hbase/util/DefaultEnvironmentEdge.java

@@ -20,19 +20,9 @@
 
 import org.apache.yetus.audience.InterfaceAudience;
 
-/**
- * Default implementation of an environment edge.
- */
 @InterfaceAudience.Private
-public class DefaultEnvironmentEdge implements EnvironmentEdge {
-  /**
-   * {@inheritDoc}
-   * <p>
-   * This implementation returns {@link System#currentTimeMillis()}
-   * </p>
-   */
-  @Override
-  public long currentTime() {
-    return System.currentTimeMillis();
-  }
+public class DefaultEnvironmentEdge extends BaseEnvironmentEdge {
+
+  // No overrides here
+
 }

hbase-common/src/main/java/org/apache/hadoop/hbase/util/EnvironmentEdge.java

@@ -28,10 +28,69 @@
  */
 @InterfaceAudience.Private
 public interface EnvironmentEdge {
+
   /**
-   * Returns the currentTime.
+   * Returns the current time using the default clock.
+   * <p>
+   * This is almost always what you want, unless managing timekeeping with a named clock.
    *
-   * @return Current time.
+   * @return The current time.
    */
   long currentTime();
+
+  /**
+   * Get the clock associated with the given identifier.
+   * @param name clock identifier
+   * @return the clock instance for the given identifier
+   */
+  Clock getClock(HashedBytes name);
+
+  /**
+   * Release the reference to and possible remove this clock.
+   * @param clock the clock
+   * @return true if the clock was removed, false if it did not exist
+   */
+  boolean removeClock(Clock clock);
+
+  /**
+   * Abstraction for an environment's time source.
+   */
+  public interface Clock {
+
+    /**
+     * Returns the clock's identifier.
+     */
+    HashedBytes getName();
+
+    /**
+     * Returns the current time using a named clock.
+     * @return The current time, according to the given named clock.
+     */
+    long currentTime();
+
+    /**
+     * Returns the current time using a named clock. Ensure the clock advanced by
+     * at least one tick before returning.
+     * <p>
+     * This method may block the current thread's execution or cause it to yield.
+     * @return The current time, according to the given named clock.
+     * @throws InterruptedException if interrupted while waiting for the clock to advance
+     */
+    default long currentTimeAdvancing() throws InterruptedException {
+      throw new UnsupportedOperationException("BaseClock does not implement currentTimeAdvancing");
+    }
+
+    /**
+     * Called to increment the reference count of the clock.
+     */
+    void get();
+
+    /**
+     * Called when the clock is removed.
+     * @return true if the reference count is zero, false otherwise.
+     */
+    boolean remove();
+
+  }
+
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/util/HashedBytes.java → hbase-common/src/main/java/org/apache/hadoop/hbase/util/HashedBytes.java

@@ -36,7 +36,12 @@ public class HashedBytes {
 
   public HashedBytes(byte[] bytes) {
     this.bytes = bytes;
-    hashCode = Bytes.hashCode(bytes);
+    hashCode = Bytes.hashCode(this.bytes);
+  }
+
+  public HashedBytes(byte[]... bytes) {
+    this.bytes = Bytes.add(bytes);
+    hashCode = Bytes.hashCode(this.bytes);
   }
 
   public byte[] getBytes() {
@@ -50,10 +55,12 @@ public int hashCode() {
 
   @Override
   public boolean equals(Object obj) {
-    if (this == obj)
+    if (this == obj) {
       return true;
-    if (obj == null || getClass() != obj.getClass())
+    }
+    if (obj == null || getClass() != obj.getClass()) {
       return false;
+    }
     HashedBytes other = (HashedBytes) obj;
     return (hashCode == other.hashCode) && Arrays.equals(bytes, other.bytes);
   }

hbase-common/src/main/java/org/apache/hadoop/hbase/util/IncrementingEnvironmentEdge.java

@@ -24,7 +24,41 @@
  * Uses an incrementing algorithm instead of the default.
  */
 @InterfaceAudience.Private
-public class IncrementingEnvironmentEdge implements EnvironmentEdge {
+public class IncrementingEnvironmentEdge extends BaseEnvironmentEdge {
+
+  class ManualIncrementingClock implements EnvironmentEdge.Clock {
+
+    private HashedBytes name;
+
+    public ManualIncrementingClock(HashedBytes name) {
+      this.name = name;
+    }
+
+    @Override
+    public HashedBytes getName() {
+      return name;
+    }
+
+    @Override
+    public long currentTime() {
+      return System.currentTimeMillis() + timeIncrement;
+    }
+
+    @Override
+    public long currentTimeAdvancing() {
+      return System.currentTimeMillis() + timeIncrement;
+    }
+
+    @Override
+    public void get() {
+    }
+
+    @Override
+    public boolean remove() {
+      return true;
+    }
+
+  }
 
   private long timeIncrement;
 
@@ -62,4 +96,15 @@ public synchronized long incrementTime(long amount) {
     timeIncrement += amount;
     return timeIncrement;
   }
+
+  @Override
+  public Clock getClock(HashedBytes name) {
+    return new ManualIncrementingClock(name);
+  }
+
+  @Override
+  public boolean removeClock(Clock clock) {
+    return true;
+  }
+
 }