From e224c9623493d6c4c2f3ff731fd3c72c0f448b19 Mon Sep 17 00:00:00 2001 From: Karthik Kambatla Date: Tue, 17 Jan 2017 17:01:03 -0800 Subject: [PATCH] YARN-5831. FairScheduler: Propagate allowPreemptionFrom flag all the way down to the app. (Yufei Gu via kasha) --- .../scheduler/fair/FSAppAttempt.java | 20 +++++++------- .../scheduler/fair/FSParentQueue.java | 15 ----------- .../scheduler/fair/FSQueue.java | 27 +++++++++++++++---- .../scheduler/fair/QueueManager.java | 15 +++-------- .../scheduler/fair/Schedulable.java | 8 +++++- .../scheduler/fair/FakeSchedulable.java | 5 ++++ .../scheduler/fair/TestFSLeafQueue.java | 1 - .../fair/TestFairSchedulerPreemption.java | 13 +++++++++ .../scheduler/fair/TestSchedulingPolicy.java | 5 ++++ 9 files changed, 65 insertions(+), 44 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSAppAttempt.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSAppAttempt.java index d7ed7d15f2..0715e3a385 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSAppAttempt.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSAppAttempt.java @@ -568,6 +568,10 @@ private Resource getPreemptedResources() { } boolean canContainerBePreempted(RMContainer container) { + if (!isPreemptable()) { + return false; + } + // Sanity check that the app owns this container if (!getLiveContainersMap().containsKey(container.getContainerId()) && !newlyAllocatedContainers.contains(container)) { @@ -581,17 +585,6 @@ boolean canContainerBePreempted(RMContainer container) { return false; } - // Check if any of the parent queues are not preemptable - // TODO (YARN-5831): Propagate the "preemptable" flag all the way down to - // the app to avoid recursing up every time. - for (FSQueue q = getQueue(); - !q.getQueueName().equals("root"); - q = q.getParent()) { - if (!q.isPreemptable()) { - return false; - } - } - // Check if the app's allocation will be over its fairshare even // after preempting this container Resource currentUsage = getResourceUsage(); @@ -1241,4 +1234,9 @@ public int hashCode() { public boolean equals(Object o) { return super.equals(o); } + + @Override + public boolean isPreemptable() { + return getQueue().isPreemptable(); + } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSParentQueue.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSParentQueue.java index 16570aa853..f2e5086b4e 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSParentQueue.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSParentQueue.java @@ -108,21 +108,6 @@ void recomputeSteadyShares() { } } - @Override - public void updatePreemptionVariables() { - super.updatePreemptionVariables(); - // For child queues - - readLock.lock(); - try { - for (FSQueue childQueue : childQueues) { - childQueue.updatePreemptionVariables(); - } - } finally { - readLock.unlock(); - } - } - @Override public Resource getDemand() { readLock.lock(); diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSQueue.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSQueue.java index d87668d8c5..ee4c35aa24 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSQueue.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSQueue.java @@ -91,6 +91,7 @@ public FSQueue(String name, FairScheduler scheduler, FSParentQueue parent) { this.queueEntity = new PrivilegedEntity(EntityType.QUEUE, name); this.metrics = FSQueueMetrics.forQueue(getName(), parent, true, scheduler.getConf()); this.parent = parent; + reinit(false); } /** @@ -98,10 +99,19 @@ public FSQueue(String name, FairScheduler scheduler, FSParentQueue parent) { * metrics. * This function is invoked when a new queue is created or reloading the * allocation configuration. + * + * @param recursive whether child queues should be reinitialized recursively */ - public void init() { + public void reinit(boolean recursive) { AllocationConfiguration allocConf = scheduler.getAllocationConfiguration(); allocConf.initFSQueue(this, scheduler); + updatePreemptionVariables(); + + if (recursive) { + for (FSQueue child : getChildQueues()) { + child.reinit(recursive); + } + } } public String getName() { @@ -307,6 +317,7 @@ void setFairSharePreemptionThreshold(float fairSharePreemptionThreshold) { this.fairSharePreemptionThreshold = fairSharePreemptionThreshold; } + @Override public boolean isPreemptable() { return preemptable; } @@ -329,7 +340,7 @@ public void update(Resource fairShare, boolean checkStarvation) { * Update the min/fair share preemption timeouts, threshold and preemption * disabled flag for this queue. */ - public void updatePreemptionVariables() { + private void updatePreemptionVariables() { // For min share timeout minSharePreemptionTimeout = scheduler.getAllocationConfiguration() .getMinSharePreemptionTimeout(getName()); @@ -348,9 +359,15 @@ public void updatePreemptionVariables() { if (fairSharePreemptionThreshold < 0 && parent != null) { fairSharePreemptionThreshold = parent.getFairSharePreemptionThreshold(); } - // For option whether allow preemption from this queue - preemptable = scheduler.getAllocationConfiguration() - .isPreemptable(getName()); + // For option whether allow preemption from this queue. + // If the parent is non-preemptable, this queue is non-preemptable as well, + // otherwise get the value from the allocation file. + if (parent != null && !parent.isPreemptable()) { + preemptable = false; + } else { + preemptable = scheduler.getAllocationConfiguration() + .isPreemptable(getName()); + } } /** diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/QueueManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/QueueManager.java index bca0ea411b..934bcfd4f2 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/QueueManager.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/QueueManager.java @@ -73,11 +73,12 @@ public FSParentQueue getRootQueue() { public void initialize(Configuration conf) throws IOException, SAXException, AllocationConfigurationException, ParserConfigurationException { rootQueue = new FSParentQueue("root", scheduler, null); - rootQueue.init(); queues.put(rootQueue.getName(), rootQueue); // Create the default queue getLeafQueue(YarnConfiguration.DEFAULT_QUEUE_NAME, true); + // Recursively reinitialize to propagate queue properties + rootQueue.reinit(true); } /** @@ -281,11 +282,9 @@ private FSQueue createNewQueues(FSQueueType queueType, queue = newParent; } - queue.init(); parent.addChildQueue(queue); setChildResourceLimits(parent, queue, queueConf); queues.put(queue.getName(), queue); - queue.updatePreemptionVariables(); // If we just created a leaf node, the newParent is null, but that's OK // because we only create a leaf node in the very last iteration. @@ -496,17 +495,11 @@ public void updateAllocationConfiguration(AllocationConfiguration queueConf) { } } } - rootQueue.recomputeSteadyShares(); - - for (FSQueue queue : queues.values()) { - queue.init(); - } + // Initialize all queues recursively + rootQueue.reinit(true); // Update steady fair shares for all queues rootQueue.recomputeSteadyShares(); - // Update the fair share preemption timeouts and preemption for all queues - // recursively - rootQueue.updatePreemptionVariables(); } /** diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/Schedulable.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/Schedulable.java index cf78405697..fcdc056577 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/Schedulable.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/Schedulable.java @@ -23,7 +23,6 @@ import org.apache.hadoop.yarn.api.records.Priority; import org.apache.hadoop.yarn.api.records.Resource; import org.apache.hadoop.yarn.server.resourcemanager.resource.ResourceWeights; -import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer; /** * A Schedulable represents an entity that can be scheduled such as an @@ -96,4 +95,11 @@ public interface Schedulable { /** Assign a fair share to this Schedulable. */ void setFairShare(Resource fairShare); + + /** + * Check whether the schedulable is preemptable. + * @return true if the schedulable is preemptable; + * false otherwise + */ + boolean isPreemptable(); } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FakeSchedulable.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FakeSchedulable.java index e802f42acd..36ff85e5a4 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FakeSchedulable.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FakeSchedulable.java @@ -137,4 +137,9 @@ public Resource getMaxShare() { @Override public void updateDemand() {} + + @Override + public boolean isPreemptable() { + return true; + } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFSLeafQueue.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFSLeafQueue.java index 98de8dbb56..2aed9bfdbf 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFSLeafQueue.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFSLeafQueue.java @@ -81,7 +81,6 @@ public void testUpdateDemand() { String queueName = "root.queue1"; FSLeafQueue schedulable = new FSLeafQueue(queueName, scheduler, null); - schedulable.init(); schedulable.setMaxShare(maxResource); assertEquals(schedulable.getMetrics().getMaxApps(), Integer.MAX_VALUE); assertEquals(schedulable.getMetrics().getSchedulingPolicy(), diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairSchedulerPreemption.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairSchedulerPreemption.java index 36ee6858ed..8bc6cf5324 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairSchedulerPreemption.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairSchedulerPreemption.java @@ -23,6 +23,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeUpdateSchedulerEvent; import org.junit.After; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import org.junit.Before; import org.junit.Test; @@ -166,6 +167,14 @@ private void setupCluster() throws IOException { // Create and add two nodes to the cluster addNode(NODE_CAPACITY_MULTIPLE * 1024, NODE_CAPACITY_MULTIPLE); addNode(NODE_CAPACITY_MULTIPLE * 1024, NODE_CAPACITY_MULTIPLE); + + // Verify if child-1 and child-2 are preemptable + FSQueue child1 = + scheduler.getQueueManager().getQueue("nonpreemptable.child-1"); + assertFalse(child1.isPreemptable()); + FSQueue child2 = + scheduler.getQueueManager().getQueue("nonpreemptable.child-2"); + assertFalse(child2.isPreemptable()); } private void sendEnoughNodeUpdatesToAssignFully() { @@ -197,6 +206,10 @@ private void submitApps(String queue1, String queue2) scheduler.update(); sendEnoughNodeUpdatesToAssignFully(); assertEquals(8, greedyApp.getLiveContainers().size()); + // Verify preemptable for queue and app attempt + assertTrue( + scheduler.getQueueManager().getQueue(queue1).isPreemptable() + == greedyApp.isPreemptable()); // Create an app that takes up all the resources on the cluster ApplicationAttemptId appAttemptId2 diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestSchedulingPolicy.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestSchedulingPolicy.java index 57c7301f7b..bd49ccaa2e 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestSchedulingPolicy.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestSchedulingPolicy.java @@ -345,6 +345,11 @@ public String toString() { ", weights:" + weights + ", demand:" + demand + ", minShare:" + minShare + "}"; } + + @Override + public boolean isPreemptable() { + return true; + } } }