Reduce processing-pool queue-lock contention with a sharded task queue (druid.processing.numThreadPools)
- Dominant language
- Java
- Stars
- 14.1k
- Forks
- 3.8k
- Avg merge
- 2d 58m
- Merged PRs (30d)
- 233
Description
### Description
Add an option to split the historical/peon **processing pool** into N independent
pools ("shards"), each with its own `PriorityBlockingQueue` and its own lock,
instead of the current single queue guarded by one `ReentrantLock`. Each task is
routed to a shard at submit time, so each queue lock sees only ~1/N of the
submit/take traffic and ~1/N of the contending worker threads.
**New config:** `druid.processing.numThreadPools` (default `1`).
- `druid.processing.numThreads` is split as evenly as possible across the pools
(remainder to the first pools). Total thread count and processing-buffer sizing
are unchanged, so direct-memory accounting is unaffected.
- Tasks are routed with `ThreadLocalRandom` — no shared counter, so the router
adds no contention of its own.
- `numThreadPools = 1` (the default) preserves the current single-pool behavior
exactly; no config change is required and no metric names change.
**Implementation sketch:**
- A `ShardedPrioritizedExecutorService` composite holding N ordinary
`PrioritizedExecutorService` instances; it delegates per-task work
(`execute`/`submit`) to a random shard and fans out lifecycle calls.
- A small `ProcessingPoolStats` interface (implemented by both the single-pool
and sharded executors) so `segment/scan/pending` and `segment/scan/active`
keep being emitted — summed across shards in the sharded case.
- The processing-pool provider selects the sharded implementation only when
`numThreadPools > 1`.
**Trade-off:** priority ordering becomes **per-shard rather than global** — a
high-priority task in one shard does not preempt work queued in another. For the
high-throughput, effectively-single-priority per-segment workload this pool
serves, that is an acceptable exchange. Deployments that need strict global
priority ordering should keep the default (`numThreadPools = 1`).
### Motivation
On historicals (and any process using the processing pool), every per-segment
scan/merge task is submitted to a single `PrioritizedExecutorService` backed by
one `PriorityBlockingQueue`. Because a priority queue is a binary heap, it uses a
**single** `ReentrantLock` for both `put` and `take` (unlike `LinkedBlockingQueue`'s
two-lock design), so every task pays the lock **twice** — once when a producer
enqueues it and once when a worker dequeues it.
On workloads with very high task rates — many small segments scanned per query at
high QPS — this single lock becomes the bottleneck:
- All `numThreads` worker threads plus all producers serialize on one lock.
- Under contention the cost is dominated by `AbstractQueuedSynchronizer`
park/unpark (futex + context switch) and by cache-line bouncing on the AQS
`state` word across cores, which grows worse-than-linearly with the number of
contending threads.
- The symptom is high query wait time while CPU utilization stays low (the time
is spent parked, off-CPU), and it gets *worse* on larger many-core nodes
because more worker threads contend on the same lock.
In our deployment, once other hotspots were removed, lock profiling on a historical
showed this processing-queue `ReentrantLock` as the dominant remaining
lock-contention source, with the pool taking tasks at ~170–180k lock
acquisitions/sec through one queue. Reducing the processing-thread count per node
(and scaling out) measurably cut the contention, confirming the single queue —
not CPU or merge buffers — was the limiter. Sharding the queue removes the
bottleneck directly instead of working around it with node capacity, and lets a
single many-core node use its threads without collapsing onto one lock.
Contributor guide
Research direction
Start by locating ProcessingPoolStats, PrioritizedExecutorService, and the processing-pool provider, then read how execute, submit, lifecycle calls, and processing-pool metrics currently work. Verify the numThreadPools configuration and its default behavior before changing the design. Done means sharded queues can be selected above one pool, thread and buffer totals remain unchanged, metrics are summed, and the single-pool behavior is preserved.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- backend, performance
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 48/100