Weighted Split Scheduling #9059
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,77 @@ | ||
| /* | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package io.trino.execution; | ||
|
|
||
| import static com.google.common.base.MoreObjects.toStringHelper; | ||
|
|
||
| public final class PartitionedSplitsInfo | ||
| { | ||
| private static final PartitionedSplitsInfo NO_SPLITS_INFO = new PartitionedSplitsInfo(0, 0); | ||
|
|
||
| private final int count; | ||
| private final long weightSum; | ||
|
|
||
| private PartitionedSplitsInfo(int splitCount, long splitsWeightSum) | ||
| { | ||
| this.count = splitCount; | ||
| this.weightSum = splitsWeightSum; | ||
| } | ||
|
|
||
| public int getCount() | ||
| { | ||
| return count; | ||
| } | ||
|
|
||
| public long getWeightSum() | ||
| { | ||
| return weightSum; | ||
| } | ||
|
|
||
| @Override | ||
| public int hashCode() | ||
| { | ||
| return (count * 31) + Long.hashCode(weightSum); | ||
|
pettyjamesm marked this conversation as resolved.
Outdated
|
||
| } | ||
|
|
||
| @Override | ||
| public boolean equals(Object other) | ||
| { | ||
| if (!(other instanceof PartitionedSplitsInfo)) { | ||
| return false; | ||
| } | ||
| PartitionedSplitsInfo otherInfo = (PartitionedSplitsInfo) other; | ||
| return this == otherInfo || (this.count == otherInfo.count && this.weightSum == otherInfo.weightSum); | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() | ||
| { | ||
| return toStringHelper(this) | ||
| .add("count", count) | ||
| .add("weightSum", weightSum) | ||
| .toString(); | ||
| } | ||
|
|
||
| public static PartitionedSplitsInfo forSplitCountAndWeightSum(int splitCount, long weightSum) | ||
| { | ||
| // Avoid allocating for the "no splits" case, also mask potential race condition between | ||
|
Member
There was a problem hiding this comment. Is the race condition because we track count and weight separately? We should consider wrapping them in a holder object that gets updated atomically -- or just make the update to those fields atomic via
Member
Author
There was a problem hiding this comment. The potential for a partially inconsistent view on this mostly comes from the worker side In this case, we know that minor data races are possible but essentially benign. If what you observe in that snapshot is Incidentally, the global |
||
| // count and weight updates that might yield a positive weight with a count of 0 | ||
| return splitCount == 0 ? NO_SPLITS_INFO : new PartitionedSplitsInfo(splitCount, weightSum); | ||
| } | ||
|
|
||
| public static PartitionedSplitsInfo forZeroSplits() | ||
| { | ||
| return NO_SPLITS_INFO; | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It seems weird that this method would have side effects (
clearLocalSplitInfo()) if the preconditions are not met (partitionedSplits != null & partitionedSplits.XXX >= 0)What's the purpose of calling
clearLocalSplitInfounder those conditions instead of just bailing out?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The
TaskPartitionedSplitCountTrackeris sort of a specific sub-view of the globalNodeTaskMaptotal split state per worker, so when we're going to bail out and throw an exception we want to undo the effect of any task-specific values on the global state first. The existing logic was doing the the same thing when / if a negative value was encountered forlocalPartitionedSplitCount, now it's been extended to consider negative weights or null arguments too.