ichsansaid commented on code in PR #8566:
URL: https://github.com/apache/hbase/pull/8566#discussion_r3871648316
##########
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);
Review Comment:
You're right. The current batchEnd = index + capacity counts all scanners
including MemStore ones that run inline, so we can end up submitting fewer
StoreFile tasks than the available capacity allows.
Ill do a changes for this
--
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]
