HBASE-28292 Make Delay prefetch property to be dynamically configured

Author: kabhishek4

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileReaderImpl.java

@@ -1659,6 +1659,11 @@ public boolean prefetchComplete() {
     return PrefetchExecutor.isCompleted(path);
   }
 
+  /**
+   * Returns true if block prefetching was started after waiting for specified delay, false
+   * otherwise
+   */
+  @Override
   public boolean prefetchStarted() {
     return PrefetchExecutor.isPrefetchStarted();
   }

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/PrefetchExecutor.java

@@ -88,9 +88,6 @@ public Thread newThread(Runnable r) {
       + Path.SEPARATOR_CHAR + ")|(" + Path.SEPARATOR_CHAR
       + HConstants.HREGION_COMPACTIONDIR_NAME.replace(".", "\\.") + Path.SEPARATOR_CHAR + ")");
 
-  // For tests. Contains computed prefetch delay
-  private static long computedPrefetchDelay;
-
   public static void request(Path path, Runnable runnable) {
     if (!prefetchPathExclude.matcher(path.toString()).find()) {
       long delay;
@@ -102,7 +99,6 @@ public static void request(Path path, Runnable runnable) {
         delay = 0;
       }
       try {
-        computedPrefetchDelay = delay;
         LOG.debug("Prefetch requested for {}, delay={} ms", path, delay);
         final Runnable tracedRunnable =
           TraceUtil.tracedRunnable(runnable, "PrefetchExecutor.request");
@@ -165,13 +161,6 @@ static ScheduledExecutorService getExecutorPool() {
     return prefetchExecutorPool;
   }
 
-  /* Visible for testing only */
-  @RestrictedApi(explanation = "Should only be called in tests. This is a non thread safe "
-    + "variable that would not yield accurate values when multiple readers created and "
-    + "calling PrefetchExecutor.request concurrently", link = "",
-    allowedOnPath = ".*/src/test/.*")
-  public static long getComputedPrefetchDelay() {return computedPrefetchDelay;}
-
   @RestrictedApi(explanation = "Should only be called in tests", link = "",
     allowedOnPath = ".*/src/test/.*")
   static Map<Path, Future<?>> getPrefetchFutures() {
@@ -208,9 +197,10 @@ public static int getPrefetchDelay() {
   public static void loadConfiguration(Configuration conf) {
     prefetchDelayMillis = conf.getInt(PREFETCH_DELAY, 1000);
     prefetchFutures.forEach((k, v) -> {
-      // Do not cancel the task which is about to complete
       ScheduledFuture sf = (ScheduledFuture) prefetchFutures.get(k);
-      if (sf.getDelay(TimeUnit.MILLISECONDS) > prefetchDelayMillis) {
+      if (!(sf.getDelay(TimeUnit.MILLISECONDS) > 0)) {
+        //the thread is still pending delay expiration and has not started to run yet, so can be
+        // re-scheduled at no cost.
         interrupt(k);
         request(k, prefetchRunnable.get(k));
       }

hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestPrefetch.java

@@ -21,7 +21,6 @@
 import static org.apache.hadoop.hbase.client.trace.hamcrest.SpanDataMatchers.hasParentSpanId;
 import static org.apache.hadoop.hbase.io.hfile.CacheConfig.CACHE_DATA_BLOCKS_COMPRESSED_KEY;
 import static org.apache.hadoop.hbase.io.hfile.PrefetchExecutor.PREFETCH_DELAY;
-import static org.apache.hadoop.hbase.io.hfile.PrefetchExecutor.getComputedPrefetchDelay;
 import static org.apache.hadoop.hbase.regionserver.CompactSplit.HBASE_REGION_SERVER_ENABLE_COMPACTION;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.allOf;
@@ -101,7 +100,6 @@ public class TestPrefetch {
   private static final int NUM_KV = 1000;
 
   private long startTime;
-  private long endTime;
   private boolean measureTiming;
 
 
@@ -314,11 +312,14 @@ private void readStoreFile(Path storeFilePath,
         // if prefetchFutures contains entry (which means it's not cancelled or completed)
         // and wait time remaining is below delay expiry watermark, it can be deduced that
         // the prefetch is not started yet.
-        assertFalse("Prefetch Should not start at this point", reader.prefetchStarted());
-        setComputeTiming(false);
+        if (getElapsedTime() >= (conf.getLong(PREFETCH_DELAY, 1000))) {
+          assertTrue("Prefetch should be started at this point", reader.prefetchStarted());
+          setComputeTiming(false);
+        } else {
+          assertFalse("Prefetch Should not start at this point", reader.prefetchStarted());
+        }
       }
     }
-    endTimer();
     long offset = 0;
     while (offset < reader.getTrailer().getLoadOnOpenDataOffset()) {
       HFileBlock block = readFunction.apply(reader, offset);
@@ -374,7 +375,7 @@ public void testOnConfigurationChange() {
   @Test
   public void testPrefetchWithDelay() throws Exception {
     PrefetchExecutorNotifier prefetchExecutorNotifier = new PrefetchExecutorNotifier(conf);
-    conf.setInt(PREFETCH_DELAY, 60000);
+    conf.setInt(PREFETCH_DELAY, 35000);
     prefetchExecutorNotifier.onConfigurationChange(conf);
 
     HFileContext context = new HFileContextBuilder().withCompression(Compression.Algorithm.GZ)
@@ -385,10 +386,6 @@ public void testPrefetchWithDelay() throws Exception {
     setComputeTiming(true);
 
     readStoreFile(storeFile);
-    //We apply delay variation to the provided delay and used the computed delay while scheduling
-    //Using the computed delay for the comparison
-    assertTrue("Total execution time is not more than delay duration",
-      getElapsedTime() > PrefetchExecutor.getComputedPrefetchDelay());
     resetTiming();
 
     conf.setInt(PREFETCH_DELAY, 1000);
@@ -551,7 +548,6 @@ public static KeyValue.Type generateKeyType(Random rand) {
 
   private void resetTiming() {
     startTime = 0;
-    endTime = 0;
     measureTiming = false;
   }
 
@@ -561,13 +557,8 @@ private void startTimer() {
     }
   }
 
-  private void endTimer() {
-    if (startTime > 0)
-    endTime = System.currentTimeMillis();
-  }
-
   private long getElapsedTime() {
-    return endTime - startTime;
+    return System.currentTimeMillis() - startTime;
   }
 
   private void setComputeTiming(boolean status) {