From 4bf5b4cec8f751c9b90f6f246e9f22327c66a568 Mon Sep 17 00:00:00 2001 From: Benjamin Teke Date: Wed, 20 Jul 2022 21:45:08 +0200 Subject: [PATCH 1/2] YARN-6538. Inter Queue preemption is not happening when DRF is configured in extreme resource scale cases. --- .../resource/DefaultResourceCalculator.java | 4 ++ .../resource/DominantResourceCalculator.java | 16 ++++++ .../util/resource/ResourceCalculator.java | 10 ++++ ...AbstractPreemptableResourceCalculator.java | 7 ++- ...cityPreemptionPolicyInterQueueWithDRF.java | 51 +++++++++++++++++++ 5 files changed, 86 insertions(+), 2 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DefaultResourceCalculator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DefaultResourceCalculator.java index a7b59806a0b4d..9e475f13d1478 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DefaultResourceCalculator.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DefaultResourceCalculator.java @@ -153,6 +153,10 @@ public boolean isAnyMajorResourceZeroOrNegative(Resource resource) { return resource.getMemorySize() <= 0; } + @Override + public boolean isAnyRequestedResourceZeroOrNegative(Resource available, Resource resource) { + return resource.getMemorySize() == 0; + } @Override public boolean isAnyMajorResourceAboveZero(Resource resource) { return resource.getMemorySize() > 0; diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DominantResourceCalculator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DominantResourceCalculator.java index 8fe9c7931a289..c21dae3721894 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DominantResourceCalculator.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DominantResourceCalculator.java @@ -642,6 +642,22 @@ public boolean isAnyMajorResourceZeroOrNegative(Resource resource) { return false; } + @Override + public boolean isAnyRequestedResourceZeroOrNegative(Resource available, Resource resource) { + int maxLength = ResourceUtils.getNumberOfCountableResourceTypes(); + for (int i = 0; i < maxLength; i++) { + ResourceInformation resourceInformation = available.getResourceInformation( + i); + if (resourceInformation.getValue() != 0L) { + ResourceInformation ri2 = resource.getResourceInformation(i); + if (ri2.getValue() <= 0L) { + return true; + } + } + } + return false; + } + @Override public boolean isAnyMajorResourceAboveZero(Resource resource) { int maxLength = ResourceUtils.getNumberOfCountableResourceTypes(); diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/ResourceCalculator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/ResourceCalculator.java index 089d19339786e..f234bd71a9c65 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/ResourceCalculator.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/ResourceCalculator.java @@ -278,6 +278,16 @@ public abstract float divide( */ public abstract boolean isAnyMajorResourceZeroOrNegative(Resource resource); + /** + * Check if resource has any available (those that have non-zero available value) + * major resource types (which are all NodeManagers included) a zero value or negative value. + * + * @param available available resource + * @param resource resource + * @return returns true if any resource is zero. + */ + public abstract boolean isAnyRequestedResourceZeroOrNegative(Resource available, Resource resource); + /** * Get resource rand normalize down using step-factor * stepFactor. diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/AbstractPreemptableResourceCalculator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/AbstractPreemptableResourceCalculator.java index c3c594512b07c..62f0f38a07aa2 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/AbstractPreemptableResourceCalculator.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/AbstractPreemptableResourceCalculator.java @@ -224,7 +224,11 @@ protected void computeFixpointAllocation(Resource totGuarant, for (Iterator i = underserved.iterator(); i .hasNext();) { - if (!rc.isAnyMajorResourceAboveZero(unassigned)) { + // Exit the loop once any of the unassigned resources reach zero, except the optional + // ones which the queue has no guarantee on. This should ensure that extreme cases + // where one of the resources are significantly larger than the other (i.e: way below 1 + // vcore per GB of memory) + if (rc.isAnyRequestedResourceZeroOrNegative(totGuarant, unassigned)) { break; } @@ -266,7 +270,6 @@ protected void computeFixpointAllocation(Resource totGuarant, } } - /** * This method is visible to allow sub-classes to override the initialization * behavior. diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyInterQueueWithDRF.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyInterQueueWithDRF.java index efd2ffc991cbd..38b7eb398a760 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyInterQueueWithDRF.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyInterQueueWithDRF.java @@ -31,6 +31,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.LeafQueue; import org.apache.hadoop.yarn.util.resource.DominantResourceCalculator; import org.apache.hadoop.yarn.util.resource.ResourceUtils; +import org.mockito.Matchers; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.argThat; @@ -89,6 +90,56 @@ public void testInterQueuePreemptionWithMultipleResource() throws Exception { getAppAttemptId(2)))); } + @Test + public void testInterQueuePreemptionWithMultipleResourceUnbalancedCase() throws IOException { + /** + * Queue structure is: + * + * 
+     *           root
+     *           /  \
+     *          a    b
+     *
+ * + * Scenario: Queue B : Queue A has 90/10% capacity share. + * Almost all the resources are allocated to Queue A, but based on memory alone + * containers still can be allocated elsewhere. The app on A requests a lot more. + * The app on Queue B needs a little memory (still fits in the remaining resources), + * but needs a lot of vcores. Enough resources must be preempted from the app on A + * to allow the app on B launch. + */ + String labelsConfig = "=65536:50,true;"; // default partition + String nodesConfig = // only one node + "n1="; + + String queuesConfig = + // guaranteed, max, used, pending + "root(=[65536:50 65536:50 61440:45 132096:525]);" + // root + "-a(=[2048:5 65536:50 61440:45 131072:500]);" + // a + "-b(=[63488:45 65536:50 0:0 1024:25]);"; // b + + + String appsConfig = + //queueName\t(priority,resource,host,expression,#repeat,reserved) + "a\t(1,61440:45,n1,,20,false);" + // app1 in a + "b\t(1,1024:25,n1,,30,false)"; // app2 in b + + buildEnv(labelsConfig, nodesConfig, queuesConfig, appsConfig, true); + Resource ul = Resource.newInstance(65536, 50); + when(((LeafQueue)(cs.getQueue("root.a"))) + .getResourceLimitForAllUsers(any(), any(), any(), any()) + ).thenReturn(ul); + policy.editSchedule(); + + // Preemption should happen in Queue a, preempt to Queue b + verify(eventHandler, times(1)).handle(Matchers.argThat( + new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor( + getAppAttemptId(1)))); + verify(eventHandler, never()).handle(Matchers.argThat( + new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor( + getAppAttemptId(2)))); + } + @Test public void testInterQueuePreemptionWithNaturalTerminationFactor() throws Exception { From 97d477d4dfdb8f5335493b959e131dae4558932f Mon Sep 17 00:00:00 2001 From: Benjamin Teke Date: Wed, 30 Nov 2022 08:39:29 +0100 Subject: [PATCH 2/2] Fix checkstyle and javac --- .../apache/hadoop/yarn/util/resource/ResourceCalculator.java | 3 ++- ...roportionalCapacityPreemptionPolicyInterQueueWithDRF.java | 5 ++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/ResourceCalculator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/ResourceCalculator.java index f234bd71a9c65..7bcfe932b89f6 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/ResourceCalculator.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/ResourceCalculator.java @@ -286,7 +286,8 @@ public abstract float divide( * @param resource resource * @return returns true if any resource is zero. */ - public abstract boolean isAnyRequestedResourceZeroOrNegative(Resource available, Resource resource); + public abstract boolean isAnyRequestedResourceZeroOrNegative( + Resource available, Resource resource); /** * Get resource rand normalize down using step-factor diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyInterQueueWithDRF.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyInterQueueWithDRF.java index 38b7eb398a760..aabea0910180d 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyInterQueueWithDRF.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyInterQueueWithDRF.java @@ -31,7 +31,6 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.LeafQueue; import org.apache.hadoop.yarn.util.resource.DominantResourceCalculator; import org.apache.hadoop.yarn.util.resource.ResourceUtils; -import org.mockito.Matchers; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.argThat; @@ -132,10 +131,10 @@ public void testInterQueuePreemptionWithMultipleResourceUnbalancedCase() throws policy.editSchedule(); // Preemption should happen in Queue a, preempt to Queue b - verify(eventHandler, times(1)).handle(Matchers.argThat( + verify(eventHandler, times(1)).handle(argThat( new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor( getAppAttemptId(1)))); - verify(eventHandler, never()).handle(Matchers.argThat( + verify(eventHandler, never()).handle(argThat( new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor( getAppAttemptId(2)))); }