Re: [PR] HBASE-30327 Implement adaptive parallel seek in StoreScanner [hbase]

["Sigma-Ma (via GitHub)" <gi...@apache.org>]

Sigma-Ma commented on code in PR #8566:
URL: https://github.com/apache/hbase/pull/8566#discussion_r3964152675


##########
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java:
##########
@@ -1257,18 +1263,155 @@ private void parallelSeek(final List<? extends KeyValueScanner> scanners, final
         latch.countDown();
       }
     }
+    InterruptedIOException interruptedException = null;
+    while(true) {
+      try {
+        latch.await();
+        break;
+      } catch (InterruptedException ie) {
+        interruptedException = (InterruptedIOException) new InterruptedIOException().initCause(ie);
+      }
+    }
+    if (interruptedException != null) {
+      throw interruptedException;
+    }
+    for (ParallelSeekHandler handler : handlers) {
+      if (handler.getErr() != null) {
+        throw new IOException(handler.getErr());
+      }
+    }
+  }
 
-    try {
-      latch.await();
-    } catch (InterruptedException ie) {
-      throw (InterruptedIOException) new InterruptedIOException().initCause(ie);
+  /**
+   * Returns the number of threads available for immediate execution in the parallel seek thread
+   * pool. Uses a conservative approach: only reports capacity when the task queue is empty AND
+   * active threads < pool size.
+   * @return number of threads available for immediate execution, or 0 if saturated
+   */
+  private int getAvailableParallelSeekCapacity() {
+    ThreadPoolExecutor pool = executor.getExecutorThreadPool(ExecutorType.RS_PARALLEL_SEEK);
+    if (!pool.getQueue().isEmpty()) {
+      return 0; // Conservative: any queued work means saturated
     }
+    return Math.max(0, pool.getCorePoolSize() - pool.getActiveCount());
+  }
+
+  /**
+   * Seeks scanners using an adaptive strategy that switches between parallel and sequential
+   * execution based on thread pool availability.
+   * <p>
+   * When the parallel seek thread pool is saturated, falls back to sequential seeking. After each
+   * sequential seek, re-checks capacity and opportunistically submits remaining scanners for
+   * parallel execution when slots become available.
+   * <p>
+   * If an IOException occurs during an inline seek, we must wait for any already-submitted handlers
+   * to complete before propagating the error. This prevents the caller from closing scanners that
+   * are still being used by worker threads.
+   *
+   * @param scanners list of KeyValueScanners to seek
+   * @param kv       the key to seek to
+   * @throws IOException if any seek operation fails
+   */
+  private void adaptiveParallelSeek(final List<? extends KeyValueScanner> scanners,
+    final ExtendedCell kv) throws IOException {
+    if (scanners.isEmpty()) return;
 
+    int scannerCount = scanners.size();
+    // Pre-count StoreFileScanners to size the latch correctly
+    int storeFileScannerCount = 0;
+    for (KeyValueScanner scanner : scanners) {
+      if (scanner instanceof StoreFileScanner) {
+        storeFileScannerCount++;
+      }
+    }
+    CountDownLatch latch = new CountDownLatch(storeFileScannerCount);
+    List<ParallelSeekHandler> handlers = new ArrayList<>(storeFileScannerCount);
+    int index = 0;
+    IOException inlineSeekError = null;
+
+    while (index < scannerCount) {
+      int capacity = getAvailableParallelSeekCapacity();
+
+      if (capacity == 0) {
+        // Sequential fallback: process one scanner on calling thread
+        KeyValueScanner scanner = scanners.get(index);
+        try {
+          scanner.seek(kv);
+        } catch (IOException e) {
+          // Must wait for already-submitted handlers before propagating error
+          inlineSeekError = e;

Review Comment:
   Could we advance index before breaking here? This StoreFileScanner has already decremented the latch, but the cleanup loop starts from the same index and decrements it again. If an earlier submitted handler is still running, the latch may reach zero early and the caller can return while that handler is still using its scanner.
   
   A test with a submitted blocking handler followed by an inline StoreFileScanner failure would cover this path.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hbase.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org