YARN-482. FS: Extend SchedulingMode to intermediate queues. (kkambatl via tucu)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1469511 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Alejandro Abdelnur 2013-04-18 18:13:36 +00:00
parent 27bef76501
commit 15d5394221
18 changed files with 509 additions and 302 deletions

View File

@ -23,6 +23,9 @@ Release 2.0.5-beta - UNRELEASED
NEW FEATURES NEW FEATURES
YARN-482. FS: Extend SchedulingMode to intermediate queues.
(kkambatl via tucu)
IMPROVEMENTS IMPROVEMENTS
YARN-365. Change NM heartbeat handling to not generate a scheduler event YARN-365. Change NM heartbeat handling to not generate a scheduler event

View File

@ -278,9 +278,7 @@ private Resource assignContainer(FSSchedulerNode node,
} }
} }
private Resource assignContainer(FSSchedulerNode node, boolean reserved) {
@Override
public Resource assignContainer(FSSchedulerNode node, boolean reserved) {
LOG.info("Node offered to app: " + getName() + " reserved: " + reserved); LOG.info("Node offered to app: " + getName() + " reserved: " + reserved);
if (reserved) { if (reserved) {
@ -345,4 +343,13 @@ public Resource assignContainer(FSSchedulerNode node, boolean reserved) {
} }
return Resources.none(); return Resources.none();
} }
public Resource assignReservedContainer(FSSchedulerNode node) {
return assignContainer(node, true);
}
@Override
public Resource assignContainer(FSSchedulerNode node) {
return assignContainer(node, false);
}
} }

View File

@ -41,9 +41,6 @@ public class FSLeafQueue extends FSQueue {
private final List<AppSchedulable> appScheds = private final List<AppSchedulable> appScheds =
new ArrayList<AppSchedulable>(); new ArrayList<AppSchedulable>();
/** Scheduling mode for jobs inside the queue (fair or FIFO) */
private SchedulingMode schedulingMode;
private final FairScheduler scheduler; private final FairScheduler scheduler;
private final QueueManager queueMgr; private final QueueManager queueMgr;
private Resource demand = Resources.createResource(0); private Resource demand = Resources.createResource(0);
@ -86,13 +83,18 @@ public Collection<AppSchedulable> getAppSchedulables() {
return appScheds; return appScheds;
} }
public void setSchedulingMode(SchedulingMode mode) { @Override
this.schedulingMode = mode; public void setPolicy(SchedulingPolicy policy)
throws AllocationConfigurationException {
if (!SchedulingPolicy.isApplicableTo(policy, SchedulingPolicy.DEPTH_LEAF)) {
throwPolicyDoesnotApplyException(policy);
}
super.policy = policy;
} }
@Override @Override
public void recomputeFairShares() { public void recomputeShares() {
schedulingMode.computeShares(getAppSchedulables(), getFairShare()); policy.computeShares(getAppSchedulables(), getFairShare());
} }
@Override @Override
@ -136,42 +138,27 @@ public void updateDemand() {
} }
@Override @Override
public Resource assignContainer(FSSchedulerNode node, boolean reserved) { public Resource assignContainer(FSSchedulerNode node) {
LOG.debug("Node offered to queue: " + getName() + " reserved: " + reserved); Resource assigned = Resources.none();
// If this queue is over its limit, reject if (LOG.isDebugEnabled()) {
if (Resources.greaterThan(getResourceUsage(), LOG.debug("Node offered to queue: " + getName());
queueMgr.getMaxResources(getName()))) {
return Resources.none();
} }
// If this node already has reserved resources for an app, first try to if (!assignContainerPreCheck(node)) {
// finish allocating resources for that app. return assigned;
if (reserved) {
for (AppSchedulable sched : appScheds) {
if (sched.getApp().getApplicationAttemptId() ==
node.getReservedContainer().getApplicationAttemptId()) {
return sched.assignContainer(node, reserved);
}
}
return Resources.none(); // We should never get here
} }
// Otherwise, chose app to schedule based on given policy. Comparator<Schedulable> comparator = policy.getComparator();
else {
Comparator<Schedulable> comparator = schedulingMode.getComparator();
Collections.sort(appScheds, comparator); Collections.sort(appScheds, comparator);
for (AppSchedulable sched : appScheds) { for (AppSchedulable sched : appScheds) {
if (sched.getRunnable()) { if (sched.getRunnable()) {
Resource assignedResource = sched.assignContainer(node, reserved); assigned = sched.assignContainer(node);
if (!assignedResource.equals(Resources.none())) { if (Resources.greaterThan(assigned, Resources.none())) {
return assignedResource; break;
} }
} }
} }
return assigned;
return Resources.none();
}
} }
@Override @Override

View File

