azagrebin commented on a change in pull request #7631: [FLINK-11391][shuffle]
Introduce PartitionShuffleDescriptor and ShuffleDeploymentDescriptor
URL: https://github.com/apache/flink/pull/7631#discussion_r256385092
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
##########
@@ -813,24 +815,27 @@ TaskDeploymentDescriptor createDeploymentDescriptor(
boolean lazyScheduling =
getExecutionGraph().getScheduleMode().allowLazyDeployment();
for (IntermediateResultPartition partition :
resultPartitions.values()) {
-
List<List<ExecutionEdge>> consumers =
partition.getConsumers();
-
+ int maxParallelism;
if (consumers.isEmpty()) {
//TODO this case only exists for test,
currently there has to be exactly one consumer in real jobs!
-
producedPartitions.add(ResultPartitionDeploymentDescriptor.from(
- partition,
-
KeyGroupRangeAssignment.UPPER_BOUND_MAX_PARALLELISM,
- lazyScheduling));
+ maxParallelism =
KeyGroupRangeAssignment.UPPER_BOUND_MAX_PARALLELISM;
} else {
Preconditions.checkState(1 == consumers.size(),
- "Only one consumer supported in
the current implementation! Found: " + consumers.size());
+ "Only one consumer supported in the
current implementation! Found: " + consumers.size());
List<ExecutionEdge> consumer = consumers.get(0);
ExecutionJobVertex vertex =
consumer.get(0).getTarget().getJobVertex();
- int maxParallelism = vertex.getMaxParallelism();
-
producedPartitions.add(ResultPartitionDeploymentDescriptor.from(partition,
maxParallelism, lazyScheduling));
+ maxParallelism = vertex.getMaxParallelism();
}
+
+ PartitionShuffleDescriptor psd =
PartitionShuffleDescriptor.from(targetSlot, executionId, partition,
maxParallelism);
+
+
producedPartitions.add(ResultPartitionDeploymentDescriptor.fromShuffleDescriptor(psd));
+
getCurrentExecutionAttempt().cachePartitionShuffleDescriptor(partition.getIntermediateResult().getId(),
psd);
Review comment:
1. (and 2) ok, I agree, let's cache it similar to what we have now or maybe
we could already prepare `Map<IntermediateDataSetID, PartitionInfo>` on
producer side?
3. As I understand, the whole concurrent caching was done when consumer
`deploy` and `scheduleOrUpdateConsumers` happened concurrently. Now, it should
not be the case and all state transitions should happen on main thread of Job
Master. So any cache maps could be just a `HashMap`.
As far as I see, there are 2 cases:
- deploy the consumer: here we rely on `partition.isConsumable` flag to
decide whether the input channel SDD/Location is known or not at the moment
(`scheduleOrUpdateConsumers` has happened or not).
- `scheduleOrUpdateConsumers`:
- here first of all, `partition.isConsumable` flag is set to true before;
- In `CREATED` and `SCHEDULED` state we do not have to do anything because
`TDD` has not been created yet and when `TDD` is created in `deploy()`, it will
use already `partition.isConsumable = true` to populate known locations;
- then if some consumers are `DEPLOYING` or `RUNNING`, their `TDD`s have
been already sent with unknown `SDD/Location`s, so they have to be updated by
`sendUpdatePartitionInfoRpcCall` using cached `PartitionInfo`s, the update
message will be applied in Task after deploy message.
Currently in master, we call `sendPartitionInfos` to resolve previous race
conditions in several places. Now we basically do not need it along with
`partialInputChannelDeploymentDescriptors`.
What do you think?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
