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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
