[
https://issues.apache.org/jira/browse/HBASE-30327?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18107680#comment-18107680
]
mazhengxuan commented on HBASE-30327:
-------------------------------------
Good idea. A separate follow-up JIRA makes sense.
For the initial metrics, I suggest:
- Number of seeks submitted to the parallel pool
- Number of seeks executed inline
- Time spent on inline seeks
- Queued and running tasks in the RS_PARALLEL_SEEK executor
The fallback ratio can be derived from the first two counters. I would leave
the batch-size histogram out for now—it may help with future tuning, but is not
necessary to evaluate the initial behavior.
These metrics should be aggregated at the RegionServer level and updated in
batches to keep the overhead low.
> 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, pull-request-available, 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:java}
> // 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:java}
> 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:java}
> 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|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)
