[
https://issues.apache.org/jira/browse/HBASE-30327?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ichsan Said updated HBASE-30327:
--------------------------------
Description:
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)
was:
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. *No blocking on full queue*: Avoids latency spikes from queue buildup
h3. Algorithm Flow
{code}
while (index < scannerCount):
capacity = getAvailableCapacity()
if capacity == 0:
// Pool saturated → sequential fallback
scanner[index].seek(key)
index++
else:
// Pool has slots → parallel batch
batchSize = min(capacity, remaining scanners)
for each scanner in batch:
if StoreFileScanner: submit to pool
else: seek inline (memstore)
latch.await()
index += batchSize
check handlers for errors
{code}
h3. Timeline Illustration
*Scenario*: 8 StoreFileScanners, pool size = 3, pool initially busy (0 capacity)
{code}
Time Pool State Action Result
──── ────────────── ───────────────────────── ──────────────────────
T0 active=3, queue=[] capacity=0 Sequential: seek
scanner[0]
T1 active=3, queue=[] capacity=0 Sequential: seek
scanner[1]
T2 active=2, queue=[] capacity=1 (slot freed!) Parallel: submit
scanner[2], await
T3 active=3, queue=[] capacity=0 Sequential: seek
scanner[3]
T4 active=1, queue=[] capacity=2 (2 slots freed!) Parallel: submit
scanner[4,5], await
T5 active=3, queue=[] capacity=0 Sequential: seek
scanner[6]
T6 active=2, queue=[] capacity=1 Parallel: submit
scanner[7], await
T7 done check errors Return
{code}
*Key insight*: Instead of submitting all 8 scanners and blocking on a long
queue, we adaptively mix sequential and parallel based on real-time pool
availability.
h3. Comparison: Current vs Adaptive
|| Aspect || Current parallelSeek || Adaptive parallelSeek ||
| Pool saturated | All 8 tasks queued, block until all done | Sequential
fallback, no queue buildup |
| Latency under load | Spikes due to queue wait | Predictable, graceful
degradation |
| Pool available | All parallel | All parallel (same behavior) |
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:
- Uses single loop through scanners (same pattern as existing {{parallelSeek}})
- Checks {{instanceof StoreFileScanner}} inline (same as {{parallelSeek}})
- Uses {{executor.getExecutorThreadPool(ExecutorType.RS_PARALLEL_SEEK)}} only
for capacity checking
- Uses {{executor.submit(handler)}} for task submission (same as
{{parallelSeek}})
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)
- 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)
> 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)
