[
https://issues.apache.org/jira/browse/HBASE-30327?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18107299#comment-18107299
]
mazhengxuan commented on HBASE-30327:
-------------------------------------
Thanks, this sounds good to me. Since you are preparing the PR, I will wait for
it and help review the concurrency and test coverage. In particular, I think we
should verify the MemStore exception path and the latch accounting when an
inline seek fails after some handlers have already been submitted.
> 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: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)
