[
https://issues.apache.org/jira/browse/HBASE-30327?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18106345#comment-18106345
]
mazhengxuan commented on HBASE-30327:
-------------------------------------
Thanks for putting this together. The general direction makes sense to me. The
parallel seek executor has an unbounded queue, so under enough concurrent scans
we can end up with many queued seek tasks while the scanner threads wait for
their own tasks to finish. Falling back to the calling thread when the pool is
busy seems like a reasonable way to add some backpressure.
One thing we should clarify is whether the goal is to reduce queue buildup or
to guarantee that no task is queued when the pool is saturated. Checking
queue.isEmpty() and getActiveCount() can help with the former, but it cannot
provide a strict guarantee because the values are only snapshots and several
StoreScanners may race with each other. I think best-effort behavior is
probably good enough for the first version.
There are a couple of details we need to be careful about in the implementation.
Capacity should only be counted for StoreFileScanners, since MemStore scanners
are executed inline. Also, once we have submitted any handlers, an IOException
from a later inline seek must not cause us to return immediately. We should
wait for the already submitted handlers before propagating the error; otherwise
the caller may close scanners that are still being used by worker threads.
For the same reason, I would keep the inline path as a direct scanner.seek()
call instead of running ParallelSeekHandler on the calling thread. The handler
has special handling for thread-local scan metrics which assumes it is running
on a worker thread.
I suggest starting with a focused test using a small parallel seek pool. We can
occupy all workers, run a seek over several StoreFileScanners, and verify that
adaptive mode falls back to the calling thread without filling the executor
queue. The same test area can cover partial capacity, mixed StoreFile/MemStore
scanners, and failures from both inline and submitted seeks.
For the first patch, I would keep the change local to StoreScanner and reuse
the existing ParallelSeekHandler and CountDownLatch. I would not change the
generic ExecutorService yet. A separate, disabled-by-default configuration also
seems useful initially, since this change trades executor queueing for more
HFile I/O on scanner/RPC threads. It gives us an easy way to compare the two
behaviors under load.
It would be good to hear other opinions on two points: whether best-effort
capacity checking is sufficient, and whether the adaptive behavior should have
its own configuration or eventually replace the existing parallel seek behavior.
> Adaptive Parallel Seek: Fallback to sequential when thread pool is saturated
> ----------------------------------------------------------------------------
>
> Key: HBASE-30327
> URL: https://issues.apache.org/jira/browse/HBASE-30327
> Project: HBase
> Issue Type: Improvement
> Components: regionserver, Scanners
> Reporter: Ichsan Said
> Priority: Major
> Labels: Scanner, performance, scan
>
> h3. Problem
> When parallel seek is enabled
> ({{hbase.storescanner.parallel.seek.enable=true}}), all StoreFileScanner seek
> operations are submitted to the {{RS_PARALLEL_SEEK}} thread pool. The pool
> uses an unbounded {{LinkedBlockingQueue}}, so submissions never block.
> However, when the pool is saturated under high concurrency:
> 1. All tasks queue up in the unbounded queue
> 2. The calling thread blocks on {{CountDownLatch.await()}} until all tasks
> complete
> 3. This can lead to increased latency and thread starvation
> h3. Proposed Solution
> Introduce an *Adaptive Parallel Seek* strategy that gracefully handles thread
> pool saturation:
> 1. *Check capacity before submission*: Use a conservative approach - only
> report capacity when the task queue is empty AND active threads < pool size
> 2. *Sequential fallback*: When capacity is 0, seek the scanner synchronously
> on the calling thread instead of queuing
> 3. *Opportunistic parallelization*: After each sequential seek, re-check
> capacity. If slots become available, submit remaining scanners (up to
> available capacity) for parallel execution
> 4. *Truly concurrent execution*: Sequential seeks on the calling thread
> overlap with submitted parallel tasks - no intermediate blocking between
> batches
> 5. *Single await at the end*: All parallel tasks share one
> {{CountDownLatch}}; the calling thread only blocks once after all submissions
> and sequential seeks are done
> h3. Algorithm Flow
> {code}
> // Pre-count StoreFileScanners to size the shared latch
> int parallelCount = count of StoreFileScanner instances in scanners
> CountDownLatch latch = new CountDownLatch(parallelCount)
> index = 0
> while (index < scannerCount):
> capacity = getAvailableCapacity()
> if capacity == 0:
> // Pool saturated → seek inline on calling thread
> if scanner instanceof StoreFileScanner:
> scanner.seek(key)
> latch.countDown() // treated as done inline
> else:
> scanner.seek(key) // memstore, no latch
> index++
> else:
> // Pool has slots → submit batch, DO NOT await, continue immediately
> batchEnd = min(index + capacity, scannerCount)
> for i in [index, batchEnd):
> if StoreFileScanner: executor.submit(handler with shared latch)
> else: scanner.seek(key) // memstore inline
> index = batchEnd
> // ← no await here, calling thread continues immediately
> // Single await for all parallel tasks
> latch.await()
> // Check all handlers for errors
> {code}
> h3. Timeline Illustration
> *Scenario*: 8 StoreFileScanners, pool size = 3, pool initially saturated
> (active=3)
> *Before - current parallelSeek (pool saturated):*
> {code}
> Time
> ──────────────────────────────────────────────────────────────────────────────►
> Calling │ submit P1..P8 (all queued) │ AWAIT │
> done
> Thread └─────────────────────────── ┴─────────────────────────────────────┘
> ▲
> blocks until all 8 complete
> Workers │ [P1 ]│ [P4 ]│ [P7 ]│
> │ [P2 ]│ [P5 ]│ [P8 ]│
> │ [P3 ]│ [P6 ]│
> {code}
> *After - adaptive parallelSeek (pool saturated):*
> {code}
> Time
> ──────────────────────────────────────────────────────────────────────────────►
> Calling │ seq │ seq │ submit │ seq │ submit │ seq │ submit │ │
> done
> Thread │ S1 │ S2 │ P3,P4 │ S5 │ P6,P7 │ S8 │ │ AWAIT │
> └──────┴──────┴────────┴──────┴────────┴──────┴────────┴────────┘
> ▲
> wait once at end
> Workers │ │ [P3 ]│ │ [P6 ]│
> │ │ [P4 ]│ │ [P7 ]│
> {code}
> *Key insight*: When pool is saturated, adaptive seek falls back to sequential
> (S1, S2)
> instead of queuing all tasks. As slots free up, remaining scanners are
> submitted
> opportunistically (P3,P4 then P6,P7). Only one {{CountDownLatch.await()}} at
> the end.
> h3. Comparison: Current vs Adaptive
> || Aspect || Current parallelSeek || Adaptive parallelSeek ||
> | Pool saturated | All tasks queued, block until all done | Sequential
> fallback, no queue buildup |
> | Mid-loop blocking | N/A | None - calling thread continues immediately after
> submit |
> | Parallel + sequential overlap | No | Yes - truly concurrent |
> | Latency under load | Spikes due to queue wait | Predictable, graceful
> degradation |
> | Pool available | All parallel | All parallel (same behavior) |
> | CountDownLatch.await() calls | Once | Once (shared latch across all
> batches) |
> h3. Configuration
> New configuration property:
> {code:java}
> hbase.storescanner.adaptive.parallel.seek.enable=false (default)
> {code}
> Configuration interaction:
> || parallel.seek.enable || adaptive.parallel.seek.enable || Behavior ||
> | false | * | Sequential seek only |
> | true | false | Existing parallel seek (current behavior) |
> | true | true | Adaptive parallel seek (new) |
> h3. Implementation Approach
> Modify {{StoreScanner.seekScanners()}} to dispatch to new
> {{adaptiveParallelSeek()}} method when both configs are enabled. The new
> method:
> - Pre-counts {{StoreFileScanner}} instances to size a single shared
> {{CountDownLatch}}
> - Uses single loop through scanners (same pattern as existing
> {{parallelSeek}})
> - Checks {{instanceof StoreFileScanner}} inline (same as {{parallelSeek}})
> - Calls {{latch.countDown()}} inline for sequentially-seeked
> {{StoreFileScanner}} instances
> - Uses {{executor.getExecutorThreadPool(ExecutorType.RS_PARALLEL_SEEK)}} only
> for capacity checking
> - Uses {{executor.submit(handler)}} for task submission (same as
> {{parallelSeek}})
> - Single {{latch.await()}} after the loop completes
> h3. Race Condition Analysis
> The capacity check is inherently racy (TOCTOU), but acceptable:
> - *Over-estimation*: Tasks get queued - handled by unbounded queue, no
> correctness issue
> - *Under-estimation*: Sequential seek when parallel possible - slower but
> correct
> - *Multiple scanners racing*: Conservative queue-empty check mitigates
> runaway queue buildup
> h3. Benefits
> - Reduces latency under high concurrency (no queue blocking)
> - Parallel and sequential seeks overlap - calling thread never idles between
> batches
> - Single {{CountDownLatch.await()}} instead of one per batch
> - Graceful degradation when pool is saturated
> - Backward compatible (disabled by default)
> - Minimal code change (reuses existing {{ParallelSeekHandler}} infrastructure)
> h3. Known Limitations / Future Work
> - No metrics for adaptive behavior monitoring (can be added in follow-up)
> - Capacity check is best-effort estimate due to {{getActiveCount()}}
> approximation
> - Two configuration properties required (maintaining backward compatibility)
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
