[hotfix][runtime] Directly pass ResourceCounter as the totalRequireme…

…nts to the RequirementMatcher As the ResourceCounter is immutable, we can directly pass it as a argument, this also give us chance to optimize the matching perfermance
tongcheng-elong · Feb 28, 2021 · 1ba2962 · 1ba2962
1 parent e6cf590
commit 1ba2962
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 8 deletions.
diff --git a/...src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultDeclarativeSlotPool.java b/...src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultDeclarativeSlotPool.java
@@ -212,7 +212,7 @@ private Optional<AllocatedSlot> matchOfferWithOutstandingRequirements(
  final Optional<ResourceProfile> match =
  requirementMatcher.match(
  slotOffer.getResourceProfile(),
- totalResourceRequirements.getResourcesWithCount(),
+ totalResourceRequirements,
  fulfilledResourceRequirements::getResourceCount);
 
  if (match.isPresent()) {

diff --git a/...n/java/org/apache/flink/runtime/resourcemanager/slotmanager/JobScopedResourceTracker.java b/...n/java/org/apache/flink/runtime/resourcemanager/slotmanager/JobScopedResourceTracker.java
@@ -87,7 +87,7 @@ public void notifyAcquiredResource(ResourceProfile resourceProfile) {
  private Optional<ResourceProfile> findMatchingRequirement(ResourceProfile resourceProfile) {
  return requirementMatcher.match(
  resourceProfile,
- resourceRequirements.getResourcesWithCount(),
+ resourceRequirements,
  resourceToRequirementMapping::getNumFulfillingResources);
  }
 

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/slots/DefaultRequirementMatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/slots/DefaultRequirementMatcher.java
@@ -18,8 +18,8 @@
 package org.apache.flink.runtime.slots;
 
 import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.util.ResourceCounter;
 
-import java.util.Collection;
 import java.util.Map;
 import java.util.Optional;
 import java.util.function.Function;
@@ -32,9 +32,17 @@ public class DefaultRequirementMatcher implements RequirementMatcher {
  @Override
  public Optional<ResourceProfile> match(
  ResourceProfile resourceProfile,
- Collection<Map.Entry<ResourceProfile, Integer>> totalRequirements,
+ ResourceCounter totalRequirements,
  Function<ResourceProfile, Integer> numAssignedResourcesLookup) {
- for (Map.Entry<ResourceProfile, Integer> requirementCandidate : totalRequirements) {
+ // Short-cut for fine-grained resource management. If there is already exactly equal
+ // requirement, we can directly match with it.
+ if (totalRequirements.getResourceCount(resourceProfile)
+ > numAssignedResourcesLookup.apply(resourceProfile)) {
+ return Optional.of(resourceProfile);
+ }
+
+ for (Map.Entry<ResourceProfile, Integer> requirementCandidate :
+ totalRequirements.getResourcesWithCount()) {
  ResourceProfile requirementProfile = requirementCandidate.getKey();
 
  // beware the order when matching resources to requirements, because

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/slots/RequirementMatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/slots/RequirementMatcher.java
@@ -18,9 +18,8 @@
 package org.apache.flink.runtime.slots;
 
 import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.util.ResourceCounter;
 
-import java.util.Collection;
-import java.util.Map;
 import java.util.Optional;
 import java.util.function.Function;
 
@@ -38,6 +37,6 @@ public interface RequirementMatcher {
  */
  Optional<ResourceProfile> match(
  ResourceProfile resourceProfile,
- Collection<Map.Entry<ResourceProfile, Integer>> totalRequirements,
+ ResourceCounter totalRequirements,
  Function<ResourceProfile, Integer> numAssignedResourcesLookup);
 }