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/11 00:16:55 UTC
[1/2] hadoop git commit: YARN-5821. Drop left-over preemption-related
code and clean up method visibilities in the Schedulable hierarchy
(Contributed by Karthik Kambatla via Daniel Templeton) [Forced Update!]
Repository: hadoop
Updated Branches:
refs/heads/YARN-4752 cda677655 -> 214067405 (forced update)
YARN-5821. Drop left-over preemption-related code and clean up method visibilities in the Schedulable hierarchy
(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/31518d06
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/31518d06
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/31518d06
Branch: refs/heads/YARN-4752
Commit: 31518d065bd2fc60f72907dda5b697d900666fc2
Parents: 07b9bf3
Author: Daniel Templeton <te...@apache.org>
Authored: Thu Nov 3 14:50:09 2016 -0700
Committer: Karthik Kambatla <ka...@cloudera.com>
Committed: Thu Nov 10 16:15:54 2016 -0800
----------------------------------------------------------------------
.../scheduler/fair/FSAppAttempt.java | 84 ++++-----------
.../scheduler/fair/FSLeafQueue.java | 102 +++----------------
.../scheduler/fair/FSParentQueue.java | 46 ++-------
.../resourcemanager/scheduler/fair/FSQueue.java | 20 ++--
.../scheduler/fair/Schedulable.java | 29 +++---
.../scheduler/fair/FakeSchedulable.java | 5 -
.../scheduler/fair/TestSchedulingPolicy.java | 5 -
7 files changed, 60 insertions(+), 231 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/31518d06/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 d9fdaba..06253d1 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
@@ -18,12 +18,10 @@
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair;
-import java.io.Serializable;
import java.text.DecimalFormat;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collection;
-import java.util.Comparator;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -81,7 +79,6 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
private FairScheduler scheduler;
private FSQueue fsQueue;
private Resource fairShare = Resources.createResource(0, 0);
- private RMContainerComparator comparator = new RMContainerComparator();
// Preemption related variables
private Resource fairshareStarvation = Resources.none();
@@ -121,7 +118,7 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
this.resourceWeights = new ResourceWeights();
}
- public ResourceWeights getResourceWeights() {
+ ResourceWeights getResourceWeights() {
return resourceWeights;
}
@@ -132,7 +129,7 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
return queue.getMetrics();
}
- public void containerCompleted(RMContainer rmContainer,
+ void containerCompleted(RMContainer rmContainer,
ContainerStatus containerStatus, RMContainerEventType event) {
try {
writeLock.lock();
@@ -491,7 +488,7 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
* @param schedulerKey Scheduler Key
* @param level NodeType
*/
- public void resetAllowedLocalityLevel(
+ void resetAllowedLocalityLevel(
SchedulerRequestKey schedulerKey, NodeType level) {
NodeType old;
try {
@@ -513,45 +510,33 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
}
// Preemption related methods
- public Resource getStarvation() {
+ Resource getStarvation() {
return Resources.add(fairshareStarvation, minshareStarvation);
}
- public void setMinshareStarvation(Resource starvation) {
+ void setMinshareStarvation(Resource starvation) {
this.minshareStarvation = starvation;
}
- public void resetMinshareStarvation() {
+ void resetMinshareStarvation() {
this.minshareStarvation = Resources.none();
}
- public void addPreemption(RMContainer container) {
+ void addPreemption(RMContainer container) {
containersToPreempt.add(container);
Resources.addTo(preemptedResources, container.getAllocatedResource());
}
- public Set<RMContainer> getPreemptionContainers() {
+ Set<RMContainer> getPreemptionContainers() {
return containersToPreempt;
}
- public Resource getPreemptedResources() {
+ private Resource getPreemptedResources() {
return preemptedResources;
}
- public void resetPreemptedResources() {
- preemptedResources = Resources.createResource(0);
- for (RMContainer container : getPreemptionContainers()) {
- Resources.addTo(preemptedResources, container.getAllocatedResource());
- }
- }
-
- public void clearPreemptedResources() {
- preemptedResources.setMemorySize(0);
- preemptedResources.setVirtualCores(0);
- }
-
- public boolean canContainerBePreempted(RMContainer container) {
+ boolean canContainerBePreempted(RMContainer container) {
// Sanity check that the app owns this container
if (!getLiveContainersMap().containsKey(container.getContainerId()) &&
!newlyAllocatedContainers.contains(container)) {
@@ -585,7 +570,7 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
* @param schedulerKey Scheduler Key
* @return Container
*/
- public Container createContainer(FSSchedulerNode node, Resource capability,
+ private Container createContainer(FSSchedulerNode node, Resource capability,
SchedulerRequestKey schedulerKey) {
NodeId nodeId = node.getRMNode().getNodeID();
@@ -593,12 +578,10 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
getApplicationAttemptId(), getNewContainerId());
// Create the container
- Container container = BuilderUtils.newContainer(containerId, nodeId,
+ return BuilderUtils.newContainer(containerId, nodeId,
node.getRMNode().getHttpAddress(), capability,
schedulerKey.getPriority(), null,
schedulerKey.getAllocationRequestId());
-
- return container;
}
/**
@@ -853,7 +836,8 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
}
Collection<SchedulerRequestKey> keysToTry = (reserved) ?
- Arrays.asList(node.getReservedContainer().getReservedSchedulerKey()) :
+ Collections.singletonList(
+ node.getReservedContainer().getReservedSchedulerKey()) :
getSchedulerKeys();
// For each priority, see if we can schedule a node local, rack local
@@ -1011,7 +995,7 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
* Node that the application has an existing reservation on
* @return whether the reservation on the given node is valid.
*/
- public boolean assignReservedContainer(FSSchedulerNode node) {
+ boolean assignReservedContainer(FSSchedulerNode node) {
RMContainer rmContainer = node.getReservedContainer();
SchedulerRequestKey reservedSchedulerKey =
rmContainer.getReservedSchedulerKey();
@@ -1040,19 +1024,6 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
return true;
}
- static class RMContainerComparator implements Comparator<RMContainer>,
- Serializable {
- @Override
- public int compare(RMContainer c1, RMContainer c2) {
- int ret = c1.getContainer().getPriority().compareTo(
- c2.getContainer().getPriority());
- if (ret == 0) {
- return c2.getContainerId().compareTo(c1.getContainerId());
- }
- return ret;
- }
- }
-
/**
* Helper method that computes the extent of fairshare fairshareStarvation.
*/
@@ -1079,7 +1050,7 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
return this.fairshareStarvation;
}
- public ResourceRequest getNextResourceRequest() {
+ ResourceRequest getNextResourceRequest() {
return appSchedulingInfo.getNextResourceRequest();
}
@@ -1197,25 +1168,4 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
updateAMContainerDiagnostics(AMState.INACTIVATED,
diagnosticMessageBldr.toString());
}
-
- /**
- * Preempt a running container according to the priority
- */
- @Override
- public RMContainer preemptContainer() {
- if (LOG.isDebugEnabled()) {
- LOG.debug("App " + getName() + " is going to preempt a running " +
- "container");
- }
-
- RMContainer toBePreempted = null;
- for (RMContainer container : getLiveContainers()) {
- if (!getPreemptionContainers().contains(container) &&
- (toBePreempted == null ||
- comparator.compare(toBePreempted, container) > 0)) {
- toBePreempted = container;
- }
- }
- return toBePreempted;
- }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/31518d06/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 90a381e..3fcf627 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
@@ -21,7 +21,6 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
-import java.util.Comparator;
import java.util.List;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
@@ -72,7 +71,7 @@ public class FSLeafQueue extends FSQueue {
private Resource amResourceUsage;
private final ActiveUsersManager activeUsersManager;
- public static final List<FSQueue> EMPTY_LIST = Collections.emptyList();
+ private static final List<FSQueue> EMPTY_LIST = Collections.emptyList();
public FSLeafQueue(String name, FairScheduler scheduler,
FSParentQueue parent) {
@@ -84,7 +83,7 @@ public class FSLeafQueue extends FSQueue {
amResourceUsage = Resource.newInstance(0, 0);
}
- public void addApp(FSAppAttempt app, boolean runnable) {
+ void addApp(FSAppAttempt app, boolean runnable) {
writeLock.lock();
try {
if (runnable) {
@@ -111,7 +110,7 @@ public class FSLeafQueue extends FSQueue {
* Removes the given app from this queue.
* @return whether or not the app was runnable
*/
- public boolean removeApp(FSAppAttempt app) {
+ boolean removeApp(FSAppAttempt app) {
boolean runnable = false;
// Remove app from runnable/nonRunnable list while holding the write lock
@@ -142,7 +141,7 @@ public class FSLeafQueue extends FSQueue {
* Removes the given app if it is non-runnable and belongs to this queue
* @return true if the app is removed, false otherwise
*/
- public boolean removeNonRunnableApp(FSAppAttempt app) {
+ boolean removeNonRunnableApp(FSAppAttempt app) {
writeLock.lock();
try {
return nonRunnableApps.remove(app);
@@ -151,7 +150,7 @@ public class FSLeafQueue extends FSQueue {
}
}
- public boolean isRunnableApp(FSAppAttempt attempt) {
+ boolean isRunnableApp(FSAppAttempt attempt) {
readLock.lock();
try {
return runnableApps.contains(attempt);
@@ -160,7 +159,7 @@ public class FSLeafQueue extends FSQueue {
}
}
- public boolean isNonRunnableApp(FSAppAttempt attempt) {
+ boolean isNonRunnableApp(FSAppAttempt attempt) {
readLock.lock();
try {
return nonRunnableApps.contains(attempt);
@@ -169,30 +168,8 @@ public class FSLeafQueue extends FSQueue {
}
}
- public void resetPreemptedResources() {
- readLock.lock();
- try {
- for (FSAppAttempt attempt : runnableApps) {
- attempt.resetPreemptedResources();
- }
- } finally {
- readLock.unlock();
- }
- }
-
- public void clearPreemptedResources() {
- readLock.lock();
- try {
- for (FSAppAttempt attempt : runnableApps) {
- attempt.clearPreemptedResources();
- }
- } finally {
- readLock.unlock();
- }
- }
-
- public List<FSAppAttempt> getCopyOfNonRunnableAppSchedulables() {
- List<FSAppAttempt> appsToReturn = new ArrayList<FSAppAttempt>();
+ List<FSAppAttempt> getCopyOfNonRunnableAppSchedulables() {
+ List<FSAppAttempt> appsToReturn = new ArrayList<>();
readLock.lock();
try {
appsToReturn.addAll(nonRunnableApps);
@@ -320,7 +297,7 @@ public class FSLeafQueue extends FSQueue {
return usage;
}
- public Resource getAmResourceUsage() {
+ Resource getAmResourceUsage() {
return amResourceUsage;
}
@@ -407,42 +384,6 @@ public class FSLeafQueue extends FSQueue {
}
@Override
- public RMContainer preemptContainer() {
- RMContainer toBePreempted = null;
-
- // If this queue is not over its fair share, reject
- if (!preemptContainerPreCheck()) {
- return toBePreempted;
- }
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Queue " + getName() + " is going to preempt a container " +
- "from its applications.");
- }
-
- // Choose the app that is most over fair share
- Comparator<Schedulable> comparator = policy.getComparator();
- FSAppAttempt candidateSched = null;
- readLock.lock();
- try {
- for (FSAppAttempt sched : runnableApps) {
- if (candidateSched == null ||
- comparator.compare(sched, candidateSched) > 0) {
- candidateSched = sched;
- }
- }
- } finally {
- readLock.unlock();
- }
-
- // Preempt from the selected app
- if (candidateSched != null) {
- toBePreempted = candidateSched.preemptContainer();
- }
- return toBePreempted;
- }
-
- @Override
public List<FSQueue> getChildQueues() {
return EMPTY_LIST;
}
@@ -451,7 +392,7 @@ public class FSLeafQueue extends FSQueue {
public List<QueueUserACLInfo> getQueueUserAclInfo(UserGroupInformation user) {
QueueUserACLInfo userAclInfo =
recordFactory.newRecordInstance(QueueUserACLInfo.class);
- List<QueueACL> operations = new ArrayList<QueueACL>();
+ List<QueueACL> operations = new ArrayList<>();
for (QueueACL operation : QueueACL.values()) {
if (hasAccess(operation, user)) {
operations.add(operation);
@@ -463,10 +404,6 @@ public class FSLeafQueue extends FSQueue {
return Collections.singletonList(userAclInfo);
}
- public long getLastTimeAtMinShare() {
- return lastTimeAtMinShare;
- }
-
private void setLastTimeAtMinShare(long lastTimeAtMinShare) {
this.lastTimeAtMinShare = lastTimeAtMinShare;
}
@@ -481,7 +418,7 @@ public class FSLeafQueue extends FSQueue {
}
}
- public int getNumNonRunnableApps() {
+ int getNumNonRunnableApps() {
readLock.lock();
try {
return nonRunnableApps.size();
@@ -533,10 +470,11 @@ public class FSLeafQueue extends FSQueue {
/**
* Check whether this queue can run this application master under the
* maxAMShare limit.
- * @param amResource
+ *
+ * @param amResource resources required to run the AM
* @return true if this queue can run
*/
- public boolean canRunAppAM(Resource amResource) {
+ boolean canRunAppAM(Resource amResource) {
if (Math.abs(maxAMShare - -1.0f) < 0.0001) {
return true;
}
@@ -561,7 +499,7 @@ public class FSLeafQueue extends FSQueue {
return Resources.fitsIn(ifRunAMResource, maxAMResource);
}
- public void addAMResourceUsage(Resource amResource) {
+ void addAMResourceUsage(Resource amResource) {
if (amResource != null) {
Resources.addTo(amResourceUsage, amResource);
}
@@ -583,16 +521,6 @@ public class FSLeafQueue extends FSQueue {
}
/**
- * Helper method to check if the queue should preempt containers
- *
- * @return true if check passes (can preempt) or false otherwise
- */
- private boolean preemptContainerPreCheck() {
- return parent.getPolicy().checkIfUsageOverFairShare(getResourceUsage(),
- getFairShare());
- }
-
- /**
* Helper method to compute the amount of minshare starvation.
*
* @return the extent of minshare starvation
http://git-wip-us.apache.org/repos/asf/hadoop/blob/31518d06/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
----------------------------------------------------------------------
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 9eb5983..16570aa 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
@@ -21,7 +21,6 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
-import java.util.Comparator;
import java.util.List;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
@@ -61,7 +60,7 @@ public class FSParentQueue extends FSQueue {
super(name, scheduler, parent);
}
- public void addChildQueue(FSQueue child) {
+ void addChildQueue(FSQueue child) {
writeLock.lock();
try {
childQueues.add(child);
@@ -70,7 +69,7 @@ public class FSParentQueue extends FSQueue {
}
}
- public void removeChildQueue(FSQueue child) {
+ void removeChildQueue(FSQueue child) {
writeLock.lock();
try {
childQueues.remove(child);
@@ -93,7 +92,7 @@ public class FSParentQueue extends FSQueue {
}
}
- public void recomputeSteadyShares() {
+ void recomputeSteadyShares() {
readLock.lock();
try {
policy.computeSteadyShares(childQueues, getSteadyFairShare());
@@ -188,7 +187,7 @@ public class FSParentQueue extends FSQueue {
@Override
public List<QueueUserACLInfo> getQueueUserAclInfo(UserGroupInformation user) {
- List<QueueUserACLInfo> userAcls = new ArrayList<QueueUserACLInfo>();
+ List<QueueUserACLInfo> userAcls = new ArrayList<>();
// Add queue acls
userAcls.add(getUserAclInfo(user));
@@ -246,39 +245,6 @@ public class FSParentQueue extends FSQueue {
}
@Override
- public RMContainer preemptContainer() {
- RMContainer toBePreempted = null;
-
- // Find the childQueue which is most over fair share
- FSQueue candidateQueue = null;
- Comparator<Schedulable> comparator = policy.getComparator();
-
- readLock.lock();
- try {
- for (FSQueue queue : childQueues) {
- // Skip selection for non-preemptable queue
- if (!queue.isPreemptable()) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("skipping from queue=" + getName()
- + " because it's a non-preemptable queue");
- }
- } else if (candidateQueue == null ||
- comparator.compare(queue, candidateQueue) > 0) {
- candidateQueue = queue;
- }
- }
- } finally {
- readLock.unlock();
- }
-
- // Let the selected queue choose which of its container to preempt
- if (candidateQueue != null) {
- toBePreempted = candidateQueue.preemptContainer();
- }
- return toBePreempted;
- }
-
- @Override
public List<FSQueue> getChildQueues() {
readLock.lock();
try {
@@ -301,7 +267,7 @@ public class FSParentQueue extends FSQueue {
super.policy = policy;
}
- public void incrementRunnableApps() {
+ void incrementRunnableApps() {
writeLock.lock();
try {
runnableApps++;
@@ -310,7 +276,7 @@ public class FSParentQueue extends FSQueue {
}
}
- public void decrementRunnableApps() {
+ void decrementRunnableApps() {
writeLock.lock();
try {
runnableApps--;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/31518d06/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
----------------------------------------------------------------------
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 e1e528e..572b5f9 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
@@ -253,7 +253,7 @@ public abstract class FSQueue implements Queue, Schedulable {
return steadyFairShare;
}
- public void setSteadyFairShare(Resource steadyFairShare) {
+ void setSteadyFairShare(Resource steadyFairShare) {
this.steadyFairShare = steadyFairShare;
metrics.setSteadyFairShare(steadyFairShare);
}
@@ -262,27 +262,27 @@ public abstract class FSQueue implements Queue, Schedulable {
return scheduler.getAllocationConfiguration().hasAccess(name, acl, user);
}
- public long getFairSharePreemptionTimeout() {
+ long getFairSharePreemptionTimeout() {
return fairSharePreemptionTimeout;
}
- public void setFairSharePreemptionTimeout(long fairSharePreemptionTimeout) {
+ void setFairSharePreemptionTimeout(long fairSharePreemptionTimeout) {
this.fairSharePreemptionTimeout = fairSharePreemptionTimeout;
}
- public long getMinSharePreemptionTimeout() {
+ long getMinSharePreemptionTimeout() {
return minSharePreemptionTimeout;
}
- public void setMinSharePreemptionTimeout(long minSharePreemptionTimeout) {
+ void setMinSharePreemptionTimeout(long minSharePreemptionTimeout) {
this.minSharePreemptionTimeout = minSharePreemptionTimeout;
}
- public float getFairSharePreemptionThreshold() {
+ float getFairSharePreemptionThreshold() {
return fairSharePreemptionThreshold;
}
- public void setFairSharePreemptionThreshold(float fairSharePreemptionThreshold) {
+ void setFairSharePreemptionThreshold(float fairSharePreemptionThreshold) {
this.fairSharePreemptionThreshold = fairSharePreemptionThreshold;
}
@@ -297,7 +297,7 @@ public abstract class FSQueue implements Queue, Schedulable {
* @param checkStarvation whether to check for fairshare or minshare
* starvation on update
*/
- public abstract void updateInternal(boolean checkStarvation);
+ abstract void updateInternal(boolean checkStarvation);
public void update(Resource fairShare, boolean checkStarvation) {
setFairShare(fairShare);
@@ -355,7 +355,7 @@ public abstract class FSQueue implements Queue, Schedulable {
*
* @return true if check passes (can assign) or false otherwise
*/
- protected boolean assignContainerPreCheck(FSSchedulerNode node) {
+ boolean assignContainerPreCheck(FSSchedulerNode node) {
if (!Resources.fitsIn(getResourceUsage(), maxShare)
|| node.getReservedContainer() != null) {
return false;
@@ -411,7 +411,7 @@ public abstract class FSQueue implements Queue, Schedulable {
return null;
}
- public boolean fitsInMaxShare(Resource additionalResource) {
+ boolean fitsInMaxShare(Resource additionalResource) {
Resource usagePlusAddition =
Resources.add(getResourceUsage(), additionalResource);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/31518d06/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
----------------------------------------------------------------------
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 289887f..cf78405 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
@@ -55,50 +55,45 @@ public interface Schedulable {
* Name of job/queue, used for debugging as well as for breaking ties in
* scheduling order deterministically.
*/
- public String getName();
+ String getName();
/**
* Maximum number of resources required by this Schedulable. This is defined as
* number of currently utilized resources + number of unlaunched resources (that
* are either not yet launched or need to be speculated).
*/
- public Resource getDemand();
+ Resource getDemand();
/** Get the aggregate amount of resources consumed by the schedulable. */
- public Resource getResourceUsage();
+ Resource getResourceUsage();
/** Minimum Resource share assigned to the schedulable. */
- public Resource getMinShare();
+ Resource getMinShare();
/** Maximum Resource share assigned to the schedulable. */
- public Resource getMaxShare();
+ Resource getMaxShare();
/** Job/queue weight in fair sharing. */
- public ResourceWeights getWeights();
+ ResourceWeights getWeights();
/** Start time for jobs in FIFO queues; meaningless for QueueSchedulables.*/
- public long getStartTime();
+ long getStartTime();
/** Job priority for jobs in FIFO queues; meaningless for QueueSchedulables. */
- public Priority getPriority();
+ Priority getPriority();
/** Refresh the Schedulable's demand and those of its children if any. */
- public void updateDemand();
+ void updateDemand();
/**
* Assign a container on this node if possible, and return the amount of
* resources assigned.
*/
- public Resource assignContainer(FSSchedulerNode node);
-
- /**
- * Preempt a container from this Schedulable if possible.
- */
- public RMContainer preemptContainer();
+ Resource assignContainer(FSSchedulerNode node);
/** Get the fair share assigned to this Schedulable. */
- public Resource getFairShare();
+ Resource getFairShare();
/** Assign a fair share to this Schedulable. */
- public void setFairShare(Resource fairShare);
+ void setFairShare(Resource fairShare);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/31518d06/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
----------------------------------------------------------------------
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 5a170cf..e802f42 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
@@ -86,11 +86,6 @@ public class FakeSchedulable implements Schedulable {
}
@Override
- public RMContainer preemptContainer() {
- return null;
- }
-
- @Override
public Resource getFairShare() {
return this.fairShare;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/31518d06/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
----------------------------------------------------------------------
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 dea2dd1..57c7301 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
@@ -330,11 +330,6 @@ public class TestSchedulingPolicy {
}
@Override
- public RMContainer preemptContainer() {
- throw new UnsupportedOperationException();
- }
-
- @Override
public Resource getFairShare() {
throw new UnsupportedOperationException();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[2/2] hadoop git commit: YARN-5783. Verify identification of starved
applications. (kasha)
Posted by ka...@apache.org.
YARN-5783. Verify identification of starved applications. (kasha)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/21406740
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/21406740
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/21406740
Branch: refs/heads/YARN-4752
Commit: 2140674052c5639f541ebcb34dd0019b984cfae1
Parents: 31518d0
Author: Karthik Kambatla <ka...@cloudera.com>
Authored: Tue Nov 8 18:09:23 2016 -0800
Committer: Karthik Kambatla <ka...@cloudera.com>
Committed: Thu Nov 10 16:15:55 2016 -0800
----------------------------------------------------------------------
.../scheduler/SchedulerApplicationAttempt.java | 16 ++
.../scheduler/common/fica/FiCaSchedulerApp.java | 16 ++
.../scheduler/fair/FSAppAttempt.java | 16 ++
.../scheduler/fair/FSPreemptionThread.java | 2 +-
.../scheduler/fair/FSStarvedApps.java | 56 +++--
.../scheduler/fair/FairScheduler.java | 7 +-
.../fair/FairSchedulerWithMockPreemption.java | 58 +++++
.../scheduler/fair/TestFSAppStarvation.java | 245 +++++++++++++++++++
8 files changed, 391 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/21406740/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.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/SchedulerApplicationAttempt.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java
index bb1d461..8b3f60c 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java
@@ -1253,6 +1253,22 @@ public class SchedulerApplicationAttempt implements SchedulableEntity {
unconfirmedAllocatedVcores.addAndGet(-res.getVirtualCores());
}
+ @Override
+ public int hashCode() {
+ return getApplicationAttemptId().hashCode();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (! (o instanceof SchedulerApplicationAttempt)) {
+ return false;
+ }
+
+ SchedulerApplicationAttempt other = (SchedulerApplicationAttempt) o;
+ return (this == other ||
+ this.getApplicationAttemptId().equals(other.getApplicationAttemptId()));
+ }
+
/**
* Different state for Application Master, user can see this state from web UI
*/
http://git-wip-us.apache.org/repos/asf/hadoop/blob/21406740/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/common/fica/FiCaSchedulerApp.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/common/fica/FiCaSchedulerApp.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/common/fica/FiCaSchedulerApp.java
index 6d9dda8..636d5b4 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/common/fica/FiCaSchedulerApp.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/common/fica/FiCaSchedulerApp.java
@@ -1100,4 +1100,20 @@ public class FiCaSchedulerApp extends SchedulerApplicationAttempt {
}
}
}
+
+ /*
+ * Overriding to appease findbugs
+ */
+ @Override
+ public int hashCode() {
+ return super.hashCode();
+ }
+
+ /*
+ * Overriding to appease findbugs
+ */
+ @Override
+ public boolean equals(Object o) {
+ return super.equals(o);
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/21406740/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 06253d1..6b88bd0 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
@@ -1168,4 +1168,20 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
updateAMContainerDiagnostics(AMState.INACTIVATED,
diagnosticMessageBldr.toString());
}
+
+ /*
+ * Overriding to appease findbugs
+ */
+ @Override
+ public int hashCode() {
+ return super.hashCode();
+ }
+
+ /*
+ * Overriding to appease findbugs
+ */
+ @Override
+ public boolean equals(Object o) {
+ return super.equals(o);
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/21406740/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 3732086..905b6f2 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
@@ -38,7 +38,7 @@ import java.util.TimerTask;
*/
public class FSPreemptionThread extends Thread {
private static final Log LOG = LogFactory.getLog(FSPreemptionThread.class);
- private final FSContext context;
+ protected final FSContext context;
private final FairScheduler scheduler;
private final long warnTimeBeforeKill;
private final Timer preemptionTimer;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/21406740/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSStarvedApps.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/FSStarvedApps.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/FSStarvedApps.java
index 670a12d..4f28e41 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/FSStarvedApps.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/FSStarvedApps.java
@@ -17,7 +17,6 @@
*/
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair;
-import com.google.common.annotations.VisibleForTesting;
import org.apache.hadoop.yarn.util.resource.Resources;
import java.io.Serializable;
@@ -25,28 +24,49 @@ import java.util.Comparator;
import java.util.concurrent.PriorityBlockingQueue;
/**
- * Helper class to track starved apps.
+ * Helper class to track starved applications.
*
* Initially, this uses a blocking queue. We could use other data structures
* in the future. This class also has some methods to simplify testing.
*/
-public class FSStarvedApps {
- private int numAppsAddedSoFar;
- private PriorityBlockingQueue<FSAppAttempt> apps;
+class FSStarvedApps {
- public FSStarvedApps() {
- apps = new PriorityBlockingQueue<>(10, new StarvationComparator());
+ // List of apps to be processed by the preemption thread.
+ private PriorityBlockingQueue<FSAppAttempt> appsToProcess;
+
+ // App being currently processed. This assumes a single reader.
+ private FSAppAttempt appBeingProcessed;
+
+ FSStarvedApps() {
+ appsToProcess = new PriorityBlockingQueue<>(10, new StarvationComparator());
}
- public void addStarvedApp(FSAppAttempt app) {
- if (!apps.contains(app)) {
- apps.add(app);
- numAppsAddedSoFar++;
+ /**
+ * Add a starved application if it is not already added.
+ * @param app application to add
+ */
+ void addStarvedApp(FSAppAttempt app) {
+ if (!app.equals(appBeingProcessed) && !appsToProcess.contains(app)) {
+ appsToProcess.add(app);
}
}
- public FSAppAttempt take() throws InterruptedException {
- return apps.take();
+ /**
+ * Blocking call to fetch the next app to process. The returned app is
+ * tracked until the next call to this method. This tracking assumes a
+ * single reader.
+ *
+ * @return starved application to process
+ * @throws InterruptedException if interrupted while waiting
+ */
+ FSAppAttempt take() throws InterruptedException {
+ // Reset appBeingProcessed before the blocking call
+ appBeingProcessed = null;
+
+ // Blocking call to fetch the next starved application
+ FSAppAttempt app = appsToProcess.take();
+ appBeingProcessed = app;
+ return app;
}
private static class StarvationComparator implements
@@ -62,14 +82,4 @@ public class FSStarvedApps {
return ret;
}
}
-
- @VisibleForTesting
- public int getNumAppsAddedSoFar() {
- return numAppsAddedSoFar;
- }
-
- @VisibleForTesting
- public int numStarvedApps() {
- return apps.size();
- }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/21406740/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 f408e24..abe8a6a 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
@@ -1244,7 +1244,7 @@ public class FairScheduler extends
}
if (this.conf.getPreemptionEnabled()) {
- preemptionThread = new FSPreemptionThread(this);
+ createPreemptionThread();
}
} finally {
writeLock.unlock();
@@ -1262,6 +1262,11 @@ public class FairScheduler extends
}
}
+ @VisibleForTesting
+ protected void createPreemptionThread() {
+ preemptionThread = new FSPreemptionThread(this);
+ }
+
private void updateReservationThreshold() {
Resource newThreshold = Resources.multiply(
getIncrementResourceCapability(),
http://git-wip-us.apache.org/repos/asf/hadoop/blob/21406740/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairSchedulerWithMockPreemption.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/FairSchedulerWithMockPreemption.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/FairSchedulerWithMockPreemption.java
new file mode 100644
index 0000000..25780cd
--- /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/FairSchedulerWithMockPreemption.java
@@ -0,0 +1,58 @@
+/**
+ * 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 java.util.HashSet;
+import java.util.Set;
+
+public class FairSchedulerWithMockPreemption extends FairScheduler {
+ @Override
+ protected void createPreemptionThread() {
+ preemptionThread = new MockPreemptionThread(this);
+ }
+
+ static class MockPreemptionThread extends FSPreemptionThread {
+ private Set<FSAppAttempt> appsAdded = new HashSet<>();
+ private int totalAppsAdded = 0;
+
+ MockPreemptionThread(FairScheduler scheduler) {
+ super(scheduler);
+ }
+
+ @Override
+ public void run() {
+ while (!Thread.interrupted()) {
+ try {
+ FSAppAttempt app = context.getStarvedApps().take();
+ appsAdded.add(app);
+ totalAppsAdded++;
+ } catch (InterruptedException e) {
+ return;
+ }
+ }
+ }
+
+ int uniqueAppsAdded() {
+ return appsAdded.size();
+ }
+
+ int totalAppsAdded() {
+ return totalAppsAdded;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/21406740/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
new file mode 100644
index 0000000..0e5511b
--- /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/TestFSAppStarvation.java
@@ -0,0 +1,245 @@
+/**
+ * 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.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.server.resourcemanager.MockNodes;
+import org.apache.hadoop.yarn.server.resourcemanager.MockRM;
+import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeAddedSchedulerEvent;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeUpdateSchedulerEvent;
+
+import org.apache.hadoop.yarn.util.resource.Resources;
+import org.junit.After;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+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 java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Test class to verify identification of app starvation
+ */
+public class TestFSAppStarvation extends FairSchedulerTestBase {
+
+ private static final File ALLOC_FILE = new File(TEST_DIR, "test-queues");
+ private final List<RMNode> rmNodes = new ArrayList<>();
+
+ // Node Capacity = NODE_CAPACITY_MULTIPLE * (1 GB or 1 vcore)
+ private static final int NODE_CAPACITY_MULTIPLE = 4;
+
+ private FairSchedulerWithMockPreemption.MockPreemptionThread preemptionThread;
+
+ @Before
+ public void setup() {
+ createConfiguration();
+ conf.set(YarnConfiguration.RM_SCHEDULER,
+ FairSchedulerWithMockPreemption.class.getCanonicalName());
+ conf.set(FairSchedulerConfiguration.ALLOCATION_FILE,
+ ALLOC_FILE.getAbsolutePath());
+ conf.setBoolean(FairSchedulerConfiguration.PREEMPTION, true);
+ conf.setFloat(FairSchedulerConfiguration.PREEMPTION_THRESHOLD, 0f);
+ }
+
+ @After
+ public void teardown() {
+ ALLOC_FILE.delete();
+ conf = null;
+ if (resourceManager != null) {
+ resourceManager.stop();
+ resourceManager = null;
+ }
+ }
+
+ /*
+ * Test to verify application starvation is computed only when preemption
+ * is enabled.
+ */
+ @Test
+ public void testPreemptionDisabled() throws Exception {
+ conf.setBoolean(FairSchedulerConfiguration.PREEMPTION, false);
+
+ setupClusterAndSubmitJobs();
+
+ assertNull("Found starved apps even when preemption is turned off",
+ scheduler.getContext().getStarvedApps());
+ }
+
+ /*
+ * Test to verify application starvation is computed correctly when
+ * preemption is turned on.
+ */
+ @Test
+ public void testPreemptionEnabled() throws Exception {
+ setupClusterAndSubmitJobs();
+
+ 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());
+
+ // Verify the apps get added again on a subsequent update
+ scheduler.update();
+ Thread.yield();
+ assertTrue("Each app is marked as starved exactly once",
+ preemptionThread.totalAppsAdded() > preemptionThread.uniqueAppsAdded());
+ }
+
+ /*
+ * Test to verify app starvation is computed only when the cluster
+ * utilization threshold is over the preemption threshold.
+ */
+ @Test
+ public void testClusterUtilizationThreshold() throws Exception {
+ // Set preemption threshold to 1.1, so the utilization is always lower
+ conf.setFloat(FairSchedulerConfiguration.PREEMPTION_THRESHOLD, 1.1f);
+
+ setupClusterAndSubmitJobs();
+
+ assertNotNull("FSContext does not have an FSStarvedApps instance",
+ scheduler.getContext().getStarvedApps());
+ assertEquals("Found starved apps when preemption threshold is over 100%", 0,
+ preemptionThread.totalAppsAdded());
+ }
+
+ private void setupClusterAndSubmitJobs() throws Exception {
+ setupStarvedCluster();
+ submitAppsToEachLeafQueue();
+ sendNodeUpdateEvents();
+
+ // Sleep to hit the preemption timeouts
+ Thread.sleep(10);
+
+ // Scheduler update to populate starved apps
+ scheduler.update();
+
+ // Wait for apps to be processed by MockPreemptionThread
+ Thread.yield();
+ }
+
+ /**
+ * Setup the cluster for starvation testing:
+ * 1. Create FS allocation file
+ * 2. Create and start MockRM
+ * 3. Add two nodes to the cluster
+ * 4. Submit an app that uses up all resources on the cluster
+ */
+ private void setupStarvedCluster() throws IOException {
+ PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
+ out.println("<?xml version=\"1.0\"?>");
+ out.println("<allocations>");
+
+ // Default queue
+ out.println("<queue name=\"default\">");
+ out.println("</queue>");
+
+ // Queue with preemption disabled
+ out.println("<queue name=\"no-preemption\">");
+ out.println("<fairSharePreemptionThreshold>0" +
+ "</fairSharePreemptionThreshold>");
+ out.println("</queue>");
+
+ // Queue with minshare preemption enabled
+ out.println("<queue name=\"minshare\">");
+ out.println("<fairSharePreemptionThreshold>0" +
+ "</fairSharePreemptionThreshold>");
+ out.println("<minSharePreemptionTimeout>0" +
+ "</minSharePreemptionTimeout>");
+ out.println("<minResources>2048mb,2vcores</minResources>");
+ out.println("</queue>");
+
+ // Queue with fairshare preemption enabled
+ out.println("<queue name=\"fairshare\">");
+ out.println("<fairSharePreemptionThreshold>1" +
+ "</fairSharePreemptionThreshold>");
+ out.println("<fairSharePreemptionTimeout>0" +
+ "</fairSharePreemptionTimeout>");
+
+ // 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>");
+
+ out.println("</queue>");
+
+ out.println("</allocations>");
+ out.close();
+
+ assertTrue("Allocation file does not exist, not running the test",
+ ALLOC_FILE.exists());
+
+ resourceManager = new MockRM(conf);
+ resourceManager.start();
+ scheduler = (FairScheduler) resourceManager.getResourceScheduler();
+ preemptionThread = (FairSchedulerWithMockPreemption.MockPreemptionThread)
+ scheduler.preemptionThread;
+
+ // Create and add two nodes to the cluster
+ addNode(NODE_CAPACITY_MULTIPLE * 1024, NODE_CAPACITY_MULTIPLE);
+ addNode(NODE_CAPACITY_MULTIPLE * 1024, NODE_CAPACITY_MULTIPLE);
+
+ // Create an app that takes up all the resources on the cluster
+ ApplicationAttemptId app
+ = createSchedulingRequest(1024, 1, "root.default", "default", 8);
+
+ scheduler.update();
+ sendNodeUpdateEvents();
+
+ assertEquals(8, scheduler.getSchedulerApp(app).getLiveContainers().size());
+ }
+
+ private void submitAppsToEachLeafQueue() {
+ String queues[] = {"no-preemption", "minshare", "fairshare.child"};
+ for (String queue : queues) {
+ createSchedulingRequest(1024, 1, "root." + queue, "user", 1);
+ }
+ scheduler.update();
+ }
+
+ private void addNode(int memory, int cores) {
+ int id = rmNodes.size() + 1;
+ RMNode node =
+ MockNodes.newNodeInfo(1, Resources.createResource(memory, cores), id,
+ "127.0.0." + id);
+ scheduler.handle(new NodeAddedSchedulerEvent(node));
+ rmNodes.add(node);
+ }
+
+ private void sendNodeUpdateEvents() {
+ for (RMNode node : rmNodes) {
+ NodeUpdateSchedulerEvent nodeUpdateSchedulerEvent =
+ new NodeUpdateSchedulerEvent(node);
+ for (int i = 0; i < NODE_CAPACITY_MULTIPLE; i++) {
+ scheduler.handle(nodeUpdateSchedulerEvent);
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org