azagrebin commented on a change in pull request #7822: [FLINK-11726][network]
Refactor the creation of ResultPartition and InputGate into NetworkEnvironment
URL: https://github.com/apache/flink/pull/7822#discussion_r267825533
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
##########
@@ -202,28 +233,97 @@ public TaskKvStateRegistry
createKvStateTaskRegistry(JobID jobId, JobVertexID jo
return kvStateRegistry.createTaskRegistry(jobId, jobVertexId);
}
+ public ResultPartition[] getResultPartitionWriters(String taskId) {
+ return allPartitions.get(taskId);
+ }
+
+ public SingleInputGate[] getInputGates(String taskId) {
+ return allInputGates.get(taskId);
+ }
+
//
--------------------------------------------------------------------------------------------
- // Task operations
+ // Create Output Writers and Input Readers
//
--------------------------------------------------------------------------------------------
- public void registerTask(Task task) throws IOException {
- final ResultPartition[] producedPartitions =
task.getProducedPartitions();
+ public ResultPartitionWriter[] createResultPartitionWriters(
+ String taskId,
+ JobID jobId,
+ ExecutionAttemptID executionId,
+ TaskActions taskActions,
+ ResultPartitionConsumableNotifier
partitionConsumableNotifier,
+ TaskIOMetricGroup metrics,
+ Collection<ResultPartitionDeploymentDescriptor>
resultPartitionDeploymentDescriptors) throws IOException {
+
+ if (isShutdown) {
+ throw new IllegalStateException("NetworkEnvironment is
shut down");
+ }
- synchronized (lock) {
- if (isShutdown) {
- throw new
IllegalStateException("NetworkEnvironment is shut down");
- }
+ ResultPartition[] parititons = new
ResultPartition[resultPartitionDeploymentDescriptors.size()];
+ int counter = 0;
+ for (ResultPartitionDeploymentDescriptor rpdd :
resultPartitionDeploymentDescriptors) {
+ parititons[counter] = new ResultPartition(
+ taskId,
+ taskActions,
+ jobId,
+ new ResultPartitionID(rpdd.getPartitionId(),
executionId),
+ rpdd.getPartitionType(),
+ rpdd.getNumberOfSubpartitions(),
+ rpdd.getMaxParallelism(),
+ resultPartitionManager,
+ partitionConsumableNotifier,
+ ioManager,
+ rpdd.sendScheduleOrUpdateConsumersMessage());
+
+ setupPartition(parititons[counter]);
+ counter++;
+ }
- for (final ResultPartition partition :
producedPartitions) {
- setupPartition(partition);
+ allPartitions.put(taskId, parititons);
+
+ metrics.initializeOutputBufferMetrics(parititons);
Review comment:
Actually. looking more into `ResultPartitionManager`, we use indexing by
execution id eventually only from task and task already knows its
partitions/gates. Maybe, if we move all these partition/gate release/close
for-loops from `NetworkEnvironment.unregisterTask` and
`ResultPartitionManager.releasePartitionsProducedBy` to `Task`, this would help
further to simplify future ShuffleService. If needed, netty `ResultPartition`
could remove itself from `ResultPartitionManager` in `release()` method. This
would be one step forward to get rid of "by execution id" indexing in
`NetworkEnvironment`/`ResultPartitionManager`.
Then we have only metrics with "task to partitions/gates" dependency. Here,
it does not seem to be trivial. In general, looking into
`TaskIOMetricGroup.initializeOutputBufferMetrics`,
`createResultPartitionWriters/gates` could just accept `ProxyMetricGroup<?>`
instead of `TaskIOMetricGroup`. Then `initializeOutputBufferMetrics` could go
into `NetworkEnvironment` using `ProxyMetricGroup.addGroup` with all netty
specific gauge classes.
Thinking more about register "per task" VS "per partition/gate", "per task"
is actually more restrictive and we do not need Shuffle API to be more flexible
for Task. Other shuffle services might also benefit from this way of
registration.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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