@ -20,6 +20,7 @@
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collection; import java.util.Collection;
import java.util.Collections;
import java.util.List; import java.util.List;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
@ -33,7 +34,6 @@ public class FSParentQueue extends FSQueue {
private static final Log LOG = LogFactory.getLog( private static final Log LOG = LogFactory.getLog(
FSParentQueue.class.getName()); FSParentQueue.class.getName());
private final List<FSQueue> childQueues = private final List<FSQueue> childQueues =
new ArrayList<FSQueue>(); new ArrayList<FSQueue>();
private final QueueManager queueMgr; private final QueueManager queueMgr;
@ -50,11 +50,11 @@ public void addChildQueue(FSQueue child) {
} }
@Override @Override
public void recomputeFairShares() { public void recomputeShares() {
SchedulingMode.getDefault().computeShares(childQueues, getFairShare()); policy.computeShares(childQueues, getFairShare());
for (FSQueue childQueue : childQueues) { for (FSQueue childQueue : childQueues) {
childQueue.getMetrics().setAvailableResourcesToQueue(childQueue.getFairShare()); childQueue.getMetrics().setAvailableResourcesToQueue(childQueue.getFairShare());
childQueue.recomputeFairShares(); childQueue.recomputeShares();
} }
} }
@ -131,13 +131,41 @@ public synchronized List<QueueUserACLInfo> getQueueUserAclInfo(
} }
@Override @Override
public Resource assignContainer(FSSchedulerNode node, boolean reserved) { public Resource assignContainer(FSSchedulerNode node) {
throw new IllegalStateException( Resource assigned = Resources.none();
"Parent queue should not be assigned container");
// If this queue is over its limit, reject
if (Resources.greaterThan(getResourceUsage(),
queueMgr.getMaxResources(getName()))) {
return assigned;
}
Collections.sort(childQueues, policy.getComparator());
for (FSQueue child : childQueues) {
assigned = child.assignContainer(node);
if (node.getReservedContainer() != null
|| Resources.greaterThan(assigned, Resources.none())) {
break;
}
}
return assigned;
} }
@Override @Override
public Collection<FSQueue> getChildQueues() { public Collection<FSQueue> getChildQueues() {
return childQueues; return childQueues;
} }
@Override
public void setPolicy(SchedulingPolicy policy)
throws AllocationConfigurationException {
boolean allowed =
SchedulingPolicy.isApplicableTo(policy, (this == queueMgr
.getRootQueue()) ? SchedulingPolicy.DEPTH_ROOT
: SchedulingPolicy.DEPTH_INTERMEDIATE);
if (!allowed) {
throwPolicyDoesnotApplyException(policy);
}
super.policy = policy;
}
} }

View File

@ -45,6 +45,8 @@ public abstract class FSQueue extends Schedulable implements Queue {
protected final RecordFactory recordFactory = protected final RecordFactory recordFactory =
RecordFactoryProvider.getRecordFactory(null); RecordFactoryProvider.getRecordFactory(null);
protected SchedulingPolicy policy = SchedulingPolicy.getDefault();
public FSQueue(String name, QueueManager queueMgr, public FSQueue(String name, QueueManager queueMgr,
FairScheduler scheduler, FSParentQueue parent) { FairScheduler scheduler, FSParentQueue parent) {
this.name = name; this.name = name;
@ -63,6 +65,19 @@ public String getQueueName() {
return name; return name;
} }
public SchedulingPolicy getPolicy() {
return policy;
}
protected void throwPolicyDoesnotApplyException(SchedulingPolicy policy)
throws AllocationConfigurationException {
throw new AllocationConfigurationException("SchedulingPolicy " + policy
+ " does not apply to queue " + getName());
}
public abstract void setPolicy(SchedulingPolicy policy)
throws AllocationConfigurationException;
@Override @Override
public double getWeight() { public double getWeight() {
return queueMgr.getQueueWeight(getName()); return queueMgr.getQueueWeight(getName());
@ -130,13 +145,27 @@ public boolean hasAccess(QueueACL acl, UserGroupInformation user) {
} }
/** /**
* Recomputes the fair shares for all queues and applications * Recomputes the shares for all child queues and applications based on this
* under this queue. * queue's current share
*/ */
public abstract void recomputeFairShares(); public abstract void recomputeShares();
/** /**
* Gets the children of this queue, if any. * Gets the children of this queue, if any.
*/ */
public abstract Collection<FSQueue> getChildQueues(); public abstract Collection<FSQueue> getChildQueues();
/**
* Helper method to check if the queue should attempt assigning resources
*
* @return true if check passes (can assign) or false otherwise
*/
protected boolean assignContainerPreCheck(FSSchedulerNode node) {
if (Resources.greaterThan(getResourceUsage(),
queueMgr.getMaxResources(getName()))
|| node.getReservedContainer() != null) {
return false;
}
return true;
}
} }

View File

@ -52,6 +52,7 @@ public class FSSchedulerNode extends SchedulerNode {
private volatile int numContainers; private volatile int numContainers;
private RMContainer reservedContainer; private RMContainer reservedContainer;
private AppSchedulable reservedAppSchedulable;
/* set of containers that are allocated containers */ /* set of containers that are allocated containers */
private final Map<ContainerId, RMContainer> launchedContainers = private final Map<ContainerId, RMContainer> launchedContainers =
@ -221,6 +222,7 @@ public synchronized void reserveResource(
" on node " + this + " for application " + application); " on node " + this + " for application " + application);
} }
this.reservedContainer = reservedContainer; this.reservedContainer = reservedContainer;
this.reservedAppSchedulable = application.getAppSchedulable();
} }
public synchronized void unreserveResource( public synchronized void unreserveResource(
@ -237,11 +239,15 @@ public synchronized void unreserveResource(
" on node " + this); " on node " + this);
} }
reservedContainer = null; this.reservedContainer = null;
this.reservedAppSchedulable = null;
} }
public synchronized RMContainer getReservedContainer() { public synchronized RMContainer getReservedContainer() {
return reservedContainer; return reservedContainer;
} }
public synchronized AppSchedulable getReservedAppSchedulable() {
return reservedAppSchedulable;
}
} }

View File

@ -217,7 +217,7 @@ protected synchronized void update() {
rootQueue.setFairShare(clusterCapacity); rootQueue.setFairShare(clusterCapacity);
// Recursively compute fair shares for all queues // Recursively compute fair shares for all queues
// and update metrics // and update metrics
rootQueue.recomputeFairShares(); rootQueue.recomputeShares();
// Update recorded capacity of root queue (child queues are updated // Update recorded capacity of root queue (child queues are updated
// when fair share is calculated). // when fair share is calculated).
@ -786,39 +786,24 @@ private synchronized void nodeUpdate(RMNode nm) {
// 1. Check for reserved applications // 1. Check for reserved applications
// 2. Schedule if there are no reservations // 2. Schedule if there are no reservations
// If we have have an application that has reserved a resource on this node AppSchedulable reservedAppSchedulable = node.getReservedAppSchedulable();
// already, we try to complete the reservation. if (reservedAppSchedulable != null) {
RMContainer reservedContainer = node.getReservedContainer(); // Reservation exists; try to fulfill the reservation
if (reservedContainer != null) { LOG.info("Trying to fulfill reservation for application "
FSSchedulerApp reservedApplication = + reservedAppSchedulable.getApp().getApplicationAttemptId()
applications.get(reservedContainer.getApplicationAttemptId()); + " on node: " + nm);
// Try to fulfill the reservation node.getReservedAppSchedulable().assignReservedContainer(node);
LOG.info("Trying to fulfill reservation for application " +
reservedApplication.getApplicationId() + " on node: " + nm);
FSLeafQueue queue = queueMgr.getLeafQueue(reservedApplication.getQueueName());
queue.assignContainer(node, true);
} }
// Otherwise, schedule at queue which is furthest below fair share
else { else {
// No reservation, schedule at queue which is farthest below fair share
int assignedContainers = 0; int assignedContainers = 0;
while (node.getReservedContainer() == null) { while (node.getReservedContainer() == null) {
// At most one task is scheduled each iteration of this loop
List<FSLeafQueue> scheds = new ArrayList<FSLeafQueue>(
queueMgr.getLeafQueues());
Collections.sort(scheds, SchedulingMode.getDefault().getComparator());
boolean assignedContainer = false; boolean assignedContainer = false;
for (FSLeafQueue sched : scheds) { if (Resources.greaterThan(
Resource assigned = sched.assignContainer(node, false); queueMgr.getRootQueue().assignContainer(node),
if (Resources.greaterThan(assigned, Resources.none()) || Resources.none())) {
node.getReservedContainer() != null) {
eventLog.log("ASSIGN", nm.getHostName(), assigned);
assignedContainers++;
assignedContainer = true; assignedContainer = true;
break;
}
} }
if (!assignedContainer) { break; } if (!assignedContainer) { break; }
if (!assignMultiple) { break; } if (!assignMultiple) { break; }

View File

@ -143,7 +143,6 @@ public FSLeafQueue getLeafQueue(String name) {
if (leafQueue == null) { if (leafQueue == null) {
return null; return null;
} }
leafQueue.setSchedulingMode(info.defaultSchedulingMode);
queue = leafQueue; queue = leafQueue;
} else if (queue instanceof FSParentQueue) { } else if (queue instanceof FSParentQueue) {
return null; return null;
@ -302,7 +301,7 @@ public void reloadAllocs() throws IOException, ParserConfigurationException,
Map<String, Integer> queueMaxApps = new HashMap<String, Integer>(); Map<String, Integer> queueMaxApps = new HashMap<String, Integer>();
Map<String, Integer> userMaxApps = new HashMap<String, Integer>(); Map<String, Integer> userMaxApps = new HashMap<String, Integer>();
Map<String, Double> queueWeights = new HashMap<String, Double>(); Map<String, Double> queueWeights = new HashMap<String, Double>();
Map<String, SchedulingMode> queueModes = new HashMap<String, SchedulingMode>(); Map<String, SchedulingPolicy> queuePolicies = new HashMap<String, SchedulingPolicy>();
Map<String, Long> minSharePreemptionTimeouts = new HashMap<String, Long>(); Map<String, Long> minSharePreemptionTimeouts = new HashMap<String, Long>();
Map<String, Map<QueueACL, AccessControlList>> queueAcls = Map<String, Map<QueueACL, AccessControlList>> queueAcls =
new HashMap<String, Map<QueueACL, AccessControlList>>(); new HashMap<String, Map<QueueACL, AccessControlList>>();
@ -310,7 +309,7 @@ public void reloadAllocs() throws IOException, ParserConfigurationException,
int queueMaxAppsDefault = Integer.MAX_VALUE; int queueMaxAppsDefault = Integer.MAX_VALUE;
long fairSharePreemptionTimeout = Long.MAX_VALUE; long fairSharePreemptionTimeout = Long.MAX_VALUE;
long defaultMinSharePreemptionTimeout = Long.MAX_VALUE; long defaultMinSharePreemptionTimeout = Long.MAX_VALUE;
SchedulingMode defaultSchedulingMode = SchedulingMode.getDefault(); SchedulingPolicy defaultSchedPolicy = SchedulingPolicy.getDefault();
// Remember all queue names so we can display them on web UI, etc. // Remember all queue names so we can display them on web UI, etc.
List<String> queueNamesInAllocFile = new ArrayList<String>(); List<String> queueNamesInAllocFile = new ArrayList<String>();
@ -339,7 +338,7 @@ public void reloadAllocs() throws IOException, ParserConfigurationException,
if ("queue".equals(element.getTagName()) || if ("queue".equals(element.getTagName()) ||
"pool".equals(element.getTagName())) { "pool".equals(element.getTagName())) {
loadQueue("root", element, minQueueResources, maxQueueResources, queueMaxApps, loadQueue("root", element, minQueueResources, maxQueueResources, queueMaxApps,
userMaxApps, queueWeights, queueModes, minSharePreemptionTimeouts, userMaxApps, queueWeights, queuePolicies, minSharePreemptionTimeouts,
queueAcls, queueNamesInAllocFile); queueAcls, queueNamesInAllocFile);
} else if ("user".equals(element.getTagName())) { } else if ("user".equals(element.getTagName())) {
String userName = element.getAttribute("name"); String userName = element.getAttribute("name");
@ -370,11 +369,12 @@ public void reloadAllocs() throws IOException, ParserConfigurationException,
} else if ("queueMaxAppsDefault".equals(element.getTagName())) { } else if ("queueMaxAppsDefault".equals(element.getTagName())) {
String text = ((Text)element.getFirstChild()).getData().trim(); String text = ((Text)element.getFirstChild()).getData().trim();
int val = Integer.parseInt(text); int val = Integer.parseInt(text);
queueMaxAppsDefault = val;} queueMaxAppsDefault = val;
else if ("defaultQueueSchedulingMode".equals(element.getTagName())) { } else if ("defaultQueueSchedulingPolicy".equals(element.getTagName())
|| "defaultQueueSchedulingMode".equals(element.getTagName())) {
String text = ((Text)element.getFirstChild()).getData().trim(); String text = ((Text)element.getFirstChild()).getData().trim();
SchedulingMode.setDefault(text); SchedulingPolicy.setDefault(text);
defaultSchedulingMode = SchedulingMode.getDefault(); defaultSchedPolicy = SchedulingPolicy.getDefault();
} else { } else {
LOG.warn("Bad element in allocations file: " + element.getTagName()); LOG.warn("Bad element in allocations file: " + element.getTagName());
} }
@ -385,7 +385,7 @@ else if ("defaultQueueSchedulingMode".equals(element.getTagName())) {
synchronized (this) { synchronized (this) {
info = new QueueManagerInfo(minQueueResources, maxQueueResources, info = new QueueManagerInfo(minQueueResources, maxQueueResources,
queueMaxApps, userMaxApps, queueWeights, userMaxAppsDefault, queueMaxApps, userMaxApps, queueWeights, userMaxAppsDefault,
queueMaxAppsDefault, defaultSchedulingMode, minSharePreemptionTimeouts, queueMaxAppsDefault, defaultSchedPolicy, minSharePreemptionTimeouts,
queueAcls, fairSharePreemptionTimeout, defaultMinSharePreemptionTimeout); queueAcls, fairSharePreemptionTimeout, defaultMinSharePreemptionTimeout);
// Root queue should have empty ACLs. As a queue's ACL is the union of // Root queue should have empty ACLs. As a queue's ACL is the union of
@ -397,13 +397,14 @@ else if ("defaultQueueSchedulingMode".equals(element.getTagName())) {
rootAcls.put(QueueACL.ADMINISTER_QUEUE, new AccessControlList(" ")); rootAcls.put(QueueACL.ADMINISTER_QUEUE, new AccessControlList(" "));
queueAcls.put(ROOT_QUEUE, rootAcls); queueAcls.put(ROOT_QUEUE, rootAcls);
// Create all queus
for (String name: queueNamesInAllocFile) { for (String name: queueNamesInAllocFile) {
FSLeafQueue queue = getLeafQueue(name); getLeafQueue(name);
if (queueModes.containsKey(name)) {
queue.setSchedulingMode(queueModes.get(name));
} else {
queue.setSchedulingMode(defaultSchedulingMode);
} }
// Set custom policies as specified
for (Map.Entry<String, SchedulingPolicy> entry : queuePolicies.entrySet()) {
queues.get(entry.getKey()).setPolicy(entry.getValue());
} }
} }
} }
@ -414,7 +415,8 @@ else if ("defaultQueueSchedulingMode".equals(element.getTagName())) {
private void loadQueue(String parentName, Element element, Map<String, Resource> minQueueResources, private void loadQueue(String parentName, Element element, Map<String, Resource> minQueueResources,
Map<String, Resource> maxQueueResources, Map<String, Integer> queueMaxApps, Map<String, Resource> maxQueueResources, Map<String, Integer> queueMaxApps,
Map<String, Integer> userMaxApps, Map<String, Double> queueWeights, Map<String, Integer> userMaxApps, Map<String, Double> queueWeights,
Map<String, SchedulingMode> queueModes, Map<String, Long> minSharePreemptionTimeouts, Map<String, SchedulingPolicy> queuePolicies,
Map<String, Long> minSharePreemptionTimeouts,
Map<String, Map<QueueACL, AccessControlList>> queueAcls, List<String> queueNamesInAllocFile) Map<String, Map<QueueACL, AccessControlList>> queueAcls, List<String> queueNamesInAllocFile)
throws AllocationConfigurationException { throws AllocationConfigurationException {
String queueName = parentName + "." + element.getAttribute("name"); String queueName = parentName + "." + element.getAttribute("name");
@ -448,9 +450,10 @@ private void loadQueue(String parentName, Element element, Map<String, Resource>
String text = ((Text)field.getFirstChild()).getData().trim(); String text = ((Text)field.getFirstChild()).getData().trim();
long val = Long.parseLong(text) * 1000L; long val = Long.parseLong(text) * 1000L;
minSharePreemptionTimeouts.put(queueName, val); minSharePreemptionTimeouts.put(queueName, val);
} else if ("schedulingMode".equals(field.getTagName())) { } else if ("schedulingPolicy".equals(field.getTagName())
|| "schedulingMode".equals(field.getTagName())) {
String text = ((Text)field.getFirstChild()).getData().trim(); String text = ((Text)field.getFirstChild()).getData().trim();
queueModes.put(queueName, SchedulingMode.parse(text)); queuePolicies.put(queueName, SchedulingPolicy.parse(text));
} else if ("aclSubmitApps".equals(field.getTagName())) { } else if ("aclSubmitApps".equals(field.getTagName())) {
String text = ((Text)field.getFirstChild()).getData().trim(); String text = ((Text)field.getFirstChild()).getData().trim();
acls.put(QueueACL.SUBMIT_APPLICATIONS, new AccessControlList(text)); acls.put(QueueACL.SUBMIT_APPLICATIONS, new AccessControlList(text));
@ -459,8 +462,9 @@ private void loadQueue(String parentName, Element element, Map<String, Resource>
acls.put(QueueACL.ADMINISTER_QUEUE, new AccessControlList(text)); acls.put(QueueACL.ADMINISTER_QUEUE, new AccessControlList(text));
} else if ("queue".endsWith(field.getTagName()) || } else if ("queue".endsWith(field.getTagName()) ||
"pool".equals(field.getTagName())) { "pool".equals(field.getTagName())) {
loadQueue(queueName, field, minQueueResources, maxQueueResources, queueMaxApps, loadQueue(queueName, field, minQueueResources, maxQueueResources,
userMaxApps, queueWeights, queueModes, minSharePreemptionTimeouts, queueMaxApps, userMaxApps, queueWeights, queuePolicies,
minSharePreemptionTimeouts,
queueAcls, queueNamesInAllocFile); queueAcls, queueNamesInAllocFile);
isLeaf = false; isLeaf = false;
} }
@ -615,13 +619,13 @@ static class QueueManagerInfo {
// below half its fair share for this long, it is allowed to preempt tasks. // below half its fair share for this long, it is allowed to preempt tasks.
public final long fairSharePreemptionTimeout; public final long fairSharePreemptionTimeout;
public final SchedulingMode defaultSchedulingMode; public final SchedulingPolicy defaultSchedulingPolicy;
public QueueManagerInfo(Map<String, Resource> minQueueResources, public QueueManagerInfo(Map<String, Resource> minQueueResources,
Map<String, Resource> maxQueueResources, Map<String, Resource> maxQueueResources,
Map<String, Integer> queueMaxApps, Map<String, Integer> userMaxApps, Map<String, Integer> queueMaxApps, Map<String, Integer> userMaxApps,
Map<String, Double> queueWeights, int userMaxAppsDefault, Map<String, Double> queueWeights, int userMaxAppsDefault,
int queueMaxAppsDefault, SchedulingMode defaultSchedulingMode, int queueMaxAppsDefault, SchedulingPolicy defaultSchedulingPolicy,
Map<String, Long> minSharePreemptionTimeouts, Map<String, Long> minSharePreemptionTimeouts,
Map<String, Map<QueueACL, AccessControlList>> queueAcls, Map<String, Map<QueueACL, AccessControlList>> queueAcls,
long fairSharePreemptionTimeout, long defaultMinSharePreemptionTimeout) { long fairSharePreemptionTimeout, long defaultMinSharePreemptionTimeout) {
@ -632,7 +636,7 @@ public QueueManagerInfo(Map<String, Resource> minQueueResources,
this.queueWeights = queueWeights; this.queueWeights = queueWeights;
this.userMaxAppsDefault = userMaxAppsDefault; this.userMaxAppsDefault = userMaxAppsDefault;
this.queueMaxAppsDefault = queueMaxAppsDefault; this.queueMaxAppsDefault = queueMaxAppsDefault;
this.defaultSchedulingMode = defaultSchedulingMode; this.defaultSchedulingPolicy = defaultSchedulingPolicy;
this.minSharePreemptionTimeouts = minSharePreemptionTimeouts; this.minSharePreemptionTimeouts = minSharePreemptionTimeouts;
this.queueAcls = queueAcls; this.queueAcls = queueAcls;
this.fairSharePreemptionTimeout = fairSharePreemptionTimeout; this.fairSharePreemptionTimeout = fairSharePreemptionTimeout;
@ -651,7 +655,7 @@ public QueueManagerInfo() {
minSharePreemptionTimeouts = new HashMap<String, Long>(); minSharePreemptionTimeouts = new HashMap<String, Long>();
defaultMinSharePreemptionTimeout = Long.MAX_VALUE; defaultMinSharePreemptionTimeout = Long.MAX_VALUE;
fairSharePreemptionTimeout = Long.MAX_VALUE; fairSharePreemptionTimeout = Long.MAX_VALUE;
defaultSchedulingMode = SchedulingMode.getDefault(); defaultSchedulingPolicy = SchedulingPolicy.getDefault();
} }
} }
} }

View File

@ -93,11 +93,9 @@ public abstract class Schedulable {
/** /**
* Assign a container on this node if possible, and return the amount of * Assign a container on this node if possible, and return the amount of
* resources assigned. If {@code reserved} is true, it means a reservation * resources assigned.
* already exists on this node, and the schedulable should fulfill that
* reservation if possible.
*/ */
public abstract Resource assignContainer(FSSchedulerNode node, boolean reserved); public abstract Resource assignContainer(FSSchedulerNode node);
/** Assign a fair share to this Schedulable. */ /** Assign a fair share to this Schedulable. */
public void setFairShare(Resource fairShare) { public void setFairShare(Resource fairShare) {

View File

@ -1,118 +0,0 @@
/**
* 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.Collection;
import java.util.Comparator;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.modes.FairSchedulingMode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.modes.FifoSchedulingMode;
@Public
@Unstable
public abstract class SchedulingMode {
private static final ConcurrentHashMap<Class<? extends SchedulingMode>, SchedulingMode> instances =
new ConcurrentHashMap<Class<? extends SchedulingMode>, SchedulingMode>();
private static SchedulingMode DEFAULT_MODE =
getInstance(FairSchedulingMode.class);
public static SchedulingMode getDefault() {
return DEFAULT_MODE;
}
public static void setDefault(String className)
throws AllocationConfigurationException {
DEFAULT_MODE = parse(className);
}
/**
* Returns a {@link SchedulingMode} instance corresponding to the passed clazz
*/
public static SchedulingMode getInstance(Class<? extends SchedulingMode> clazz) {
SchedulingMode mode = instances.get(clazz);
if (mode == null) {
mode = ReflectionUtils.newInstance(clazz, null);
instances.put(clazz, mode);
}
return mode;
}
/**
* Returns {@link SchedulingMode} instance corresponding to the
* {@link SchedulingMode} passed as a string. The mode can be "fair" for
* FairSchedulingMode of "fifo" for FifoSchedulingMode. For custom
* {@link SchedulingMode}s in the RM classpath, the mode should be canonical
* class name of the {@link SchedulingMode}.
*
* @param mode canonical class name or "fair" or "fifo"
* @throws AllocationConfigurationException
*/
@SuppressWarnings("unchecked")
public static SchedulingMode parse(String mode)
throws AllocationConfigurationException {
@SuppressWarnings("rawtypes")
Class clazz;
String text = mode.toLowerCase();
if (text.equals("fair")) {
clazz = FairSchedulingMode.class;
} else if (text.equals("fifo")) {
clazz = FifoSchedulingMode.class;
} else {
try {
clazz = Class.forName(mode);
} catch (ClassNotFoundException cnfe) {
throw new AllocationConfigurationException(mode
+ " SchedulingMode class not found!");
}
}
if (!SchedulingMode.class.isAssignableFrom(clazz)) {
throw new AllocationConfigurationException(mode
+ " does not extend SchedulingMode");
}
return getInstance(clazz);
}
/**
* @return returns the name of SchedulingMode
*/
public abstract String getName();
/**
* The comparator returned by this method is to be used for sorting the
* {@link Schedulable}s in that queue.
*
* @return the comparator to sort by
*/
public abstract Comparator<Schedulable> getComparator();
/**
* Computes and updates the shares of {@link Schedulable}s as per the
* SchedulingMode, to be used later at schedule time.
*
* @param schedulables {@link Schedulable}s whose shares are to be updated
* @param totalResources Total {@link Resource}s in the cluster
*/
public abstract void computeShares(
Collection<? extends Schedulable> schedulables, Resource totalResources);
}

View File

@ -0,0 +1,145 @@
/**
* 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.Collection;
import java.util.Comparator;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies.FairSharePolicy;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies.FifoPolicy;
@Public
@Unstable
public abstract class SchedulingPolicy {
private static final ConcurrentHashMap<Class<? extends SchedulingPolicy>, SchedulingPolicy> instances =
new ConcurrentHashMap<Class<? extends SchedulingPolicy>, SchedulingPolicy>();
private static SchedulingPolicy DEFAULT_POLICY =
getInstance(FairSharePolicy.class);
public static final byte DEPTH_LEAF = (byte) 1;
public static final byte DEPTH_INTERMEDIATE = (byte) 2;
public static final byte DEPTH_ROOT = (byte) 4;
public static final byte DEPTH_PARENT = (byte) 6; // Root and Intermediate
public static final byte DEPTH_ANY = (byte) 7;
public static SchedulingPolicy getDefault() {
return DEFAULT_POLICY;
}
public static void setDefault(String className)
throws AllocationConfigurationException {
DEFAULT_POLICY = parse(className);
}
/**
* Returns a {@link SchedulingPolicy} instance corresponding to the passed clazz
*/
public static SchedulingPolicy getInstance(Class<? extends SchedulingPolicy> clazz) {
SchedulingPolicy policy = instances.get(clazz);
if (policy == null) {
policy = ReflectionUtils.newInstance(clazz, null);
instances.put(clazz, policy);
}
return policy;
}
/**
* Returns {@link SchedulingPolicy} instance corresponding to the
* {@link SchedulingPolicy} passed as a string. The policy can be "fair" for
* FairsharePolicy or "fifo" for FifoPolicy. For custom
* {@link SchedulingPolicy}s in the RM classpath, the policy should be
* canonical class name of the {@link SchedulingPolicy}.
*
* @param policy canonical class name or "fair" or "fifo"
* @throws AllocationConfigurationException
*/
@SuppressWarnings("unchecked")
public static SchedulingPolicy parse(String policy)
throws AllocationConfigurationException {
@SuppressWarnings("rawtypes")
Class clazz;
String text = policy.toLowerCase();
if (text.equals("fair")) {
clazz = FairSharePolicy.class;
} else if (text.equals("fifo")) {
clazz = FifoPolicy.class;
} else {
try {
clazz = Class.forName(policy);
} catch (ClassNotFoundException cnfe) {
throw new AllocationConfigurationException(policy
+ " SchedulingPolicy class not found!");
}
}
if (!SchedulingPolicy.class.isAssignableFrom(clazz)) {
throw new AllocationConfigurationException(policy
+ " does not extend SchedulingPolicy");
}
return getInstance(clazz);
}
/**
* @return returns the name of {@link SchedulingPolicy}
*/
public abstract String getName();
/**
* Specifies the depths in the hierarchy, this {@link SchedulingPolicy}
* applies to
*
* @return depth equal to one of fields {@link SchedulingPolicy}#DEPTH_*
*/
public abstract byte getApplicableDepth();
/**
* Checks if the specified {@link SchedulingPolicy} can be used for a queue at
* the specified depth in the hierarchy
*
* @param policy {@link SchedulingPolicy} we are checking the
* depth-applicability for
* @param depth queue's depth in the hierarchy
* @return true if policy is applicable to passed depth, false otherwise
*/
public static boolean isApplicableTo(SchedulingPolicy policy, byte depth) {
return ((policy.getApplicableDepth() & depth) == depth) ? true : false;
}
/**
* The comparator returned by this method is to be used for sorting the
* {@link Schedulable}s in that queue.
*
* @return the comparator to sort by
*/
public abstract Comparator<Schedulable> getComparator();
/**
* Computes and updates the shares of {@link Schedulable}s as per the
* {@link SchedulingPolicy}, to be used later at schedule time.
*
* @param schedulables {@link Schedulable}s whose shares are to be updated
* @param totalResources Total {@link Resource}s in the cluster
*/
public abstract void computeShares(
Collection<? extends Schedulable> schedulables, Resource totalResources);
}

View File

@ -15,7 +15,7 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.modes; package org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies;
import java.io.Serializable; import java.io.Serializable;
import java.util.Collection; import java.util.Collection;
@ -24,13 +24,13 @@
import org.apache.hadoop.yarn.api.records.Resource; import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.Resources; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.Resources;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.Schedulable; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.Schedulable;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.SchedulingMode; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.SchedulingPolicy;
import com.google.common.annotations.VisibleForTesting; import com.google.common.annotations.VisibleForTesting;
public class FairSchedulingMode extends SchedulingMode { public class FairSharePolicy extends SchedulingPolicy {
@VisibleForTesting @VisibleForTesting
public static final String NAME = "FairShare"; public static final String NAME = "Fairshare";
private FairShareComparator comparator = new FairShareComparator(); private FairShareComparator comparator = new FairShareComparator();
@Override @Override
@ -211,4 +211,9 @@ private static Resource computeShare(Schedulable sched, double r2sRatio) {
share = Math.min(share, sched.getDemand().getMemory()); share = Math.min(share, sched.getDemand().getMemory());
return Resources.createResource((int) share); return Resources.createResource((int) share);
} }
@Override
public byte getApplicableDepth() {
return SchedulingPolicy.DEPTH_ANY;
}
} }

View File

@ -15,7 +15,7 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.modes; package org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies;
import java.io.Serializable; import java.io.Serializable;
import java.util.Collection; import java.util.Collection;
@ -24,11 +24,11 @@
import org.apache.hadoop.yarn.api.records.Resource; import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.Schedulable; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.Schedulable;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.Resources; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.Resources;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.SchedulingMode; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.SchedulingPolicy;
import com.google.common.annotations.VisibleForTesting; import com.google.common.annotations.VisibleForTesting;
public class FifoSchedulingMode extends SchedulingMode { public class FifoPolicy extends SchedulingPolicy {
@VisibleForTesting @VisibleForTesting
public static final String NAME = "FIFO"; public static final String NAME = "FIFO";
private FifoComparator comparator = new FifoComparator(); private FifoComparator comparator = new FifoComparator();
@ -73,4 +73,9 @@ public void computeShares(Collection<? extends Schedulable> schedulables,
sched.setFairShare(Resources.createResource(0)); sched.setFairShare(Resources.createResource(0));
} }
} }
@Override
public byte getApplicableDepth() {
return SchedulingPolicy.DEPTH_LEAF;
}
} }

View File

@ -68,7 +68,7 @@ public FakeSchedulable(Resource demand, Resource minShare, double weight, Resour
} }
@Override @Override
public Resource assignContainer(FSSchedulerNode node, boolean reserved) { public Resource assignContainer(FSSchedulerNode node) {
return null; return null;
} }

View File

@ -24,7 +24,7 @@
import junit.framework.Assert; import junit.framework.Assert;
import org.apache.hadoop.yarn.server.resourcemanager.resource.Resources; import org.apache.hadoop.yarn.server.resourcemanager.resource.Resources;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.modes.FairSchedulingMode; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies.FairSharePolicy;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
@ -33,12 +33,12 @@
*/ */
public class TestComputeFairShares { public class TestComputeFairShares {
private List<Schedulable> scheds; private List<Schedulable> scheds;
private SchedulingMode schedulingMode; private SchedulingPolicy schedulingMode;
@Before @Before
public void setUp() throws Exception { public void setUp() throws Exception {
scheds = new ArrayList<Schedulable>(); scheds = new ArrayList<Schedulable>();
schedulingMode = new FairSchedulingMode(); schedulingMode = new FairSharePolicy();
} }
/** /**

View File

@ -73,7 +73,7 @@
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeAddedSchedulerEvent; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeAddedSchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeRemovedSchedulerEvent; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeRemovedSchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeUpdateSchedulerEvent; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeUpdateSchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.modes.FifoSchedulingMode; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies.FifoPolicy;
import org.apache.hadoop.yarn.util.BuilderUtils; import org.apache.hadoop.yarn.util.BuilderUtils;
import org.junit.After; import org.junit.After;
import org.junit.Before; import org.junit.Before;
@ -1359,7 +1359,7 @@ public void testFifoWithinQueue() throws Exception {
FSSchedulerApp app2 = scheduler.applications.get(attId2); FSSchedulerApp app2 = scheduler.applications.get(attId2);
FSLeafQueue queue1 = scheduler.getQueueManager().getLeafQueue("queue1"); FSLeafQueue queue1 = scheduler.getQueueManager().getLeafQueue("queue1");
queue1.setSchedulingMode(new FifoSchedulingMode()); queue1.setPolicy(new FifoPolicy());
scheduler.update(); scheduler.update();
@ -1382,6 +1382,79 @@ public void testFifoWithinQueue() throws Exception {
assertEquals(1, app2.getLiveContainers().size()); assertEquals(1, app2.getLiveContainers().size());
} }
/**
* Test to verify the behavior of
* {@link FSQueue#assignContainer(FSSchedulerNode)})
*
* Create two queues under root (fifoQueue and fairParent), and two queues
* under fairParent (fairChild1 and fairChild2). Submit two apps to the
* fifoQueue and one each to the fairChild* queues, all apps requiring 4
* containers each of the total 16 container capacity
*
* Assert the number of containers for each app after 4, 8, 12 and 16 updates.
*
* @throws Exception
*/
@Test(timeout = 5000)
public void testAssignContainer() throws Exception {
final String user = "user1";
final String fifoQueue = "fifo";
final String fairParent = "fairParent";
final String fairChild1 = fairParent + ".fairChild1";
final String fairChild2 = fairParent + ".fairChild2";
RMNode node1 = MockNodes.newNodeInfo(1, Resources.createResource(8192));
RMNode node2 = MockNodes.newNodeInfo(1, Resources.createResource(8192));
NodeAddedSchedulerEvent nodeEvent1 = new NodeAddedSchedulerEvent(node1);
NodeAddedSchedulerEvent nodeEvent2 = new NodeAddedSchedulerEvent(node2);
scheduler.handle(nodeEvent1);
scheduler.handle(nodeEvent2);
ApplicationAttemptId attId1 =
createSchedulingRequest(1024, fifoQueue, user, 4);
ApplicationAttemptId attId2 =
createSchedulingRequest(1024, fairChild1, user, 4);
ApplicationAttemptId attId3 =
createSchedulingRequest(1024, fairChild2, user, 4);
ApplicationAttemptId attId4 =
createSchedulingRequest(1024, fifoQueue, user, 4);
FSSchedulerApp app1 = scheduler.applications.get(attId1);
FSSchedulerApp app2 = scheduler.applications.get(attId2);
FSSchedulerApp app3 = scheduler.applications.get(attId3);
FSSchedulerApp app4 = scheduler.applications.get(attId4);
scheduler.getQueueManager().getLeafQueue(fifoQueue)
.setPolicy(SchedulingPolicy.parse("fifo"));
scheduler.update();
NodeUpdateSchedulerEvent updateEvent1 = new NodeUpdateSchedulerEvent(node1);
NodeUpdateSchedulerEvent updateEvent2 = new NodeUpdateSchedulerEvent(node2);
for (int i = 0; i < 8; i++) {
scheduler.handle(updateEvent1);
scheduler.handle(updateEvent2);
if ((i + 1) % 2 == 0) {
// 4 node updates: fifoQueue should have received 2, and fairChild*
// should have received one each
String ERR =
"Wrong number of assigned containers after " + (i + 1) + " updates";
if (i < 4) {
// app1 req still not met
assertEquals(ERR, (i + 1), app1.getLiveContainers().size());
assertEquals(ERR, 0, app4.getLiveContainers().size());
} else {
// app1 req has been met, app4 should be served now
assertEquals(ERR, 4, app1.getLiveContainers().size());
assertEquals(ERR, (i - 3), app4.getLiveContainers().size());
}
assertEquals(ERR, (i + 1) / 2, app2.getLiveContainers().size());
assertEquals(ERR, (i + 1) / 2, app3.getLiveContainers().size());
}
}
}
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
@Test @Test

View File

@ -1,59 +0,0 @@
/**
* 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 static org.junit.Assert.assertTrue;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.modes.FairSchedulingMode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.modes.FifoSchedulingMode;
import org.junit.Test;
public class TestSchedulingMode {
@Test(timeout = 1000)
public void testParseSchedulingMode() throws AllocationConfigurationException {
// Class name
SchedulingMode sm = SchedulingMode
.parse(FairSchedulingMode.class.getName());
assertTrue("Invalid scheduler name",
sm.getName().equals(FairSchedulingMode.NAME));
// Canonical name
sm = SchedulingMode.parse(FairSchedulingMode.class
.getCanonicalName());
assertTrue("Invalid scheduler name",
sm.getName().equals(FairSchedulingMode.NAME));
// Class
sm = SchedulingMode.getInstance(FairSchedulingMode.class);
assertTrue("Invalid scheduler name",
sm.getName().equals(FairSchedulingMode.NAME));
// Shortname - fair
sm = SchedulingMode.parse("fair");
assertTrue("Invalid scheduler name",
sm.getName().equals(FairSchedulingMode.NAME));
// Shortname - fifo
sm = SchedulingMode.parse("fifo");
assertTrue("Invalid scheduler name",
sm.getName().equals(FifoSchedulingMode.NAME));
}
}

View File

@ -0,0 +1,109 @@
/**
* 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 static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies.FairSharePolicy;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies.FifoPolicy;
import org.junit.Test;
import org.mockito.Mockito;
public class TestSchedulingPolicy {
@Test(timeout = 1000)
public void testParseSchedulingPolicy()
throws AllocationConfigurationException {
// Class name
SchedulingPolicy sm = SchedulingPolicy
.parse(FairSharePolicy.class.getName());
assertTrue("Invalid scheduler name",
sm.getName().equals(FairSharePolicy.NAME));
// Canonical name
sm = SchedulingPolicy.parse(FairSharePolicy.class
.getCanonicalName());
assertTrue("Invalid scheduler name",
sm.getName().equals(FairSharePolicy.NAME));
// Class
sm = SchedulingPolicy.getInstance(FairSharePolicy.class);
assertTrue("Invalid scheduler name",
sm.getName().equals(FairSharePolicy.NAME));
// Shortname - fair
sm = SchedulingPolicy.parse("fair");
assertTrue("Invalid scheduler name",
sm.getName().equals(FairSharePolicy.NAME));
// Shortname - fifo
sm = SchedulingPolicy.parse("fifo");
assertTrue("Invalid scheduler name",
sm.getName().equals(FifoPolicy.NAME));
}
/**
* Trivial tests that make sure
* {@link SchedulingPolicy#isApplicableTo(SchedulingPolicy, byte)} works as
* expected for the possible values of depth
*
* @throws AllocationConfigurationException
*/
@Test(timeout = 1000)
public void testIsApplicableTo() throws AllocationConfigurationException {
final String ERR = "Broken SchedulingPolicy#isApplicableTo";
// fifo
SchedulingPolicy policy = SchedulingPolicy.parse("fifo");
assertTrue(ERR,
SchedulingPolicy.isApplicableTo(policy, SchedulingPolicy.DEPTH_LEAF));
assertFalse(ERR, SchedulingPolicy.isApplicableTo(
SchedulingPolicy.parse("fifo"), SchedulingPolicy.DEPTH_INTERMEDIATE));
assertFalse(ERR, SchedulingPolicy.isApplicableTo(
SchedulingPolicy.parse("fifo"), SchedulingPolicy.DEPTH_ROOT));
// fair
policy = SchedulingPolicy.parse("fair");
assertTrue(ERR,
SchedulingPolicy.isApplicableTo(policy, SchedulingPolicy.DEPTH_LEAF));
assertTrue(ERR, SchedulingPolicy.isApplicableTo(policy,
SchedulingPolicy.DEPTH_INTERMEDIATE));
assertTrue(ERR,
SchedulingPolicy.isApplicableTo(policy, SchedulingPolicy.DEPTH_ROOT));
assertTrue(ERR,
SchedulingPolicy.isApplicableTo(policy, SchedulingPolicy.DEPTH_PARENT));
assertTrue(ERR,
SchedulingPolicy.isApplicableTo(policy, SchedulingPolicy.DEPTH_ANY));
policy = Mockito.mock(SchedulingPolicy.class);
Mockito.when(policy.getApplicableDepth()).thenReturn(
SchedulingPolicy.DEPTH_PARENT);
assertTrue(ERR, SchedulingPolicy.isApplicableTo(policy,
SchedulingPolicy.DEPTH_INTERMEDIATE));
assertTrue(ERR,
SchedulingPolicy.isApplicableTo(policy, SchedulingPolicy.DEPTH_ROOT));
assertTrue(ERR,
SchedulingPolicy.isApplicableTo(policy, SchedulingPolicy.DEPTH_PARENT));
assertFalse(ERR,
SchedulingPolicy.isApplicableTo(policy, SchedulingPolicy.DEPTH_ANY));
}
}