tillrohrmann commented on a change in pull request #10462: [FLINK-15031][runtime] Calculate required shuffle memory before allocating slots if resources are specified URL: https://github.com/apache/flink/pull/10462#discussion_r354865620
########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/ExecutionJobVertexTaskInputsOutputsDescriptorBuilder.java ########## @@ -0,0 +1,119 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.flink.runtime.scheduler; + +import org.apache.flink.runtime.executiongraph.ExecutionJobVertex; +import org.apache.flink.runtime.executiongraph.IntermediateResult; +import org.apache.flink.runtime.jobgraph.DistributionPattern; +import org.apache.flink.runtime.jobgraph.IntermediateDataSet; +import org.apache.flink.runtime.jobgraph.IntermediateDataSetID; +import org.apache.flink.runtime.jobgraph.JobEdge; +import org.apache.flink.runtime.jobgraph.JobVertexID; +import org.apache.flink.runtime.shuffle.TaskInputsOutputsDescriptor; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.flink.util.Preconditions.checkState; + +/** + * Utils to build a {@link TaskInputsOutputsDescriptor} from an {@link ExecutionJobVertex}. + * The result descriptor would suites for an instance with max possible input channels and result subpartitions. + */ +final class ExecutionJobVertexTaskInputsOutputsDescriptorBuilder { + + static TaskInputsOutputsDescriptor buildTaskInputsOutputsDescriptor( + final ExecutionJobVertex executionJobVertex, + final Map<JobVertexID, ExecutionJobVertex> vertices) { + + final Map<IntermediateDataSetID, Integer> maxPossibleNumbersOfInputChannelsPerInstance = + getMaxPossibleNumbersOfInputChannelsPerInstance(executionJobVertex); + final Map<IntermediateDataSetID, Integer> maxPossibleNumbersOfResultSubpartitionsPerInstance = + getMaxPossibleNumbersOfResultSubpartitionsPerInstance(executionJobVertex, vertices); + + return TaskInputsOutputsDescriptor.from( + maxPossibleNumbersOfInputChannelsPerInstance, + maxPossibleNumbersOfResultSubpartitionsPerInstance); + } + + private static Map<IntermediateDataSetID, Integer> getMaxPossibleNumbersOfInputChannelsPerInstance( + final ExecutionJobVertex executionJobVertex) { + + Map<IntermediateDataSetID, Integer> maxPossibleNumbers = new HashMap<>(); + final List<JobEdge> inputEdges = executionJobVertex.getJobVertex().getInputs(); + for (int i = 0; i < inputEdges.size(); i++) { + final JobEdge inputEdge = inputEdges.get(i); + final IntermediateResult consumedResult = executionJobVertex.getInputs().get(i); + + // the inputs order should match in JobGraph and ExecutionGraph + checkState(consumedResult.getId().equals(inputEdge.getSourceId())); + + final int maxPossibleNumber = getMaxPossibleNumberOfEdgesToTarget( + executionJobVertex.getParallelism(), + consumedResult.getNumberOfAssignedPartitions(), + inputEdge.getDistributionPattern()); + maxPossibleNumbers.put(consumedResult.getId(), maxPossibleNumber); + } + + return maxPossibleNumbers; + } Review comment: I just want to point out that by assuming that a consumer always consumes from all of its inputs, me might break certain setups. For example, assume that we have a batch job with a a hash join and where the scheduler first schedules the build side and the join operator. Only after the build side is complete, we start the probe side. In this scenario we might make the job work with fewer network buffers than we calculate here. I also know that this is a bit of a corner case and I'm not entirely sure how important it is. ---------------------------------------------------------------- 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
