Allow more leaf drivers to be scheduled if leaf drivers in other stages are blocked #27103
Conversation
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
Reviewer's GuideIntroduce tracking of blocked leaf drivers to free up scheduling slots when splits are blocked: SplitProcessor now notifies TaskEntry of blocked/running state, TaskEntry tracks blocked splits and updates concurrency accordingly, ThreadPerDriverTaskExecutor includes blocked splits in target calculation, and concurrency controller adds basic argument validation. Class diagram for updated SplitProcessor and TaskEntryclassDiagram
class SplitProcessor {
+TaskId taskId
+int splitId
+SplitRunner split
+Tracer tracer
+SplitBlockedStateChangeListener blockStateChangeListener
+run(SchedulerContext context)
<<interface>> SplitBlockedStateChangeListener
+splitBlocked()
+splitRunning()
}
class TaskEntry {
-int runningLeafSplits
-int blockedLeafSplits
-Set<SplitRunner> blockedLeafSplitSet
+runSplit(SplitRunner split, boolean trackLeafSplit)
+leafSplitDone(QueuedSplit split)
+blockedLeafSplits()
+updateConcurrency()
+onSplitBlocked(SplitRunner split)
+onSplitRunning(SplitRunner split)
}
SplitProcessor --> SplitProcessor.SplitBlockedStateChangeListener
TaskEntry --> SplitProcessor.SplitBlockedStateChangeListener
Class diagram for ThreadPerDriverTaskExecutor and ConcurrencyController changesclassDiagram
class ThreadPerDriverTaskExecutor {
-scheduleMoreLeafSplits()
}
class ConcurrencyController {
+update(double utilization, int currentConcurrency)
}
ThreadPerDriverTaskExecutor --> TaskEntry
TaskEntry --> ConcurrencyController
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey there - I've reviewed your changes - here's some feedback:
- In scheduleMoreLeafSplits, consider tracking a global blocked leaf split count rather than iterating tasks on every scheduling pass to avoid potential performance overhead when many tasks are present.
- Add a comment explaining why blockedLeafSplits are subtracted from runningLeafSplits in the concurrency update logic to clarify the intended scheduling impact for future maintainers.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In scheduleMoreLeafSplits, consider tracking a global blocked leaf split count rather than iterating tasks on every scheduling pass to avoid potential performance overhead when many tasks are present.
- Add a comment explaining why blockedLeafSplits are subtracted from runningLeafSplits in the concurrency update logic to clarify the intended scheduling impact for future maintainers.
## Individual Comments
### Comment 1
<location> `core/trino-main/src/main/java/io/trino/execution/executor/dedicated/TaskEntry.java:238-99` </location>
<code_context>
private record QueuedSplit(SplitRunner split, SettableFuture<Void> done) {}
+ private synchronized void onSplitBlocked(SplitRunner split)
+ {
+ if (blockedLeafSplitSet.add(split)) {
+ blockedLeafSplits++;
+ }
+ }
+
</code_context>
<issue_to_address>
**issue (bug_risk):** Blocked split tracking may not handle repeated state transitions correctly.
Repeated calls to splitBlocked or splitRunning for the same SplitRunner may lead to incorrect blockedLeafSplits counts. Please review if additional state validation is needed to maintain accurate tracking.
</issue_to_address>
### Comment 2
<location> `core/trino-main/src/main/java/io/trino/execution/executor/dedicated/SplitProcessor.java:100-107` </location>
<code_context>
}
}
else {
+ blockStateChangeListener.splitBlocked();
processSpan.addEvent("blocked");
processSpan.end();
if (!context.block(blocked)) {
processSpan = null;
return;
}
+ blockStateChangeListener.splitRunning();
}
processSpan = newSpan(splitSpan, processSpan);
</code_context>
<issue_to_address>
**suggestion:** Listener callbacks for splitBlocked and splitRunning may not cover all transition scenarios.
Consider cases where splits transition repeatedly or encounter errors; ensure the listener is notified for all relevant state changes and add hooks if necessary.
Suggested implementation:
```java
interface SplitBlockedStateChangeListener
{
void splitBlocked();
void splitRunning();
void splitUnblocked();
void splitError(Throwable error);
void splitCompleted();
```
```java
else {
blockStateChangeListener.splitBlocked();
processSpan.addEvent("blocked");
processSpan.end();
if (!context.block(blocked)) {
processSpan = null;
blockStateChangeListener.splitUnblocked();
return;
}
blockStateChangeListener.splitRunning();
}
processSpan = newSpan(splitSpan, processSpan);
```
```java
}
// Example: handle error scenario
catch (Throwable error) {
blockStateChangeListener.splitError(error);
throw error;
}
// Example: handle completion scenario
finally {
blockStateChangeListener.splitCompleted();
}
```
- You will need to implement the new methods (`splitUnblocked`, `splitError`, `splitCompleted`) in all classes that implement `SplitBlockedStateChangeListener`.
- Review all split state transitions in the file and ensure the appropriate listener method is called for each transition (blocked, running, unblocked, error, completed).
- If there are other places in the file where splits transition state (not shown in the provided code), add the relevant listener calls there as well.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
|
|
||
| for (SplitRunner split : running) { | ||
| split.close(); | ||
| } |
There was a problem hiding this comment.
issue (bug_risk): Blocked split tracking may not handle repeated state transitions correctly.
Repeated calls to splitBlocked or splitRunning for the same SplitRunner may lead to incorrect blockedLeafSplits counts. Please review if additional state validation is needed to maintain accurate tracking.
Comment on lines
+100
to
+107
| blockStateChangeListener.splitBlocked(); | ||
| processSpan.addEvent("blocked"); | ||
| processSpan.end(); | ||
| if (!context.block(blocked)) { | ||
| processSpan = null; | ||
| return; | ||
| } | ||
| blockStateChangeListener.splitRunning(); |
There was a problem hiding this comment.
suggestion: Listener callbacks for splitBlocked and splitRunning may not cover all transition scenarios.
Consider cases where splits transition repeatedly or encounter errors; ensure the listener is notified for all relevant state changes and add hooks if necessary.
Suggested implementation:
interface SplitBlockedStateChangeListener
{
void splitBlocked();
void splitRunning();
void splitUnblocked();
void splitError(Throwable error);
void splitCompleted(); else {
blockStateChangeListener.splitBlocked();
processSpan.addEvent("blocked");
processSpan.end();
if (!context.block(blocked)) {
processSpan = null;
blockStateChangeListener.splitUnblocked();
return;
}
blockStateChangeListener.splitRunning();
}
processSpan = newSpan(splitSpan, processSpan); }
// Example: handle error scenario
catch (Throwable error) {
blockStateChangeListener.splitError(error);
throw error;
}
// Example: handle completion scenario
finally {
blockStateChangeListener.splitCompleted();
}- You will need to implement the new methods (
splitUnblocked,splitError,splitCompleted) in all classes that implementSplitBlockedStateChangeListener. - Review all split state transitions in the file and ensure the appropriate listener method is called for each transition (blocked, running, unblocked, error, completed).
- If there are other places in the file where splits transition state (not shown in the provided code), add the relevant listener calls there as well.
|
This pull request has gone a while without any activity. Ask for help on #core-dev on Trino slack. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
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
Motivation behind this change is described in #27121
Additional context and related issues
Release notes
( ) This is not user-visible or is docs only, and no release notes are required.
( ) Release notes are required. Please propose a release note for me.
( ) Release notes are required, with the following suggested text:
Summary by Sourcery
Enable scheduling of additional leaf drivers when existing drivers are blocked by tracking blocked splits and adjusting concurrency and scheduling logic.
Bug Fixes:
Enhancements: