rbankar7 opened a new pull request, #19938:
URL: https://github.com/apache/druid/pull/19938
Fixes #19937.
### Description
Under very high task rates, the processing pool's single task queue becomes a
lock-contention bottleneck. Every per-segment scan/merge task is submitted
to a
single `PrioritizedExecutorService` backed by one `PriorityBlockingQueue`,
which
—being a binary heap— is guarded by a **single** `ReentrantLock` for both
`put`
and `take` (unlike `LinkedBlockingQueue`'s two-lock design). So every task
pays
the lock twice, and all `numThreads` workers plus all producers serialize on
one
lock. On many-core nodes serving many small segments at high QPS, contention
on
this one lock (AQS park/unpark + cache-line bouncing on the AQS `state` word)
dominates, showing up as high query wait time while CPU stays low — and it
gets
worse as the node's thread count grows.
This PR adds an option to split the processing pool into N independent pools
("shards"), each with its own `PriorityBlockingQueue` and its own lock, so
each
queue lock only sees ~1/N of the submit/take traffic and ~1/N of the
contending
threads. It is fully opt-in and defaults to the existing single-pool
behavior.
See the issue (#19937) for the motivation and profiling background.
#### Added `druid.processing.numThreadPools` (default `1`)
`druid.processing.numThreads` is split as evenly as possible across the pools
(remainder to the first pools), so total thread count and processing-buffer
sizing are unchanged (direct-memory accounting is unaffected). The value is
clamped to `[1, numThreads]` (you can't have more pools than threads).
`numThreadPools = 1` preserves the current behavior exactly.
#### Added `ShardedPrioritizedExecutorService`
A composite `ListeningExecutorService` holding N ordinary
`PrioritizedExecutorService` instances. Per-task work (`execute`/`submit`) is
routed to a shard chosen with `ThreadLocalRandom` — no shared counter, so the
router adds no contention of its own — and the chosen shard does all the
priority wrapping and ordering. Lifecycle calls fan out to every shard;
`getQueueSize()`/`getActiveTasks()` are summed across shards.
**Design choice — routing:** random routing was chosen over round-robin
(which
needs a shared atomic counter, reintroducing a contended cache line) .
For a homogeneous, high-rate
task stream, random spreads load evenly enough that per-shard queue depths
stay
balanced.
**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 needing strict global
priority ordering should keep the default (`numThreadPools = 1`).
#### Added `ProcessingPoolStats` interface
Extracted a small interface (`getQueueSize()` / `getActiveTasks()`)
implemented
by both `PrioritizedExecutorService` and `ShardedPrioritizedExecutorService`.
`MetricsEmittingQueryProcessingPool` now emits `segment/scan/pending` and
`segment/scan/active` via `instanceof ProcessingPoolStats` instead of a
concrete
class, so those metrics keep working for both pool types (summed across
shards
for the sharded case) — no metric names change.
#### Wiring
`DruidProcessingModule.createProcessingExecutorPool` selects the sharded
implementation only when `numThreadPools > 1`; otherwise it uses the existing
`PrioritizedExecutorService` unchanged.
#### Release note
Added `druid.processing.numThreads Pools` (default `1`), which splits the
processing pool into that many independent thread pools/queues, each with
its own
lock. Increasing it relieves contention on the single processing-queue lock
under
very high task rates (e.g. Historicals scanning many small segments at high
QPS),
at the cost of per-pool rather than global task-priority ordering. The
default
(`1`) is identical to previous behavior.
<hr>
##### Key changed/added classes in this PR
* `ProcessingPoolStats` (new) — capability interface for pool
queue/active-task stats.
* `ShardedPrioritizedExecutorService` (new) — composite of N
`PrioritizedExecutorService` shards with random routing.
* `PrioritizedExecutorService` — implements `ProcessingPoolStats`;
extracted a shared `makeThreadPoolExecutor` helper; added a startup log line.
* `DruidProcessingConfig` — new `numThreadPools` property (clamped to `[1,
numThreads]`) + getter; backward-compatible constructor retained.
* `MetricsEmittingQueryProcessingPool` — emits scan metrics via
`ProcessingPoolStats`.
* `DruidProcessingModule` — selects the sharded pool when `numThreadPools >
1`.
<hr>
This PR has:
- [x] been self-reviewed.
- [x] using the [concurrency
checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md).
- [x] added documentation for new or modified features or behaviors.
- [x] a release note entry in the PR description.
- [x] added Javadocs for most classes and all non-trivial methods. Linked
related entities via Javadoc links.
- [x] added comments explaining the "why" and the intent of the code
wherever would not be obvious for an unfamiliar reader.
- [x] added unit tests or modified existing tests to cover new code paths.
- [ ] been tested in a test Druid cluster.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
