HBASE-17090 Procedure v2 - fast wake if nothing else is running (Matteo Bertozzi)

hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/ProcedureExecutor.java

@@ -1536,7 +1536,7 @@ public class ProcedureExecutor<TEnvironment> {
         final Procedure procedure = scheduler.poll(keepAliveTime, TimeUnit.MILLISECONDS);
         if (procedure == null) continue;
 
-        activeExecutorCount.incrementAndGet();
+        store.setRunningProcedureCount(activeExecutorCount.incrementAndGet());
         executionStartTime.set(EnvironmentEdgeManager.currentTime());
         try {
           if (isTraceEnabled) {
@@ -1544,7 +1544,7 @@ public class ProcedureExecutor<TEnvironment> {
           }
           executeProcedure(procedure);
         } finally {
-          activeExecutorCount.decrementAndGet();
+          store.setRunningProcedureCount(activeExecutorCount.decrementAndGet());
           lastUpdate = EnvironmentEdgeManager.currentTime();
           executionStartTime.set(Long.MAX_VALUE);
         }

hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/NoopProcedureStore.java

@@ -51,6 +51,11 @@ public class NoopProcedureStore extends ProcedureStoreBase {
     return numThreads;
   }
 
+  @Override
+  public void setRunningProcedureCount(final int count) {
+    // no-op
+  }
+
   @Override
   public void load(final ProcedureLoader loader) throws IOException {
     loader.setMaxProcId(0);

hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/ProcedureStore.java

@@ -150,6 +150,12 @@ public interface ProcedureStore {
    */
   int getNumThreads();
 
+  /**
+   * Set the number of procedure running.
+   * This can be used, for example, by the store to know how long to wait before a sync.
+   */
+  void setRunningProcedureCount(int count);
+
   /**
    * Acquire the lease for the procedure store.
    */

hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/wal/WALProcedureStore.java

@@ -136,7 +136,9 @@ public class WALProcedureStore extends ProcedureStoreBase {
   private LinkedTransferQueue<ByteSlot> slotsCache = null;
   private Set<ProcedureWALFile> corruptedLogs = null;
   private FSDataOutputStream stream = null;
+  private int runningProcCount = 1;
   private long flushLogId = 0;
+  private int syncMaxSlot = 1;
   private int slotIndex = 0;
   private Thread syncThread;
   private ByteSlot[] slots;
@@ -198,6 +200,8 @@ public class WALProcedureStore extends ProcedureStoreBase {
 
     // Init buffer slots
     loading.set(true);
+    runningProcCount = numSlots;
+    syncMaxSlot = numSlots;
     slots = new ByteSlot[numSlots];
     slotsCache = new LinkedTransferQueue();
     while (slotsCache.size() < numSlots) {
@@ -288,6 +292,12 @@ public class WALProcedureStore extends ProcedureStoreBase {
     return slots == null ? 0 : slots.length;
   }
 
+  @Override
+  public void setRunningProcedureCount(final int count) {
+    LOG.debug("set running procedure count=" + count + " slots=" + slots.length);
+    this.runningProcCount = count > 0 ? Math.min(count, slots.length) : slots.length;
+  }
+
   public ProcedureStoreTracker getStoreTracker() {
     return storeTracker;
   }
@@ -623,7 +633,7 @@ public class WALProcedureStore extends ProcedureStoreBase {
           throw new RuntimeException("sync aborted", syncException.get());
         } else if (inSync.get()) {
           syncCond.await();
-        } else if (slotIndex == slots.length) {
+        } else if (slotIndex >= syncMaxSlot) {
           slotCond.signal();
           syncCond.await();
         } else {
@@ -642,7 +652,7 @@ public class WALProcedureStore extends ProcedureStoreBase {
       }
 
       // Notify that the slots are full
-      if (slotIndex == slots.length) {
+      if (slotIndex == syncMaxSlot) {
         waitCond.signal();
         slotCond.signal();
       }
@@ -725,8 +735,10 @@ public class WALProcedureStore extends ProcedureStoreBase {
             }
           }
           // Wait SYNC_WAIT_MSEC or the signal of "slots full" before flushing
+          syncMaxSlot = runningProcCount;
+          assert syncMaxSlot > 0 : "unexpected syncMaxSlot=" + syncMaxSlot;
           final long syncWaitSt = System.currentTimeMillis();
-          if (slotIndex != slots.length) {
+          if (slotIndex != syncMaxSlot) {
             slotCond.await(syncWaitMsec, TimeUnit.MILLISECONDS);
           }
 
@@ -734,7 +746,7 @@ public class WALProcedureStore extends ProcedureStoreBase {
           final long syncWaitMs = currentTs - syncWaitSt;
           final float rollSec = getMillisFromLastRoll() / 1000.0f;
           final float syncedPerSec = totalSyncedToStore / rollSec;
-          if (LOG.isTraceEnabled() && (syncWaitMs > 10 || slotIndex < slots.length)) {
+          if (LOG.isTraceEnabled() && (syncWaitMs > 10 || slotIndex < syncMaxSlot)) {
             LOG.trace(String.format("Sync wait %s, slotIndex=%s , totalSynced=%s (%s/sec)",
                       StringUtils.humanTimeDiff(syncWaitMs), slotIndex,
                       StringUtils.humanSize(totalSyncedToStore),
@@ -813,27 +825,31 @@ public class WALProcedureStore extends ProcedureStoreBase {
     return totalSynced;
   }
 
-  protected long syncSlots(FSDataOutputStream stream, ByteSlot[] slots, int offset, int count)
+  protected long syncSlots(final FSDataOutputStream stream, final ByteSlot[] slots,
-      throws IOException {
+      final int offset, final int count) throws IOException {
     long totalSynced = 0;
     for (int i = 0; i < count; ++i) {
-      ByteSlot data = slots[offset + i];
+      final ByteSlot data = slots[offset + i];
       data.writeTo(stream);
       totalSynced += data.size();
     }
 
+    syncStream(stream);
+    sendPostSyncSignal();
+
+    if (LOG.isTraceEnabled()) {
+      LOG.trace("Sync slots=" + count + '/' + syncMaxSlot +
+                ", flushed=" + StringUtils.humanSize(totalSynced));
+    }
+    return totalSynced;
+  }
+
+  protected void syncStream(final FSDataOutputStream stream) throws IOException {
     if (useHsync) {
       stream.hsync();
     } else {
       stream.hflush();
     }
-    sendPostSyncSignal();
-
-    if (LOG.isTraceEnabled()) {
-      LOG.trace("Sync slots=" + count + '/' + slots.length +
-                ", flushed=" + StringUtils.humanSize(totalSynced));
-    }
-    return totalSynced;
   }
 
   private boolean rollWriterWithRetries() {

hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/store/wal/ProcedureWALPerformanceEvaluation.java

@@ -149,7 +149,7 @@ public class ProcedureWALPerformanceEvaluation extends AbstractHBaseTool {
       // Start worker threads.
       long start = System.currentTimeMillis();
       for (int i = 0; i < numThreads; i++) {
-        futures[i] = executor.submit(this.new Worker(start));
+        futures[i] = executor.submit(new Worker(start));
       }
       boolean failure = false;
       try {
@@ -197,8 +197,8 @@ public class ProcedureWALPerformanceEvaluation extends AbstractHBaseTool {
    * If procedure store fails to roll log file (throws IOException), all threads quit, and at
    * least one returns value of {@link AbstractHBaseTool#EXIT_FAILURE}.
    */
-  class Worker implements Callable<Integer> {
+  private final class Worker implements Callable<Integer> {
-    final long start;
+    private final long start;
 
     public Worker(long start) {
       this.start = start;
@@ -243,7 +243,7 @@ public class ProcedureWALPerformanceEvaluation extends AbstractHBaseTool {
     }
   }
 
-  public class NoSyncWalProcedureStore extends WALProcedureStore {
+  private class NoSyncWalProcedureStore extends WALProcedureStore {
     public NoSyncWalProcedureStore(final Configuration conf, final FileSystem fs,
         final Path logDir) {
       super(conf, fs, logDir, new WALProcedureStore.LeaseRecovery() {
@@ -255,13 +255,8 @@ public class ProcedureWALPerformanceEvaluation extends AbstractHBaseTool {
     }
 
     @Override
-    protected long syncSlots(FSDataOutputStream stream, ByteSlot[] slots, int offset, int count)
+    protected void syncStream(FSDataOutputStream stream) {
-        throws IOException {
+      // no-op
-      long totalSynced = 0;
-      for (int i = 0; i < count; ++i) {
-        totalSynced += slots[offset + i].size();
-      }
-      return totalSynced;
     }
   }