You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by ka...@apache.org on 2016/11/17 02:02:52 UTC
[50/50] hadoop git commit: YARN-5885. [YARN-4752] Cleanup YARN-4752
for merge (Contributed by Karthik Kambatla via Daniel Templeton)
YARN-5885. [YARN-4752] Cleanup YARN-4752 for merge (Contributed by Karthik Kambatla via Daniel Templeton)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/59bc9d16
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/59bc9d16
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/59bc9d16
Branch: refs/heads/YARN-4752
Commit: 59bc9d16a8514d65e893f4f30979973c54d2b0a3
Parents: a69763b
Author: Daniel Templeton <te...@apache.org>
Authored: Wed Nov 16 17:18:33 2016 -0800
Committer: Karthik Kambatla <ka...@cloudera.com>
Committed: Wed Nov 16 17:58:36 2016 -0800
----------------------------------------------------------------------
.../scheduler/fair/FSAppAttempt.java | 61 +++++--
.../scheduler/fair/FSLeafQueue.java | 37 ++++-
.../scheduler/fair/FSPreemptionThread.java | 21 ++-
.../scheduler/fair/FSSchedulerNode.java | 8 +-
.../scheduler/fair/FairScheduler.java | 7 -
.../scheduler/fair/TestFSAppStarvation.java | 48 ++++--
.../scheduler/fair/TestFSLeafQueue.java | 164 +------------------
.../fair/TestQueueManagerRealScheduler.java | 128 +++++++++++++++
8 files changed, 257 insertions(+), 217 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/59bc9d16/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
----------------------------------------------------------------------
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 f5bc2cd..39f4a3d 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
@@ -28,6 +28,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import com.google.common.annotations.VisibleForTesting;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.Private;
@@ -81,12 +82,14 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
private Resource fairShare = Resources.createResource(0, 0);
// Preemption related variables
- private Resource fairshareStarvation = Resources.none();
- private Resource minshareStarvation = Resources.none();
private final Resource preemptedResources = Resources.clone(Resources.none());
private final Set<RMContainer> containersToPreempt = new HashSet<>();
+ private Resource fairshareStarvation = Resources.none();
private long lastTimeAtFairShare;
+ // minShareStarvation attributed to this application by the leaf queue
+ private Resource minshareStarvation = Resources.none();
+
// Used to record node reservation by an app.
// Key = RackName, Value = Set of Nodes reserved by app on rack
private Map<String, Set<String>> reservations = new HashMap<>();
@@ -149,7 +152,7 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
// Remove from the list of containers
liveContainers.remove(rmContainer.getContainerId());
- removePreemption(rmContainer);
+ untrackContainerForPreemption(rmContainer);
Resource containerResource = rmContainer.getContainer().getResource();
RMAuditLogger.logSuccess(getUser(), AuditConstants.RELEASE_CONTAINER,
@@ -510,26 +513,42 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
}
// Preemption related methods
+
+ /**
+ * Get overall starvation - fairshare and attributed minshare.
+ *
+ * @return total starvation attributed to this application
+ */
Resource getStarvation() {
return Resources.add(fairshareStarvation, minshareStarvation);
}
+ /**
+ * Set the minshare attributed to this application. To be called only from
+ * {@link FSLeafQueue#updateStarvedApps}.
+ *
+ * @param starvation minshare starvation attributed to this app
+ */
void setMinshareStarvation(Resource starvation) {
this.minshareStarvation = starvation;
}
+ /**
+ * Reset the minshare starvation attributed to this application. To be
+ * called only from {@link FSLeafQueue#updateStarvedApps}
+ */
void resetMinshareStarvation() {
this.minshareStarvation = Resources.none();
}
- void addPreemption(RMContainer container) {
+ void trackContainerForPreemption(RMContainer container) {
containersToPreempt.add(container);
synchronized (preemptedResources) {
Resources.addTo(preemptedResources, container.getAllocatedResource());
}
}
- void removePreemption(RMContainer container) {
+ private void untrackContainerForPreemption(RMContainer container) {
synchronized (preemptedResources) {
Resources.subtractFrom(preemptedResources,
container.getAllocatedResource());
@@ -540,7 +559,6 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
Set<RMContainer> getPreemptionContainers() {
return containersToPreempt;
}
-
private Resource getPreemptedResources() {
synchronized (preemptedResources) {
@@ -563,8 +581,8 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
}
// Check if any of the parent queues are not preemptable
- // TODO (KK): Propagate the "preemptable" flag all the way down to the app
- // to avoid recursing up every time.
+ // 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()) {
@@ -585,8 +603,9 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
/**
* Create and return a container object reflecting an allocation for the
- * given appliction on the given node with the given capability and
+ * given application on the given node with the given capability and
* priority.
+ *
* @param node Node
* @param capability Capability
* @param schedulerKey Scheduler Key
@@ -1076,6 +1095,15 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
return appSchedulingInfo.getNextResourceRequest();
}
+ /**
+ * Helper method that captures if this app is identified to be starved.
+ * @return true if the app is starved for fairshare, false otherwise
+ */
+ @VisibleForTesting
+ boolean isStarvedForFairShare() {
+ return !Resources.isNone(fairshareStarvation);
+ }
+
/* Schedulable methods implementation */
@Override
@@ -1105,14 +1133,13 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
@Override
public Resource getResourceUsage() {
- // Here the getPreemptedResources() always return zero, except in
- // a preemption round
- // In the common case where preempted resource is zero, return the
- // current consumption Resource object directly without calling
- // Resources.subtract which creates a new Resource object for each call.
- return getPreemptedResources().equals(Resources.none()) ?
- getCurrentConsumption() :
- Resources.subtract(getCurrentConsumption(), getPreemptedResources());
+ /*
+ * getResourcesToPreempt() returns zero, except when there are containers
+ * to preempt. Avoid creating an object in the common case.
+ */
+ return getPreemptedResources().equals(Resources.none())
+ ? getCurrentConsumption()
+ : Resources.subtract(getCurrentConsumption(), getPreemptedResources());
}
@Override
http://git-wip-us.apache.org/repos/asf/hadoop/blob/59bc9d16/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSLeafQueue.java
----------------------------------------------------------------------
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/FSLeafQueue.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/FSLeafQueue.java
index 3fcf627..343e9c3 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/FSLeafQueue.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/FSLeafQueue.java
@@ -49,8 +49,9 @@ import static org.apache.hadoop.yarn.util.resource.Resources.none;
@Private
@Unstable
public class FSLeafQueue extends FSQueue {
- private static final Log LOG = LogFactory.getLog(
- FSLeafQueue.class.getName());
+ private static final Log LOG = LogFactory.getLog(FSLeafQueue.class.getName());
+ private static final List<FSQueue> EMPTY_LIST = Collections.emptyList();
+
private FairScheduler scheduler;
private FSContext context;
@@ -71,7 +72,6 @@ public class FSLeafQueue extends FSQueue {
private Resource amResourceUsage;
private final ActiveUsersManager activeUsersManager;
- private static final List<FSQueue> EMPTY_LIST = Collections.emptyList();
public FSLeafQueue(String name, FairScheduler scheduler,
FSParentQueue parent) {
@@ -210,7 +210,7 @@ public class FSLeafQueue extends FSQueue {
try {
policy.computeShares(runnableApps, getFairShare());
if (checkStarvation) {
- updatedStarvedApps();
+ updateStarvedApps();
}
} finally {
readLock.unlock();
@@ -234,7 +234,7 @@ public class FSLeafQueue extends FSQueue {
* one application that is starved. And, even if the queue is not
* starved due to fairshare, there might still be starved applications.
*/
- private void updatedStarvedApps() {
+ private void updateStarvedApps() {
// First identify starved applications and track total amount of
// starvation (in resources)
Resource fairShareStarvation = Resources.clone(none());
@@ -549,10 +549,33 @@ public class FSLeafQueue extends FSQueue {
/**
* Helper method for tests to check if a queue is starved for minShare.
- * @return whether starved for minShare.
+ * @return whether starved for minshare
*/
@VisibleForTesting
- boolean isStarvedForMinShare() {
+ private boolean isStarvedForMinShare() {
return !Resources.isNone(minShareStarvation());
}
+
+ /**
+ * Helper method for tests to check if a queue is starved for fairshare.
+ * @return whether starved for fairshare
+ */
+ @VisibleForTesting
+ private boolean isStarvedForFairShare() {
+ for (FSAppAttempt app : runnableApps) {
+ if (app.isStarvedForFairShare()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Helper method for tests to check if a queue is starved.
+ * @return whether starved for either minshare or fairshare
+ */
+ @VisibleForTesting
+ boolean isStarved() {
+ return isStarvedForMinShare() || isStarvedForFairShare();
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/59bc9d16/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSPreemptionThread.java
----------------------------------------------------------------------
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/FSPreemptionThread.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/FSPreemptionThread.java
index 01c830c..3579857 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/FSPreemptionThread.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/FSPreemptionThread.java
@@ -36,14 +36,14 @@ import java.util.TimerTask;
/**
* Thread that handles FairScheduler preemption.
*/
-public class FSPreemptionThread extends Thread {
+class FSPreemptionThread extends Thread {
private static final Log LOG = LogFactory.getLog(FSPreemptionThread.class);
protected final FSContext context;
private final FairScheduler scheduler;
private final long warnTimeBeforeKill;
private final Timer preemptionTimer;
- public FSPreemptionThread(FairScheduler scheduler) {
+ FSPreemptionThread(FairScheduler scheduler) {
this.scheduler = scheduler;
this.context = scheduler.getContext();
FairSchedulerConfiguration fsConf = scheduler.getConf();
@@ -80,8 +80,10 @@ public class FSPreemptionThread extends Thread {
* Given an app, identify containers to preempt to satisfy the app's next
* resource request.
*
- * @param starvedApp
- * @return
+ * @param starvedApp starved application for which we are identifying
+ * preemption targets
+ * @return list of containers to preempt to satisfy starvedApp, null if the
+ * app cannot be satisfied by preempting any running containers
*/
private List<RMContainer> identifyContainersToPreempt(
FSAppAttempt starvedApp) {
@@ -103,14 +105,13 @@ public class FSPreemptionThread extends Thread {
// Reset containers for the new node being considered.
containers.clear();
+ // TODO (YARN-5829): Attempt to reserve the node for starved app. The
+ // subsequent if-check needs to be reworked accordingly.
FSAppAttempt nodeReservedApp = node.getReservedAppSchedulable();
if (nodeReservedApp != null && !nodeReservedApp.equals(starvedApp)) {
// This node is already reserved by another app. Let us not consider
// this for preemption.
continue;
-
- // TODO (KK): If the nodeReservedApp is over its fairshare, may be it
- // is okay to unreserve it if we find enough resources.
}
// Figure out list of containers to consider
@@ -137,13 +138,15 @@ public class FSPreemptionThread extends Thread {
// FSSchedulerNode#removeContainerForPreemption.
node.addContainersForPreemption(containers);
return containers;
+ } else {
+ // TODO (YARN-5829): Unreserve the node for the starved app.
}
}
}
return null;
}
- public void preemptContainers(List<RMContainer> containers) {
+ private void preemptContainers(List<RMContainer> containers) {
// Warn application about containers to be killed
for (RMContainer container : containers) {
ApplicationAttemptId appAttemptId = container.getApplicationAttemptId();
@@ -151,7 +154,7 @@ public class FSPreemptionThread extends Thread {
FSLeafQueue queue = app.getQueue();
LOG.info("Preempting container " + container +
" from queue " + queue.getName());
- app.addPreemption(container);
+ app.trackContainerForPreemption(container);
}
// Schedule timer task to kill containers
http://git-wip-us.apache.org/repos/asf/hadoop/blob/59bc9d16/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSSchedulerNode.java
----------------------------------------------------------------------
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/FSSchedulerNode.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/FSSchedulerNode.java
index a605af6..a27a222 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/FSSchedulerNode.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/FSSchedulerNode.java
@@ -105,7 +105,7 @@ public class FSSchedulerNode extends SchedulerNode {
this.reservedAppSchedulable = null;
}
- public synchronized FSAppAttempt getReservedAppSchedulable() {
+ synchronized FSAppAttempt getReservedAppSchedulable() {
return reservedAppSchedulable;
}
@@ -118,14 +118,14 @@ public class FSSchedulerNode extends SchedulerNode {
*
* @param containers container to mark
*/
- public void addContainersForPreemption(Collection<RMContainer> containers) {
+ void addContainersForPreemption(Collection<RMContainer> containers) {
containersForPreemption.addAll(containers);
}
/**
* @return set of containers marked for preemption.
*/
- public Set<RMContainer> getContainersForPreemption() {
+ Set<RMContainer> getContainersForPreemption() {
return containersForPreemption;
}
@@ -134,7 +134,7 @@ public class FSSchedulerNode extends SchedulerNode {
*
* @param container container to remove
*/
- public void removeContainerForPreemption(RMContainer container) {
+ void removeContainerForPreemption(RMContainer container) {
containersForPreemption.remove(container);
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/59bc9d16/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
----------------------------------------------------------------------
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/FairScheduler.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/FairScheduler.java
index abe8a6a..571f2e6 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/FairScheduler.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/FairScheduler.java
@@ -998,13 +998,6 @@ public class FairScheduler extends
* Check if preemption is enabled and the utilization threshold for
* preemption is met.
*
- * TODO (KK): Should we handle the case where usage is less than preemption
- * threshold, but there are applications requesting resources on nodes that
- * are otherwise occupied by long running applications over their
- * fairshare? What if they are occupied by applications not over their
- * fairshare? Does this mean YARN should not allocate all resources on a
- * node to long-running services?
- *
* @return true if preemption should be attempted, false otherwise.
*/
private boolean shouldAttemptPreemption() {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/59bc9d16/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFSAppStarvation.java
----------------------------------------------------------------------
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/TestFSAppStarvation.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/TestFSAppStarvation.java
index 323152d..a5b2d86 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/TestFSAppStarvation.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/TestFSAppStarvation.java
@@ -41,10 +41,12 @@ import java.io.PrintWriter;
*/
public class TestFSAppStarvation extends FairSchedulerTestBase {
- private static final File ALLOC_FILE = new File(TEST_DIR, "test-queues");
+ private static final File ALLOC_FILE = new File(TEST_DIR, "test-QUEUES");
// Node Capacity = NODE_CAPACITY_MULTIPLE * (1 GB or 1 vcore)
private static final int NODE_CAPACITY_MULTIPLE = 4;
+ private static final String[] QUEUES =
+ {"no-preemption", "minshare", "fairshare.child", "drf.child"};
private FairSchedulerWithMockPreemption.MockPreemptionThread preemptionThread;
@@ -93,13 +95,15 @@ public class TestFSAppStarvation extends FairSchedulerTestBase {
assertNotNull("FSContext does not have an FSStarvedApps instance",
scheduler.getContext().getStarvedApps());
- assertEquals("Expecting 2 starved applications, one each for the " +
- "minshare and fairshare queues", 2,
- preemptionThread.uniqueAppsAdded());
+ assertEquals("Expecting 3 starved applications, one each for the "
+ + "minshare and fairshare queues",
+ 3, preemptionThread.uniqueAppsAdded());
// Verify the apps get added again on a subsequent update
scheduler.update();
Thread.yield();
+
+ verifyLeafQueueStarvation();
assertTrue("Each app is marked as starved exactly once",
preemptionThread.totalAppsAdded() > preemptionThread.uniqueAppsAdded());
}
@@ -121,6 +125,16 @@ public class TestFSAppStarvation extends FairSchedulerTestBase {
preemptionThread.totalAppsAdded());
}
+ private void verifyLeafQueueStarvation() {
+ for (String q : QUEUES) {
+ if (!q.equals("no-preemption")) {
+ boolean isStarved =
+ scheduler.getQueueManager().getLeafQueue(q, false).isStarved();
+ assertTrue(isStarved);
+ }
+ }
+ }
+
private void setupClusterAndSubmitJobs() throws Exception {
setupStarvedCluster();
submitAppsToEachLeafQueue();
@@ -167,21 +181,24 @@ public class TestFSAppStarvation extends FairSchedulerTestBase {
out.println("<minResources>2048mb,2vcores</minResources>");
out.println("</queue>");
- // Queue with fairshare preemption enabled
+ // FAIR queue with fairshare preemption enabled
out.println("<queue name=\"fairshare\">");
out.println("<fairSharePreemptionThreshold>1" +
"</fairSharePreemptionThreshold>");
out.println("<fairSharePreemptionTimeout>0" +
"</fairSharePreemptionTimeout>");
+ out.println("<schedulingPolicy>fair</schedulingPolicy>");
+ addChildQueue(out);
+ out.println("</queue>");
- // Child queue under fairshare with same settings
- out.println("<queue name=\"child\">");
+ // DRF queue with fairshare preemption enabled
+ out.println("<queue name=\"drf\">");
out.println("<fairSharePreemptionThreshold>1" +
"</fairSharePreemptionThreshold>");
out.println("<fairSharePreemptionTimeout>0" +
"</fairSharePreemptionTimeout>");
- out.println("</queue>");
-
+ out.println("<schedulingPolicy>drf</schedulingPolicy>");
+ addChildQueue(out);
out.println("</queue>");
out.println("</allocations>");
@@ -210,9 +227,18 @@ public class TestFSAppStarvation extends FairSchedulerTestBase {
assertEquals(8, scheduler.getSchedulerApp(app).getLiveContainers().size());
}
+ private void addChildQueue(PrintWriter out) {
+ // Child queue under fairshare with same settings
+ out.println("<queue name=\"child\">");
+ out.println("<fairSharePreemptionThreshold>1" +
+ "</fairSharePreemptionThreshold>");
+ out.println("<fairSharePreemptionTimeout>0" +
+ "</fairSharePreemptionTimeout>");
+ out.println("</queue>");
+ }
+
private void submitAppsToEachLeafQueue() {
- String queues[] = {"no-preemption", "minshare", "fairshare.child"};
- for (String queue : queues) {
+ for (String queue : QUEUES) {
createSchedulingRequest(1024, 1, "root." + queue, "user", 1);
}
scheduler.update();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/59bc9d16/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
----------------------------------------------------------------------
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 b8f4a4d..98de8db 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
@@ -105,12 +105,8 @@ public class TestFSLeafQueue extends FairSchedulerTestBase {
PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
out.println("<?xml version=\"1.0\"?>");
out.println("<allocations>");
- out.println("<queue name=\"queueA\">");
- out.println("<minResources>2048mb,0vcores</minResources>");
- out.println("</queue>");
- out.println("<queue name=\"queueB\">");
- out.println("<minResources>2048mb,0vcores</minResources>");
- out.println("</queue>");
+ out.println("<queue name=\"queueA\"></queue>");
+ out.println("<queue name=\"queueB\"></queue>");
out.println("</allocations>");
out.close();
@@ -143,162 +139,6 @@ public class TestFSLeafQueue extends FairSchedulerTestBase {
scheduler.update();
Collection<FSLeafQueue> queues = scheduler.getQueueManager().getLeafQueues();
assertEquals(3, queues.size());
-
- // Queue A should be above min share, B below.
- FSLeafQueue queueA =
- scheduler.getQueueManager().getLeafQueue("queueA", false);
- FSLeafQueue queueB =
- scheduler.getQueueManager().getLeafQueue("queueB", false);
-// TODO: assertFalse(queueA.isStarvedForMinShare());
-// TODO: assertTrue(queueB.isStarvedForMinShare());
-
- // Node checks in again, should allocate for B
- scheduler.handle(nodeEvent2);
- // Now B should have min share ( = demand here)
-// TODO: assertFalse(queueB.isStarvedForMinShare());
- }
-
- @Test (timeout = 5000)
- public void testIsStarvedForFairShare() throws Exception {
- conf.set(FairSchedulerConfiguration.ALLOCATION_FILE, ALLOC_FILE);
- PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
- out.println("<?xml version=\"1.0\"?>");
- out.println("<allocations>");
- out.println("<queue name=\"queueA\">");
- out.println("<weight>.2</weight>");
- out.println("</queue>");
- out.println("<queue name=\"queueB\">");
- out.println("<weight>.8</weight>");
- out.println("<fairSharePreemptionThreshold>.4</fairSharePreemptionThreshold>");
- out.println("<queue name=\"queueB1\">");
- out.println("</queue>");
- out.println("<queue name=\"queueB2\">");
- out.println("<fairSharePreemptionThreshold>.6</fairSharePreemptionThreshold>");
- out.println("</queue>");
- out.println("</queue>");
- out.println("<defaultFairSharePreemptionThreshold>.5</defaultFairSharePreemptionThreshold>");
- out.println("</allocations>");
- out.close();
-
- resourceManager = new MockRM(conf);
- resourceManager.start();
- scheduler = (FairScheduler) resourceManager.getResourceScheduler();
-
- // Add one big node (only care about aggregate capacity)
- RMNode node1 =
- MockNodes.newNodeInfo(1, Resources.createResource(10 * 1024, 10), 1,
- "127.0.0.1");
- NodeAddedSchedulerEvent nodeEvent1 = new NodeAddedSchedulerEvent(node1);
- scheduler.handle(nodeEvent1);
-
- scheduler.update();
-
- // Queue A wants 4 * 1024. Node update gives this all to A
- createSchedulingRequest(1 * 1024, "queueA", "user1", 4);
- scheduler.update();
- NodeUpdateSchedulerEvent nodeEvent2 = new NodeUpdateSchedulerEvent(node1);
- for (int i = 0; i < 4; i ++) {
- scheduler.handle(nodeEvent2);
- }
-
- QueueManager queueMgr = scheduler.getQueueManager();
- FSLeafQueue queueA = queueMgr.getLeafQueue("queueA", false);
- assertEquals(4 * 1024, queueA.getResourceUsage().getMemorySize());
-
- // Both queue B1 and queue B2 want 3 * 1024
- createSchedulingRequest(1 * 1024, "queueB.queueB1", "user1", 3);
- createSchedulingRequest(1 * 1024, "queueB.queueB2", "user1", 3);
- scheduler.update();
- for (int i = 0; i < 4; i ++) {
- scheduler.handle(nodeEvent2);
- }
-
- FSLeafQueue queueB1 = queueMgr.getLeafQueue("queueB.queueB1", false);
- FSLeafQueue queueB2 = queueMgr.getLeafQueue("queueB.queueB2", false);
- assertEquals(2 * 1024, queueB1.getResourceUsage().getMemorySize());
- assertEquals(2 * 1024, queueB2.getResourceUsage().getMemorySize());
-
- // For queue B1, the fairSharePreemptionThreshold is 0.4, and the fair share
- // threshold is 1.6 * 1024
-// TODO: assertFalse(queueB1.isStarvedForFairShare());
-
- // For queue B2, the fairSharePreemptionThreshold is 0.6, and the fair share
- // threshold is 2.4 * 1024
-// TODO: assertTrue(queueB2.isStarvedForFairShare());
-
- // Node checks in again
- scheduler.handle(nodeEvent2);
- scheduler.handle(nodeEvent2);
- assertEquals(3 * 1024, queueB1.getResourceUsage().getMemorySize());
- assertEquals(3 * 1024, queueB2.getResourceUsage().getMemorySize());
-
- // Both queue B1 and queue B2 usages go to 3 * 1024
-// TODO: assertFalse(queueB1.isStarvedForFairShare());
-// TODO: assertFalse(queueB2.isStarvedForFairShare());
- }
-
- @Test (timeout = 5000)
- public void testIsStarvedForFairShareDRF() throws Exception {
- conf.set(FairSchedulerConfiguration.ALLOCATION_FILE, ALLOC_FILE);
- PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
- out.println("<?xml version=\"1.0\"?>");
- out.println("<allocations>");
- out.println("<queue name=\"queueA\">");
- out.println("<weight>.5</weight>");
- out.println("</queue>");
- out.println("<queue name=\"queueB\">");
- out.println("<weight>.5</weight>");
- out.println("</queue>");
- out.println("<defaultFairSharePreemptionThreshold>1</defaultFairSharePreemptionThreshold>");
- out.println("<defaultQueueSchedulingPolicy>drf</defaultQueueSchedulingPolicy>");
- out.println("</allocations>");
- out.close();
-
- resourceManager = new MockRM(conf);
- resourceManager.start();
- scheduler = (FairScheduler) resourceManager.getResourceScheduler();
-
- // Add one big node (only care about aggregate capacity)
- RMNode node1 =
- MockNodes.newNodeInfo(1, Resources.createResource(10 * 1024, 10), 1,
- "127.0.0.1");
- NodeAddedSchedulerEvent nodeEvent1 = new NodeAddedSchedulerEvent(node1);
- scheduler.handle(nodeEvent1);
-
- scheduler.update();
-
- // Queue A wants 7 * 1024, 1. Node update gives this all to A
- createSchedulingRequest(7 * 1024, 1, "queueA", "user1", 1);
- scheduler.update();
- NodeUpdateSchedulerEvent nodeEvent2 = new NodeUpdateSchedulerEvent(node1);
- scheduler.handle(nodeEvent2);
-
- QueueManager queueMgr = scheduler.getQueueManager();
- FSLeafQueue queueA = queueMgr.getLeafQueue("queueA", false);
- assertEquals(7 * 1024, queueA.getResourceUsage().getMemorySize());
- assertEquals(1, queueA.getResourceUsage().getVirtualCores());
-
- // Queue B has 3 reqs :
- // 1) 2 * 1024, 5 .. which will be granted
- // 2) 1 * 1024, 1 .. which will be granted
- // 3) 1 * 1024, 1 .. which wont
- createSchedulingRequest(2 * 1024, 5, "queueB", "user1", 1);
- createSchedulingRequest(1 * 1024, 2, "queueB", "user1", 2);
- scheduler.update();
- for (int i = 0; i < 3; i ++) {
- scheduler.handle(nodeEvent2);
- }
-
- FSLeafQueue queueB = queueMgr.getLeafQueue("queueB", false);
- assertEquals(3 * 1024, queueB.getResourceUsage().getMemorySize());
- assertEquals(6, queueB.getResourceUsage().getVirtualCores());
-
- scheduler.update();
-
- // Verify that Queue us not starved for fair share..
- // Since the Starvation logic now uses DRF when the policy = drf, The
- // Queue should not be starved
-// TODO: assertFalse(queueB.isStarvedForFairShare());
}
@Test
http://git-wip-us.apache.org/repos/asf/hadoop/blob/59bc9d16/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestQueueManagerRealScheduler.java
----------------------------------------------------------------------
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/TestQueueManagerRealScheduler.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/TestQueueManagerRealScheduler.java
new file mode 100644
index 0000000..5736f75
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestQueueManagerRealScheduler.java
@@ -0,0 +1,128 @@
+/**
+ * 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.hadoop.yarn.server.resourcemanager.scheduler.fair;
+
+import org.apache.hadoop.yarn.server.resourcemanager.MockRM;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.io.PrintWriter;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * QueueManager tests that require a real scheduler
+ */
+public class TestQueueManagerRealScheduler extends FairSchedulerTestBase {
+ private final static File ALLOC_FILE = new File(TEST_DIR, "test-queue-mgr");
+
+ @Before
+ public void setup() throws IOException {
+ createConfiguration();
+ writeAllocFile(30, 40);
+ conf.set(FairSchedulerConfiguration.ALLOCATION_FILE,
+ ALLOC_FILE.getAbsolutePath());
+
+ resourceManager = new MockRM(conf);
+ resourceManager.start();
+ scheduler = (FairScheduler) resourceManager.getResourceScheduler();
+ }
+
+ @After
+ public void teardown() {
+ ALLOC_FILE.deleteOnExit();
+ if (resourceManager != null) {
+ resourceManager.stop();
+ resourceManager = null;
+ }
+ }
+
+ private void writeAllocFile(int defaultFairShareTimeout,
+ int fairShareTimeout) throws IOException {
+ PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
+ out.println("<?xml version=\"1.0\"?>");
+ out.println("<allocations>");
+ out.println("<queue name=\"default\">");
+ out.println("</queue>");
+ out.println("<queue name=\"queueA\">");
+ out.println("</queue>");
+ out.println("<queue name=\"queueB\">");
+ out.println("<queue name=\"queueB1\">");
+ out.println("<minSharePreemptionTimeout>5</minSharePreemptionTimeout>");
+ out.println("</queue>");
+ out.println("<queue name=\"queueB2\">");
+ out.println("</queue>");
+ out.println("</queue>");
+ out.println("<queue name=\"queueC\">");
+ out.println("</queue>");
+ out.println("<defaultMinSharePreemptionTimeout>15"
+ + "</defaultMinSharePreemptionTimeout>");
+ out.println("<defaultFairSharePreemptionTimeout>" +
+ + defaultFairShareTimeout + "</defaultFairSharePreemptionTimeout>");
+ out.println("<fairSharePreemptionTimeout>"
+ + fairShareTimeout + "</fairSharePreemptionTimeout>");
+ out.println("</allocations>");
+ out.close();
+ }
+
+ @Test
+ public void testBackwardsCompatiblePreemptionConfiguration()
+ throws IOException {
+ // Check the min/fair share preemption timeout for each queue
+ QueueManager queueMgr = scheduler.getQueueManager();
+ assertEquals(30000, queueMgr.getQueue("root")
+ .getFairSharePreemptionTimeout());
+ assertEquals(30000, queueMgr.getQueue("default")
+ .getFairSharePreemptionTimeout());
+ assertEquals(30000, queueMgr.getQueue("queueA")
+ .getFairSharePreemptionTimeout());
+ assertEquals(30000, queueMgr.getQueue("queueB")
+ .getFairSharePreemptionTimeout());
+ assertEquals(30000, queueMgr.getQueue("queueB.queueB1")
+ .getFairSharePreemptionTimeout());
+ assertEquals(30000, queueMgr.getQueue("queueB.queueB2")
+ .getFairSharePreemptionTimeout());
+ assertEquals(30000, queueMgr.getQueue("queueC")
+ .getFairSharePreemptionTimeout());
+ assertEquals(15000, queueMgr.getQueue("root")
+ .getMinSharePreemptionTimeout());
+ assertEquals(15000, queueMgr.getQueue("default")
+ .getMinSharePreemptionTimeout());
+ assertEquals(15000, queueMgr.getQueue("queueA")
+ .getMinSharePreemptionTimeout());
+ assertEquals(15000, queueMgr.getQueue("queueB")
+ .getMinSharePreemptionTimeout());
+ assertEquals(5000, queueMgr.getQueue("queueB.queueB1")
+ .getMinSharePreemptionTimeout());
+ assertEquals(15000, queueMgr.getQueue("queueB.queueB2")
+ .getMinSharePreemptionTimeout());
+ assertEquals(15000, queueMgr.getQueue("queueC")
+ .getMinSharePreemptionTimeout());
+
+ // Lower the fairshare preemption timeouts and verify it is picked
+ // correctly.
+ writeAllocFile(25, 30);
+ scheduler.reinitialize(conf, resourceManager.getRMContext());
+ assertEquals(25000, queueMgr.getQueue("root")
+ .getFairSharePreemptionTimeout());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org