Improve concurrency in TaskQueue #17828
Open
Conversation
This file contains 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
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java
Fixed
Show fixed
Hide fixed
gianm
approved these changes
Mar 27, 2025
| * 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); |
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.
There was a problem hiding this comment.
Updated. Thanks for the review.
cryptoe
reviewed
Mar 28, 2025
| @@ -280,18 +282,17 @@ public void start() | |||
| @LifecycleStop | |||
| public void stop() | |||
| { | |||
| giant.lock(); | |||
| giant.writeLock().lock(); | |||
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
| @@ -4297,7 +4297,7 @@ private Map<String, Task> getActiveTaskMap() | |||
| final Optional<TaskQueue> taskQueue = taskMaster.getTaskQueue(); | |||
| final List<Task> tasks; | |||
| if (taskQueue.isPresent()) { | |||
| tasks = taskQueue.get().getActiveTasksForDatasource(dataSource); | |||
| return taskQueue.get().getActiveTasksForDatasource(dataSource); | |||
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.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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: