Repository navigation
Improve concurrency in TaskQueue - #17828
Merged
Merged
Conversation
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
gianm
approved these changes
Mar 27, 2025
gianm
left a comment
gianm
left a comment
Contributor
There was a problem hiding this comment.
LGTM, feel free to merge after considering the comment about the name of the giant lock.
| * queue management is in progress. {@link #start} and {@link #stop} methods | ||
| * acquire a WRITE lock whereas all other operations acquire a READ lock. | ||
| */ | ||
| private final ReentrantReadWriteLock giant = new ReentrantReadWriteLock(true); |
Contributor
There was a problem hiding this comment.
Consider renaming to startStopLock, since its main purpose now is to synchronize start and stop with each other and with the other operations.
Contributor
Author
There was a problem hiding this comment.
Updated. Thanks for the review.
cryptoe
reviewed
Mar 28, 2025
| public void stop() | ||
| { | ||
| giant.lock(); | ||
| giant.writeLock().lock(); |
Contributor
There was a problem hiding this comment.
It would be good we we add tests triggering this behavior ie
one threads is hitting start/stop method while the other is trying to insert tasks to activeTasks
Contributor
Author
There was a problem hiding this comment.
Added TaskQueueConcurrencyTest.
| final List<Task> tasks; | ||
| if (taskQueue.isPresent()) { | ||
| tasks = taskQueue.get().getActiveTasksForDatasource(dataSource); | ||
| return taskQueue.get().getActiveTasksForDatasource(dataSource); |
Contributor
There was a problem hiding this comment.
We can probably remove line 4296 and 4298 and fold them inside the else at 4301. The resulting method would be much leaner.
Contributor
Contributor
Author
|
@maytasm , this PR needs some more tests. I will add them soon and get the PR merged. |
cryptoe
approved these changes
May 19, 2025
riovic918data
added a commit
to riovic918data/druid
that referenced
this pull request
Jun 12, 2026
Description
The giant lock in TaskQueue is acquired when performing any update or read operation, such as:
add a task
remove a task
read the list of tasks
sync from storage
On large clusters with several concurrent tasks, this can become a bottleneck.
This patch attempts to improve the concurrency in TaskQueue by using a concurrent hash map instead.
Changes
Use a ConcurrentHashMap to track active tasks. The giant lock was needed only to handle
competing updates made to the same task ID. This can be handled by a ConcurrentHashMap too.
Perform any update on the entry for a task ID within ConcurrentHashMap.compute to ensure atomicity
Convert the giant lock from a ReentrantLock to a ReentrantReadWriteLock
Repurpose giant lock to ensure that the TaskQueue start and stop is mutually exclusive from any other
normal operation on the TaskQueue
Keep a lastUpdatedTime in every TaskEntry to handle race conditions when syncing from metadata store
Handling race conditions
The only possible race conditions are with syncFromStorage(),
and are handled by maintaining a lastUpdatedTime for every task entry.
All updates to a task entry happen inside a ConcurrentHashMap.compute(taskId, entry -> update(entry))
and are thus thread-safe.
[A] Sync should not add a task to queue if it has just been removed
This is handled by ensuring that only the syncFromStorage() method can remove tasks from the queue.
A task that completed after the poll started would have a lastUpdatedTime after syncStartTime.
In this case, since an entry already exists in the queue, a new one will not be added.
The next invocation of syncFromStorage() will add the task, if necessary.
[B] Sync should not remove a task from queue if it has just been added
A task added after the poll started would have a lastUpdatedTime after syncStartTime.
In this case, the entry will not be removed.
The next invocation of syncFromStorage() will clean up the task, if necessary.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
6 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
The
giantlock inTaskQueueis acquired when performing any update or read operation, such as:On large clusters with several concurrent tasks, this can become a bottleneck.
This patch attempts to improve the concurrency in
TaskQueueby using a concurrent hash map instead.Changes
ConcurrentHashMapto track active tasks. The giant lock was needed only to handlecompeting updates made to the same task ID. This can be handled by a
ConcurrentHashMaptoo.ConcurrentHashMap.computeto ensure atomicitygiantlock from aReentrantLockto aReentrantReadWriteLockgiantlock to ensure that theTaskQueuestartandstopis mutually exclusive from any othernormal operation on the
TaskQueuelastUpdatedTimein everyTaskEntryto handle race conditions when syncing from metadata storeHandling race conditions
The only possible race conditions are with
syncFromStorage(),and are handled by maintaining a
lastUpdatedTimefor every task entry.All updates to a task entry happen inside a
ConcurrentHashMap.compute(taskId, entry -> update(entry))and are thus thread-safe.
[A] Sync should not add a task to queue if it has just been removed
This is handled by ensuring that only the
syncFromStorage()method can remove tasks from the queue.A task that completed after the poll started would have a
lastUpdatedTimeaftersyncStartTime.In this case, since an entry already exists in the queue, a new one will not be added.
The next invocation of
syncFromStorage()will add the task, if necessary.[B] Sync should not remove a task from queue if it has just been added
A task added after the poll started would have a
lastUpdatedTimeaftersyncStartTime.In this case, the entry will not be removed.
The next invocation of
syncFromStorage()will clean up the task, if necessary.Pending
This PR has: