zhuzhurk 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_r355107324
########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/ResourceRequirementsRetriever.java ########## @@ -0,0 +1,127 @@ +/* + * 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.configuration.MemorySize; +import org.apache.flink.runtime.clusterframework.types.ResourceProfile; +import org.apache.flink.runtime.executiongraph.ExecutionJobVertex; +import org.apache.flink.runtime.instance.SlotSharingGroupId; +import org.apache.flink.runtime.jobgraph.JobVertexID; +import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup; +import org.apache.flink.runtime.shuffle.ShuffleMaster; + +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +import static org.apache.flink.util.Preconditions.checkNotNull; +import static org.apache.flink.util.Preconditions.checkState; + +/** + * Class that builds and host resource requirements for {@link ExecutionJobVertex} and {@link SlotSharingGroup}. + */ +public final class ResourceRequirementsRetriever { + + private final Map<JobVertexID, ResourceProfile> vertexResourceProfiles = new HashMap<>(); + + private final Map<SlotSharingGroupId, ResourceProfile> groupResourceProfiles = new HashMap<>(); + + public ResourceRequirementsRetriever( + final Map<JobVertexID, ExecutionJobVertex> vertices, + final ShuffleMaster<?> shuffleMaster) { + + checkNotNull(vertices); + checkNotNull(shuffleMaster); + + buildResourceRequirements(vertices, shuffleMaster); + } + + private void buildResourceRequirements( + final Map<JobVertexID, ExecutionJobVertex> vertices, + final ShuffleMaster<?> shuffleMaster) { + + final Set<SlotSharingGroup> slotSharingGroups = new HashSet<>(); + + for (ExecutionJobVertex vertex : vertices.values()) { + final SlotSharingGroup group = vertex.getSlotSharingGroup(); + if (group != null) { + slotSharingGroups.add(group); + } + + final ResourceProfile enrichedResourceProfile = getEnrichedResourceProfile(vertex, vertices, shuffleMaster); + vertexResourceProfiles.put(vertex.getJobVertexId(), enrichedResourceProfile); + } + + for (SlotSharingGroup group : slotSharingGroups) { + checkState(group.getJobVertexIds().size() > 0); + + ResourceProfile totalResources = null; + for (JobVertexID jobVertexID : group.getJobVertexIds()) { + final ResourceProfile vertexResources = getJobVertexResourceRequirement(jobVertexID); + totalResources = totalResources == null + ? vertexResources + : totalResources.merge(vertexResources); + } + groupResourceProfiles.put(group.getSlotSharingGroupId(), totalResources); + } + } + + /** + * Enrich the original resource profile with required shuffle memory. + */ + private ResourceProfile getEnrichedResourceProfile( + final ExecutionJobVertex vertex, + final Map<JobVertexID, ExecutionJobVertex> vertices, + final ShuffleMaster<?> shuffleMaster) { + + final MemorySize requiredShuffleMemory = shuffleMaster.getShuffleMemoryForTask( + ExecutionJobVertexTaskInputsOutputsDescriptorBuilder.buildTaskInputsOutputsDescriptor(vertex, vertices)); + + final ResourceProfile original = vertex.getResourceProfile(); + final ResourceProfile enriched; + if (original.equals(ResourceProfile.UNKNOWN)) { + enriched = ResourceProfile.UNKNOWN; Review comment: I think sooner or later we should do that, but not sure whether we should do it right here. Allocating slots regarding real shuffle memory is beneficial for jobs with UNKNOWN resources. Especially in yarn deployments, a task may keep failing on deployed to a TM with insufficient shuffle memory, but will not launch new TMs since the slot allocation succeeded. However, there are still a few open questions / possible issues for it: 1. Should we make shuffle memory independent from ResourceProfile? Currently resources of a ResourceProfile should be all unknown or all specified, and ResourceProfile.UNKNOWN as a whole is specially treated in many places. 2. A task would always fail to allocate a slot if it requires shuffle memory more than a single slot's shuffle memory. Currently a task can steal shuffle memory from other slots but this change prevents it in ahead. Note that this happens only if #10146 is merged but dynamic slot allocation is fully not supported. ---------------------------------------------------------------- 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
