ichsansaid commented on code in PR #8566:
URL: https://github.com/apache/hbase/pull/8566#discussion_r3872290326
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java:
##########
@@ -1271,6 +1277,135 @@ private void parallelSeek(final List<? extends
KeyValueScanner> scanners, final
}
}
+ /**
+ * 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;
+ if (scanner instanceof StoreFileScanner) {
+ latch.countDown();
+ }
+ index++;
+ break;
+ }
+ if (scanner instanceof StoreFileScanner) {
+ latch.countDown();
+ }
+ index++;
+ } else {
+ // Opportunistic parallel: submit batch up to available capacity
+ int batchEnd = Math.min(index + capacity, scannerCount);
+ for (int i = index; i < batchEnd; i++) {
+ KeyValueScanner scanner = scanners.get(i);
+ if (scanner instanceof StoreFileScanner) {
+ ParallelSeekHandler seekHandler =
+ new ParallelSeekHandler(scanner, kv, this.readPt, latch);
+ executor.submit(seekHandler);
+ handlers.add(seekHandler);
+ } else {
+ try {
+ scanner.seek(kv);
+ } catch (IOException e) {
+ // Must wait for already-submitted handlers before propagating
error
+ inlineSeekError = e;
+ // Count down latch for remaining StoreFileScanners in this
batch that won't be
+ // processed
+ for (int j = i + 1; j < batchEnd; j++) {
+ if (scanners.get(j) instanceof StoreFileScanner) {
+ latch.countDown();
+ }
+ }
+ index = batchEnd;
+ break;
+ }
+ }
+ }
+ if (inlineSeekError != null) {
+ break;
+ }
+ index = batchEnd;
+ }
+ }
+
+ // Count down latch for any remaining unprocessed StoreFileScanners
+ for (int i = index; i < scannerCount; i++) {
+ if (scanners.get(i) instanceof StoreFileScanner) {
+ latch.countDown();
+ }
+ }
+
+ try {
+ latch.await();
+ } catch (InterruptedException ie) {
+ throw (InterruptedIOException) new
InterruptedIOException().initCause(ie);
+ }
Review Comment:
Good point. The fix is to loop on latch.await() and catch
InterruptedException without throwing immediately. This works because
CountDownLatch.await() clears the interrupt flag when it throws
InterruptedException ([Java
docs](https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/util/concurrent/CountDownLatch.html)),
so the next call will block normally until all submitted handlers finish.
Once the latch reaches zero, we restore the interrupt status via
Thread.currentThread().interrupt() and then throw InterruptedIOException. This
ensures no scanner is closed while a worker thread is still using it.
One thing I noticed:
The existing parallelSeek has the same behavior, it throws
InterruptedIOException immediately on interruption without waiting for
submitted handlers to finish. Should we fix that in this PR as well, or would
you prefer to track it as a separate JIRA to keep this patch focused?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
