Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

YARN-6040. Introduce api independent PendingAsk to replace usage of ResourceRequest within Scheduler classes. (wangda)

This commit is contained in:
Wangda Tan 2017-04-13 11:11:57 -07:00
parent 107f685f64
commit 408d23477f
20 changed files with 807 additions and 549 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/QueuePriorityContainerCandidateSelector.java
View File

@ -290,9 +290,9 @@ public class QueuePriorityContainerCandidateSelector
FiCaSchedulerApp app =
preemptionContext.getScheduler().getApplicationAttempt(
reservedContainer.getApplicationAttemptId());
ResourceRequest offswithRequest = app.getAppSchedulingInfo().getResourceRequest(
reservedContainer.getReservedSchedulerKey(), ResourceRequest.ANY);
if (!offswithRequest.getRelaxLocality()) {
if (!app.getAppSchedulingInfo().canDelayTo(
reservedContainer.getReservedSchedulerKey(), ResourceRequest.ANY)) {
// This is a hard locality request
return false;
}

146
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.java
View File

@ -34,16 +34,18 @@ import org.apache.hadoop.yarn.server.resourcemanager.RMServerUtils;
import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.SchedulingMode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.LocalitySchedulingPlacementSet;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.ResourceRequestUpdateResult;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.SchedulingPlacementSet;
import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.PendingAsk;
import org.apache.hadoop.yarn.util.resource.Resources;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@ -583,16 +585,10 @@ public class AppSchedulingInfo {
return schedulerKeys.keySet();
}
@SuppressWarnings("unchecked")
public Map<String, ResourceRequest> getResourceRequests(
SchedulerRequestKey schedulerKey) {
SchedulingPlacementSet ps = schedulerKeyToPlacementSets.get(schedulerKey);
if (null != ps) {
return ps.getResourceRequests();
}
return Collections.emptyMap();
}
/**
* Used by REST API to fetch ResourceRequest
* @return All pending ResourceRequests.
*/
public List<ResourceRequest> getAllResourceRequests() {
List<ResourceRequest> ret = new ArrayList<>();
try {
@ -606,27 +602,31 @@ public class AppSchedulingInfo {
return ret;
}
public ResourceRequest getResourceRequest(SchedulerRequestKey schedulerKey,
String resourceName) {
public SchedulingPlacementSet getFirstSchedulingPlacementSet() {
try {
this.readLock.lock();
SchedulingPlacementSet ps =
schedulerKeyToPlacementSets.get(schedulerKey);
return (ps == null) ? null : ps.getResourceRequest(resourceName);
readLock.lock();
for (SchedulerRequestKey key : schedulerKeys.keySet()) {
SchedulingPlacementSet ps = schedulerKeyToPlacementSets.get(key);
if (null != ps) {
return ps;
}
}
return null;
} finally {
this.readLock.unlock();
readLock.unlock();
}
}
public Resource getResource(SchedulerRequestKey schedulerKey) {
public PendingAsk getNextPendingAsk() {
try {
this.readLock.lock();
ResourceRequest request =
getResourceRequest(schedulerKey, ResourceRequest.ANY);
return (request == null) ? null : request.getCapability();
readLock.lock();
SchedulerRequestKey firstRequestKey = schedulerKeys.firstKey();
return getPendingAsk(firstRequestKey, ResourceRequest.ANY);
} finally {
this.readLock.unlock();
readLock.unlock();
}
}
/**
@ -649,9 +649,24 @@ public class AppSchedulingInfo {
return null;
}
public PendingAsk getPendingAsk(SchedulerRequestKey schedulerKey) {
return getPendingAsk(schedulerKey, ResourceRequest.ANY);
}
public PendingAsk getPendingAsk(SchedulerRequestKey schedulerKey,
String resourceName) {
try {
this.readLock.lock();
SchedulingPlacementSet ps = schedulerKeyToPlacementSets.get(schedulerKey);
return (ps == null) ? PendingAsk.ZERO : ps.getPendingAsk(resourceName);
} finally {
this.readLock.unlock();
}
}
/**
* Returns if the place (node/rack today) is either blacklisted by the
* application (user) or the system
* application (user) or the system.
*
* @param resourceName
* the resourcename
@ -724,7 +739,6 @@ public class AppSchedulingInfo {
public List<ResourceRequest> allocate(NodeType type,
SchedulerNode node, SchedulerRequestKey schedulerKey,
ResourceRequest request,
Container containerAllocated) {
try {
writeLock.lock();
@ -733,19 +747,13 @@ public class AppSchedulingInfo {
updateMetricsForAllocatedContainer(type, containerAllocated);
}
return schedulerKeyToPlacementSets.get(schedulerKey)
.allocate(schedulerKey, type, node, request);
return schedulerKeyToPlacementSets.get(schedulerKey).allocate(
schedulerKey, type, node);
} finally {
writeLock.unlock();
}
}
public List<ResourceRequest> allocate(NodeType type,
SchedulerNode node, SchedulerRequestKey schedulerKey,
Container containerAllocated) {
return allocate(type, node, schedulerKey, null, containerAllocated);
}
public void checkForDeactivation() {
if (schedulerKeys.isEmpty()) {
activeUsersManager.deactivateApplication(user, applicationId);
@ -758,18 +766,20 @@ public class AppSchedulingInfo {
QueueMetrics oldMetrics = queue.getMetrics();
QueueMetrics newMetrics = newQueue.getMetrics();
for (SchedulingPlacementSet ps : schedulerKeyToPlacementSets.values()) {
ResourceRequest request = ps.getResourceRequest(ResourceRequest.ANY);
if (request != null && request.getNumContainers() > 0) {
oldMetrics.decrPendingResources(user, request.getNumContainers(),
request.getCapability());
newMetrics.incrPendingResources(user, request.getNumContainers(),
request.getCapability());
PendingAsk ask = ps.getPendingAsk(ResourceRequest.ANY);
if (ask.getCount() > 0) {
oldMetrics.decrPendingResources(user, ask.getCount(),
ask.getPerAllocationResource());
newMetrics.incrPendingResources(user, ask.getCount(),
ask.getPerAllocationResource());
Resource delta = Resources.multiply(request.getCapability(),
request.getNumContainers());
Resource delta = Resources.multiply(ask.getPerAllocationResource(),
ask.getCount());
// Update Queue
queue.decPendingResource(request.getNodeLabelExpression(), delta);
newQueue.incPendingResource(request.getNodeLabelExpression(), delta);
queue.decPendingResource(
ps.getPrimaryRequestedNodePartition(), delta);
newQueue.incPendingResource(
ps.getPrimaryRequestedNodePartition(), delta);
}
}
oldMetrics.moveAppFrom(this);
@ -789,16 +799,16 @@ public class AppSchedulingInfo {
this.writeLock.lock();
QueueMetrics metrics = queue.getMetrics();
for (SchedulingPlacementSet ps : schedulerKeyToPlacementSets.values()) {
ResourceRequest request = ps.getResourceRequest(ResourceRequest.ANY);
if (request != null && request.getNumContainers() > 0) {
metrics.decrPendingResources(user, request.getNumContainers(),
request.getCapability());
PendingAsk ask = ps.getPendingAsk(ResourceRequest.ANY);
if (ask.getCount() > 0) {
metrics.decrPendingResources(user, ask.getCount(),
ask.getPerAllocationResource());
// Update Queue
queue.decPendingResource(
request.getNodeLabelExpression(),
Resources.multiply(request.getCapability(),
request.getNumContainers()));
ps.getPrimaryRequestedNodePartition(),
Resources.multiply(ask.getPerAllocationResource(),
ask.getCount()));
}
}
metrics.finishAppAttempt(applicationId, pending, user);
@ -906,4 +916,38 @@ public class AppSchedulingInfo {
return (SchedulingPlacementSet<N>) schedulerKeyToPlacementSets.get(
schedulerkey);
}
/**
* Can delay to next?.
*
* @param schedulerKey schedulerKey
* @param resourceName resourceName
*
* @return If request exists, return {relaxLocality}
* Otherwise, return true.
*/
public boolean canDelayTo(
SchedulerRequestKey schedulerKey, String resourceName) {
try {
this.readLock.lock();
SchedulingPlacementSet ps =
schedulerKeyToPlacementSets.get(schedulerKey);
return (ps == null) || ps.canDelayTo(resourceName);
} finally {
this.readLock.unlock();
}
}
public boolean acceptNodePartition(SchedulerRequestKey schedulerKey,
String nodePartition, SchedulingMode schedulingMode) {
try {
this.readLock.lock();
SchedulingPlacementSet ps =
schedulerKeyToPlacementSets.get(schedulerKey);
return (ps != null) && ps.acceptNodePartition(nodePartition,
schedulingMode);
} finally {
this.readLock.unlock();
}
}
}

48
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java
View File

@ -18,7 +18,6 @@
package org.apache.hadoop.yarn.server.resourcemanager.scheduler;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@ -78,6 +77,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.Scheduli
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.SchedulingPlacementSet;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.policy.SchedulableEntity;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.PendingAsk;
import org.apache.hadoop.yarn.server.scheduler.OpportunisticContainerContext;
import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
import org.apache.hadoop.yarn.state.InvalidStateTransitionException;
@ -283,11 +283,6 @@ public class SchedulerApplicationAttempt implements SchedulableEntity {
return appSchedulingInfo.getUser();
}
public Map<String, ResourceRequest> getResourceRequests(
SchedulerRequestKey schedulerKey) {
return appSchedulingInfo.getResourceRequests(schedulerKey);
}
public Set<ContainerId> getPendingRelease() {
return this.pendingRelease;
}
@ -299,34 +294,28 @@ public class SchedulerApplicationAttempt implements SchedulableEntity {
public Collection<SchedulerRequestKey> getSchedulerKeys() {
return appSchedulingInfo.getSchedulerKeys();
}
public ResourceRequest getResourceRequest(
public PendingAsk getPendingAsk(
SchedulerRequestKey schedulerKey, String resourceName) {
try {
readLock.lock();
return appSchedulingInfo.getResourceRequest(schedulerKey, resourceName);
} finally {
readLock.unlock();
}
}
public int getTotalRequiredResources(
SchedulerRequestKey schedulerKey) {
try {
readLock.lock();
ResourceRequest request =
getResourceRequest(schedulerKey, ResourceRequest.ANY);
return request == null ? 0 : request.getNumContainers();
return appSchedulingInfo.getPendingAsk(schedulerKey, resourceName);
} finally {
readLock.unlock();
}
}
public Resource getResource(SchedulerRequestKey schedulerKey) {
public int getOutstandingAsksCount(SchedulerRequestKey schedulerKey) {
return getOutstandingAsksCount(schedulerKey, ResourceRequest.ANY);
}
public int getOutstandingAsksCount(SchedulerRequestKey schedulerKey,
String resourceName) {
try {
readLock.lock();
return appSchedulingInfo.getResource(schedulerKey);
SchedulingPlacementSet ps = appSchedulingInfo.getSchedulingPlacementSet(
schedulerKey);
return ps == null ? 0 : ps.getOutstandingAsksCount(resourceName);
} finally {
readLock.unlock();
}
@ -625,16 +614,13 @@ public class SchedulerApplicationAttempt implements SchedulableEntity {
try {
readLock.lock();
for (SchedulerRequestKey schedulerKey : getSchedulerKeys()) {
Map<String, ResourceRequest> requests = getResourceRequests(
schedulerKey);
if (requests != null) {
SchedulingPlacementSet ps = getSchedulingPlacementSet(schedulerKey);
if (ps != null &&
ps.getOutstandingAsksCount(ResourceRequest.ANY) > 0) {
LOG.debug("showRequests:" + " application=" + getApplicationId()
+ " headRoom=" + getHeadroom() + " currentConsumption="
+ attemptResourceUsage.getUsed().getMemorySize());
for (ResourceRequest request : requests.values()) {
LOG.debug("showRequests:" + " application=" + getApplicationId()
+ " request=" + request);
}
ps.showRequests();
}
}
} finally {

4
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/AbstractContainerAllocator.java
View File

@ -24,6 +24,7 @@ import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AppSchedulingInfo;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivitiesLogger;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivitiesManager;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivityDiagnosticConstant;
@ -46,6 +47,7 @@ public abstract class AbstractContainerAllocator {
private static final Log LOG = LogFactory.getLog(AbstractContainerAllocator.class);
FiCaSchedulerApp application;
AppSchedulingInfo appInfo;
final ResourceCalculator rc;
final RMContext rmContext;
ActivitiesManager activitiesManager;
@ -59,6 +61,8 @@ public abstract class AbstractContainerAllocator {
ResourceCalculator rc, RMContext rmContext,
ActivitiesManager activitiesManager) {
this.application = application;
this.appInfo =
application == null ? null : application.getAppSchedulingInfo();
this.rc = rc;
this.rmContext = rmContext;
this.activitiesManager = activitiesManager;

136
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/RegularContainerAllocator.java
View File

@ -39,7 +39,6 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceLimits;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerAppUtils;
import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivityDiagnosticConstant;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivitiesLogger;
@ -53,6 +52,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaS
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.PlacementSet;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.PlacementSetUtils;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.SchedulingPlacementSet;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.PendingAsk;
import org.apache.hadoop.yarn.server.utils.BuilderUtils;
import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
import org.apache.hadoop.yarn.util.resource.Resources;
@ -63,8 +63,6 @@ import org.apache.hadoop.yarn.util.resource.Resources;
*/
public class RegularContainerAllocator extends AbstractContainerAllocator {
private static final Log LOG = LogFactory.getLog(RegularContainerAllocator.class);
private ResourceRequest lastResourceRequest = null;
public RegularContainerAllocator(FiCaSchedulerApp application,
ResourceCalculator rc, RMContext rmContext,
@ -102,9 +100,10 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
Priority priority = schedulerKey.getPriority();
FiCaSchedulerNode node = PlacementSetUtils.getSingleNode(ps);
ResourceRequest anyRequest =
application.getResourceRequest(schedulerKey, ResourceRequest.ANY);
if (null == anyRequest) {
PendingAsk offswitchPendingAsk = application.getPendingAsk(schedulerKey,
ResourceRequest.ANY);
if (offswitchPendingAsk.getCount() <= 0) {
ActivitiesLogger.APP.recordSkippedAppActivityWithoutAllocation(
activitiesManager, node, application, priority,
ActivityDiagnosticConstant.PRIORITY_SKIPPED_BECAUSE_NULL_ANY_REQUEST);
@ -112,10 +111,10 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
}
// Required resource
Resource required = anyRequest.getCapability();
Resource required = offswitchPendingAsk.getPerAllocationResource();
// Do we need containers at this 'priority'?
if (application.getTotalRequiredResources(schedulerKey) <= 0) {
if (application.getOutstandingAsksCount(schedulerKey) <= 0) {
ActivitiesLogger.APP.recordSkippedAppActivityWithoutAllocation(
activitiesManager, node, application, priority,
ActivityDiagnosticConstant.APPLICATION_PRIORITY_DO_NOT_NEED_RESOURCE);
@ -140,11 +139,9 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
}
}
// Is the node-label-expression of this offswitch resource request
// matches the node's label?
// Is the nodePartition of pending request matches the node's partition
// If not match, jump to next priority.
if (!SchedulerUtils.checkResourceRequestMatchingNodePartition(
anyRequest.getNodeLabelExpression(), ps.getPartition(),
if (!appInfo.acceptNodePartition(schedulerKey, node.getPartition(),
schedulingMode)) {
ActivitiesLogger.APP.recordSkippedAppActivityWithoutAllocation(
activitiesManager, node, application, priority,
@ -181,8 +178,11 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
// This is to make sure non-partitioned-resource-request will prefer
// to be allocated to non-partitioned nodes
int missedNonPartitionedRequestSchedulingOpportunity = 0;
if (anyRequest.getNodeLabelExpression()
.equals(RMNodeLabelsManager.NO_LABEL)) {
// Only do this when request associated with given scheduler key accepts
// NO_LABEL under RESPECT_EXCLUSIVITY mode
if (StringUtils.equals(RMNodeLabelsManager.NO_LABEL,
appInfo.getSchedulingPlacementSet(schedulerKey)
.getPrimaryRequestedNodePartition())) {
missedNonPartitionedRequestSchedulingOpportunity =
application.addMissedNonPartitionedRequestSchedulingOpportunity(
schedulerKey);
@ -263,8 +263,9 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
public float getLocalityWaitFactor(
SchedulerRequestKey schedulerKey, int clusterNodes) {
// Estimate: Required unique resources (i.e. hosts + racks)
int requiredResources =
Math.max(application.getResourceRequests(schedulerKey).size() - 1, 0);
int requiredResources = Math.max(
application.getSchedulingPlacementSet(schedulerKey)
.getUniqueLocationAsks() - 1, 0);
// waitFactor can't be more than '1'
// i.e. no point skipping more than clustersize opportunities
@ -296,13 +297,12 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
}
// If we have only ANY requests for this schedulerKey, we should not
// delay its scheduling.
if (application.getResourceRequests(schedulerKey).size() == 1) {
if (application.getSchedulingPlacementSet(schedulerKey)
.getUniqueLocationAsks() == 1) {
return true;
}
// 'Delay' off-switch
ResourceRequest offSwitchRequest =
application.getResourceRequest(schedulerKey, ResourceRequest.ANY);
long missedOpportunities =
application.getSchedulingOpportunities(schedulerKey);
@ -310,8 +310,8 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
if (application.getCSLeafQueue().getRackLocalityAdditionalDelay() > -1) {
return missedOpportunities > getActualRackLocalityDelay();
} else {
long requiredContainers =
application.getTotalRequiredResources(schedulerKey);
long requiredContainers = application.getOutstandingAsksCount(
schedulerKey);
float localityWaitFactor = getLocalityWaitFactor(schedulerKey,
rmContext.getScheduler().getNumClusterNodes());
// Cap the delay by the number of nodes in the cluster.
@ -321,9 +321,8 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
}
// Check if we need containers on this rack
ResourceRequest rackLocalRequest =
application.getResourceRequest(schedulerKey, node.getRackName());
if (rackLocalRequest == null || rackLocalRequest.getNumContainers() <= 0) {
if (application.getOutstandingAsksCount(schedulerKey, node.getRackName())
<= 0) {
return false;
}
@ -338,24 +337,21 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
// Check if we need containers on this host
if (type == NodeType.NODE_LOCAL) {
// Now check if we need containers on this host...
ResourceRequest nodeLocalRequest =
application.getResourceRequest(schedulerKey, node.getNodeName());
if (nodeLocalRequest != null) {
return nodeLocalRequest.getNumContainers() > 0;
}
return application.getOutstandingAsksCount(schedulerKey,
node.getNodeName()) > 0;
}
return false;
}
private ContainerAllocation assignNodeLocalContainers(
Resource clusterResource, ResourceRequest nodeLocalResourceRequest,
Resource clusterResource, PendingAsk nodeLocalAsk,
FiCaSchedulerNode node, SchedulerRequestKey schedulerKey,
RMContainer reservedContainer, SchedulingMode schedulingMode,
ResourceLimits currentResoureLimits) {
if (canAssign(schedulerKey, node, NodeType.NODE_LOCAL, reservedContainer)) {
return assignContainer(clusterResource, node, schedulerKey,
nodeLocalResourceRequest, NodeType.NODE_LOCAL, reservedContainer,
nodeLocalAsk, NodeType.NODE_LOCAL, reservedContainer,
schedulingMode, currentResoureLimits);
}
@ -367,13 +363,13 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
}
private ContainerAllocation assignRackLocalContainers(
Resource clusterResource, ResourceRequest rackLocalResourceRequest,
Resource clusterResource, PendingAsk rackLocalAsk,
FiCaSchedulerNode node, SchedulerRequestKey schedulerKey,
RMContainer reservedContainer, SchedulingMode schedulingMode,
ResourceLimits currentResoureLimits) {
if (canAssign(schedulerKey, node, NodeType.RACK_LOCAL, reservedContainer)) {
return assignContainer(clusterResource, node, schedulerKey,
rackLocalResourceRequest, NodeType.RACK_LOCAL, reservedContainer,
rackLocalAsk, NodeType.RACK_LOCAL, reservedContainer,
schedulingMode, currentResoureLimits);
}
@ -385,13 +381,13 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
}
private ContainerAllocation assignOffSwitchContainers(
Resource clusterResource, ResourceRequest offSwitchResourceRequest,
Resource clusterResource, PendingAsk offSwitchAsk,
FiCaSchedulerNode node, SchedulerRequestKey schedulerKey,
RMContainer reservedContainer, SchedulingMode schedulingMode,
ResourceLimits currentResoureLimits) {
if (canAssign(schedulerKey, node, NodeType.OFF_SWITCH, reservedContainer)) {
return assignContainer(clusterResource, node, schedulerKey,
offSwitchResourceRequest, NodeType.OFF_SWITCH, reservedContainer,
offSwitchAsk, NodeType.OFF_SWITCH, reservedContainer,
schedulingMode, currentResoureLimits);
}
@ -413,12 +409,12 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
NodeType requestLocalityType = null;
// Data-local
ResourceRequest nodeLocalResourceRequest =
application.getResourceRequest(schedulerKey, node.getNodeName());
if (nodeLocalResourceRequest != null) {
PendingAsk nodeLocalAsk =
application.getPendingAsk(schedulerKey, node.getNodeName());
if (nodeLocalAsk.getCount() > 0) {
requestLocalityType = NodeType.NODE_LOCAL;
allocation =
assignNodeLocalContainers(clusterResource, nodeLocalResourceRequest,
assignNodeLocalContainers(clusterResource, nodeLocalAsk,
node, schedulerKey, reservedContainer, schedulingMode,
currentResoureLimits);
if (Resources.greaterThan(rc, clusterResource,
@ -429,10 +425,10 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
}
// Rack-local
ResourceRequest rackLocalResourceRequest =
application.getResourceRequest(schedulerKey, node.getRackName());
if (rackLocalResourceRequest != null) {
if (!rackLocalResourceRequest.getRelaxLocality()) {
PendingAsk rackLocalAsk =
application.getPendingAsk(schedulerKey, node.getRackName());
if (rackLocalAsk.getCount() > 0) {
if (!appInfo.canDelayTo(schedulerKey, node.getRackName())) {
ActivitiesLogger.APP.recordSkippedAppActivityWithoutAllocation(
activitiesManager, node, application, priority,
ActivityDiagnosticConstant.SKIP_PRIORITY_BECAUSE_OF_RELAX_LOCALITY);
@ -444,7 +440,7 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
requestLocalityType;
allocation =
assignRackLocalContainers(clusterResource, rackLocalResourceRequest,
assignRackLocalContainers(clusterResource, rackLocalAsk,
node, schedulerKey, reservedContainer, schedulingMode,
currentResoureLimits);
if (Resources.greaterThan(rc, clusterResource,
@ -455,10 +451,10 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
}
// Off-switch
ResourceRequest offSwitchResourceRequest =
application.getResourceRequest(schedulerKey, ResourceRequest.ANY);
if (offSwitchResourceRequest != null) {
if (!offSwitchResourceRequest.getRelaxLocality()) {
PendingAsk offSwitchAsk =
application.getPendingAsk(schedulerKey, ResourceRequest.ANY);
if (offSwitchAsk.getCount() > 0) {
if (!appInfo.canDelayTo(schedulerKey, ResourceRequest.ANY)) {
ActivitiesLogger.APP.recordSkippedAppActivityWithoutAllocation(
activitiesManager, node, application, priority,
ActivityDiagnosticConstant.SKIP_PRIORITY_BECAUSE_OF_RELAX_LOCALITY);
@ -470,7 +466,7 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
requestLocalityType;
allocation =
assignOffSwitchContainers(clusterResource, offSwitchResourceRequest,
assignOffSwitchContainers(clusterResource, offSwitchAsk,
node, schedulerKey, reservedContainer, schedulingMode,
currentResoureLimits);
@ -491,41 +487,25 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
private ContainerAllocation assignContainer(Resource clusterResource,
FiCaSchedulerNode node, SchedulerRequestKey schedulerKey,
ResourceRequest request, NodeType type, RMContainer rmContainer,
PendingAsk pendingAsk, NodeType type, RMContainer rmContainer,
SchedulingMode schedulingMode, ResourceLimits currentResoureLimits) {
Priority priority = schedulerKey.getPriority();
lastResourceRequest = request;
if (LOG.isDebugEnabled()) {
LOG.debug("assignContainers: node=" + node.getNodeName()
+ " application=" + application.getApplicationId()
+ " priority=" + schedulerKey.getPriority()
+ " request=" + request + " type=" + type);
+ " pendingAsk=" + pendingAsk + " type=" + type);
}
// check if the resource request can access the label
if (!SchedulerUtils.checkResourceRequestMatchingNodePartition(
request.getNodeLabelExpression(), node.getPartition(),
schedulingMode)) {
// this is a reserved container, but we cannot allocate it now according
// to label not match. This can be caused by node label changed
// We should un-reserve this container.
ActivitiesLogger.APP.recordAppActivityWithoutAllocation(activitiesManager,
node, application, priority,
ActivityDiagnosticConstant.REQUEST_CAN_NOT_ACCESS_NODE_LABEL,
ActivityState.REJECTED);
return new ContainerAllocation(rmContainer, null,
AllocationState.LOCALITY_SKIPPED);
}
Resource capability = request.getCapability();
Resource capability = pendingAsk.getPerAllocationResource();
Resource available = node.getUnallocatedResource();
Resource totalResource = node.getTotalResource();
if (!Resources.lessThanOrEqual(rc, clusterResource,
capability, totalResource)) {
LOG.warn("Node : " + node.getNodeID()
+ " does not have sufficient resource for request : " + request
+ " does not have sufficient resource for ask : " + pendingAsk
+ " node total capability : " + node.getTotalResource());
// Skip this locality request
ActivitiesLogger.APP.recordSkippedAppActivityWithoutAllocation(
@ -617,9 +597,8 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
}
}
ContainerAllocation result =
new ContainerAllocation(unreservedContainer, request.getCapability(),
AllocationState.ALLOCATED);
ContainerAllocation result = new ContainerAllocation(unreservedContainer,
pendingAsk.getPerAllocationResource(), AllocationState.ALLOCATED);
result.containerNodeType = type;
result.setToKillContainers(toKillContainers);
return result;
@ -643,9 +622,8 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
}
}
ContainerAllocation result =
new ContainerAllocation(null, request.getCapability(),
AllocationState.RESERVED);
ContainerAllocation result = new ContainerAllocation(null,
pendingAsk.getPerAllocationResource(), AllocationState.RESERVED);
result.containerNodeType = type;
result.setToKillContainers(null);
return result;
@ -661,7 +639,7 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
boolean shouldAllocOrReserveNewContainer(
SchedulerRequestKey schedulerKey, Resource required) {
int requiredContainers =
application.getTotalRequiredResources(schedulerKey);
application.getOutstandingAsksCount(schedulerKey);
int reservedContainers = application.getNumReservedContainers(schedulerKey);
int starvation = 0;
if (reservedContainers > 0) {
@ -716,7 +694,7 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
SchedulerRequestKey schedulerKey, Container container) {
// Inform the application
RMContainer allocatedContainer = application.allocate(node, schedulerKey,
lastResourceRequest, container);
container);
allocationResult.updatedContainer = allocatedContainer;
@ -820,7 +798,7 @@ public class RegularContainerAllocator extends AbstractContainerAllocator {
}
} else {
// pre-check when allocating reserved container
if (application.getTotalRequiredResources(schedulerKey) == 0) {
if (application.getOutstandingAsksCount(schedulerKey) == 0) {
// Release
return new ContainerAllocation(reservedContainer, null,
AllocationState.QUEUE_SKIPPED);

57
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/common/PendingAsk.java Normal file
View File

@ -0,0 +1,57 @@
/**
* 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.common;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.util.resource.Resources;
/**
* {@link PendingAsk} is the class to include minimal information of how much
* resource to ask under constraints (e.g. on one host / rack / node-attributes)
* , etc.
*/
public class PendingAsk {
private final Resource perAllocationResource;
private final int count;
public final static PendingAsk ZERO = new PendingAsk(Resources.none(), 0);
public PendingAsk(Resource res, int num) {
this.perAllocationResource = res;
this.count = num;
}
public Resource getPerAllocationResource() {
return perAllocationResource;
}
public int getCount() {
return count;
}
@Override
public String toString() {
StringBuilder sb = new StringBuilder();
sb.append("<per-allocation-resource=");
sb.append(getPerAllocationResource());
sb.append(",repeat=");
sb.append(getCount());
sb.append(">");
return sb.toString();
}
}

55
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
View File

@ -71,6 +71,8 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.Placeme
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.SchedulingPlacementSet;
import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.PendingAsk;
import org.apache.hadoop.yarn.server.utils.BuilderUtils;
import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator;
import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
@ -209,8 +211,7 @@ public class FiCaSchedulerApp extends SchedulerApplicationAttempt {
}
public RMContainer allocate(FiCaSchedulerNode node,
SchedulerRequestKey schedulerKey, ResourceRequest request,
Container container) {
SchedulerRequestKey schedulerKey, Container container) {
try {
readLock.lock();
@ -220,7 +221,16 @@ public class FiCaSchedulerApp extends SchedulerApplicationAttempt {
// Required sanity check - AM can call 'allocate' to update resource
// request without locking the scheduler, hence we need to check
if (getTotalRequiredResources(schedulerKey) <= 0) {
if (getOutstandingAsksCount(schedulerKey) <= 0) {
return null;
}
SchedulingPlacementSet<FiCaSchedulerNode> ps =
appSchedulingInfo.getSchedulingPlacementSet(schedulerKey);
if (null == ps) {
LOG.warn("Failed to get " + SchedulingPlacementSet.class.getName()
+ " for application=" + getApplicationId() + " schedulerRequestKey="
+ schedulerKey);
return null;
}
@ -228,7 +238,7 @@ public class FiCaSchedulerApp extends SchedulerApplicationAttempt {
RMContainer rmContainer = new RMContainerImpl(container, schedulerKey,
this.getApplicationAttemptId(), node.getNodeID(),
appSchedulingInfo.getUser(), this.rmContext,
request.getNodeLabelExpression());
ps.getPrimaryRequestedNodePartition());
((RMContainerImpl) rmContainer).setQueueName(this.getQueueName());
// FIXME, should set when confirmed
@ -697,21 +707,36 @@ public class FiCaSchedulerApp extends SchedulerApplicationAttempt {
return false;
}
public synchronized Map<String, Resource> getTotalPendingRequestsPerPartition() {
public Map<String, Resource> getTotalPendingRequestsPerPartition() {
try {
readLock.lock();
Map<String, Resource> ret = new HashMap<String, Resource>();
Resource res = null;
for (SchedulerRequestKey key : appSchedulingInfo.getSchedulerKeys()) {
ResourceRequest rr = appSchedulingInfo.getResourceRequest(key, "*");
if ((res = ret.get(rr.getNodeLabelExpression())) == null) {
res = Resources.createResource(0, 0);
ret.put(rr.getNodeLabelExpression(), res);
Map<String, Resource> ret = new HashMap<>();
for (SchedulerRequestKey schedulerKey : appSchedulingInfo
.getSchedulerKeys()) {
SchedulingPlacementSet<FiCaSchedulerNode> ps =
appSchedulingInfo.getSchedulingPlacementSet(schedulerKey);
String nodePartition = ps.getPrimaryRequestedNodePartition();
Resource res = ret.get(nodePartition);
if (null == res) {
res = Resources.createResource(0);
ret.put(nodePartition, res);
}
PendingAsk ask = ps.getPendingAsk(ResourceRequest.ANY);
if (ask.getCount() > 0) {
Resources.addTo(res, Resources
.multiply(ask.getPerAllocationResource(),
ask.getCount()));
}
}
Resources.addTo(res,
Resources.multiply(rr.getCapability(), rr.getNumContainers()));
return ret;
} finally {
readLock.unlock();
}
return ret;
}
public void markContainerForPreemption(ContainerId cont) {

157
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
View File

@ -18,16 +18,6 @@
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.Private;
@ -56,11 +46,22 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.PendingAsk;
import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
import org.apache.hadoop.yarn.server.utils.BuilderUtils;
import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator;
import org.apache.hadoop.yarn.util.resource.Resources;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
/**
* Represents an application attempt from the viewpoint of the Fair Scheduler.
*/
@ -422,7 +423,7 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
}
public RMContainer allocate(NodeType type, FSSchedulerNode node,
SchedulerRequestKey schedulerKey, ResourceRequest request,
SchedulerRequestKey schedulerKey, PendingAsk pendingAsk,
Container reservedContainer) {
RMContainer rmContainer;
Container container;
@ -443,13 +444,13 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
// Required sanity check - AM can call 'allocate' to update resource
// request without locking the scheduler, hence we need to check
if (getTotalRequiredResources(schedulerKey) <= 0) {
if (getOutstandingAsksCount(schedulerKey) <= 0) {
return null;
}
container = reservedContainer;
if (container == null) {
container = createContainer(node, request.getCapability(),
container = createContainer(node, pendingAsk.getPerAllocationResource(),
schedulerKey);
}
@ -465,7 +466,7 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
// Update consumption and track allocations
List<ResourceRequest> resourceRequestList = appSchedulingInfo.allocate(
type, node, schedulerKey, request, container);
type, node, schedulerKey, container);
this.attemptResourceUsage.incUsed(container.getResource());
// Update resource requests related to "request" and store in RMContainer
@ -642,7 +643,7 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
* in {@link FSSchedulerNode}..
* return whether reservation was possible with the current threshold limits
*/
private boolean reserve(ResourceRequest request, FSSchedulerNode node,
private boolean reserve(Resource perAllocationResource, FSSchedulerNode node,
Container reservedContainer, NodeType type,
SchedulerRequestKey schedulerKey) {
@ -651,7 +652,7 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
" app_id=" + getApplicationId());
if (reservedContainer == null) {
reservedContainer =
createContainer(node, request.getCapability(),
createContainer(node, perAllocationResource,
schedulerKey);
getMetrics().reserveResource(getUser(),
reservedContainer.getResource());
@ -773,8 +774,8 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
*
* @param node
* The node to try placing the container on.
* @param request
* The ResourceRequest we're trying to satisfy.
* @param pendingAsk
* The {@link PendingAsk} we're trying to satisfy.
* @param type
* The locality of the assignment.
* @param reserved
@ -786,11 +787,11 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
* made, returns an empty resource.
*/
private Resource assignContainer(
FSSchedulerNode node, ResourceRequest request, NodeType type,
FSSchedulerNode node, PendingAsk pendingAsk, NodeType type,
boolean reserved, SchedulerRequestKey schedulerKey) {
// How much does this request need?
Resource capability = request.getCapability();
Resource capability = pendingAsk.getPerAllocationResource();
// How much does the node have?
Resource available = node.getUnallocatedResource();
@ -804,7 +805,7 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
if (Resources.fitsIn(capability, available)) {
// Inform the application of the new container for this request
RMContainer allocatedContainer =
allocate(type, node, schedulerKey, request,
allocate(type, node, schedulerKey, pendingAsk,
reservedContainer);
if (allocatedContainer == null) {
// Did the application need this resource?
@ -841,7 +842,7 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
// The desired container won't fit here, so reserve
if (isReservable(capability) &&
reserve(request, node, reservedContainer, type, schedulerKey)) {
reserve(pendingAsk.getPerAllocationResource(), node, reservedContainer, type, schedulerKey)) {
updateAMDiagnosticMsg(capability, " exceeds the available resources of "
+ "the node and the request is reserved)");
if (LOG.isDebugEnabled()) {
@ -867,19 +868,16 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
getQueue().getPolicy().getResourceCalculator(), capacity);
}
private boolean hasNodeOrRackLocalRequests(SchedulerRequestKey schedulerKey) {
return getResourceRequests(schedulerKey).size() > 1;
}
/**
* Whether the AM container for this app is over maxAMShare limit.
*/
private boolean isOverAMShareLimit() {
// Check the AM resource usage for the leaf queue
if (!isAmRunning() && !getUnmanagedAM()) {
List<ResourceRequest> ask = appSchedulingInfo.getAllResourceRequests();
if (ask.isEmpty() || !getQueue().canRunAppAM(
ask.get(0).getCapability())) {
// Return true if we have not ask, or queue is not be able to run app's AM
PendingAsk ask = appSchedulingInfo.getNextPendingAsk();
if (ask.getCount() == 0 || !getQueue().canRunAppAM(
ask.getPerAllocationResource())) {
return true;
}
}
@ -901,6 +899,11 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
// (not scheduled) in order to promote better locality.
try {
writeLock.lock();
// TODO (wandga): All logics in this method should be added to
// SchedulerPlacement#canDelayTo which is independent from scheduler.
// Scheduler can choose to use various/pluggable delay-scheduling
// implementation.
for (SchedulerRequestKey schedulerKey : keysToTry) {
// Skip it for reserved container, since
// we already check it in isValidReservation.
@ -910,14 +913,16 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
addSchedulingOpportunity(schedulerKey);
ResourceRequest rackLocalRequest = getResourceRequest(schedulerKey,
PendingAsk rackLocalPendingAsk = getPendingAsk(schedulerKey,
node.getRackName());
ResourceRequest localRequest = getResourceRequest(schedulerKey,
PendingAsk nodeLocalPendingAsk = getPendingAsk(schedulerKey,
node.getNodeName());
if (localRequest != null && !localRequest.getRelaxLocality()) {
if (nodeLocalPendingAsk.getCount() > 0
&& !appSchedulingInfo.canDelayTo(schedulerKey,
node.getNodeName())) {
LOG.warn("Relax locality off is not supported on local request: "
+ localRequest);
+ nodeLocalPendingAsk);
}
NodeType allowedLocality;
@ -933,23 +938,23 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
scheduler.getRackLocalityThreshold());
}
if (rackLocalRequest != null && rackLocalRequest.getNumContainers() != 0
&& localRequest != null && localRequest.getNumContainers() != 0) {
if (rackLocalPendingAsk.getCount() > 0
&& nodeLocalPendingAsk.getCount() > 0) {
if (LOG.isTraceEnabled()) {
LOG.trace("Assign container on " + node.getNodeName()
+ " node, assignType: NODE_LOCAL" + ", allowedLocality: "
+ allowedLocality + ", priority: " + schedulerKey.getPriority()
+ ", app attempt id: " + this.attemptId);
}
return assignContainer(node, localRequest, NodeType.NODE_LOCAL,
return assignContainer(node, nodeLocalPendingAsk, NodeType.NODE_LOCAL,
reserved, schedulerKey);
}
if (rackLocalRequest != null && !rackLocalRequest.getRelaxLocality()) {
if (!appSchedulingInfo.canDelayTo(schedulerKey, node.getRackName())) {
continue;
}
if (rackLocalRequest != null && rackLocalRequest.getNumContainers() != 0
if (rackLocalPendingAsk.getCount() > 0
&& (allowedLocality.equals(NodeType.RACK_LOCAL) || allowedLocality
.equals(NodeType.OFF_SWITCH))) {
if (LOG.isTraceEnabled()) {
@ -958,20 +963,19 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
+ allowedLocality + ", priority: " + schedulerKey.getPriority()
+ ", app attempt id: " + this.attemptId);
}
return assignContainer(node, rackLocalRequest, NodeType.RACK_LOCAL,
return assignContainer(node, rackLocalPendingAsk, NodeType.RACK_LOCAL,
reserved, schedulerKey);
}
ResourceRequest offSwitchRequest = getResourceRequest(schedulerKey,
PendingAsk offswitchAsk = getPendingAsk(schedulerKey,
ResourceRequest.ANY);
if (offSwitchRequest != null && !offSwitchRequest.getRelaxLocality()) {
if (!appSchedulingInfo.canDelayTo(schedulerKey, ResourceRequest.ANY)) {
continue;
}
if (offSwitchRequest != null
&& offSwitchRequest.getNumContainers() != 0) {
if (!hasNodeOrRackLocalRequests(schedulerKey) || allowedLocality
.equals(NodeType.OFF_SWITCH)) {
if (offswitchAsk.getCount() > 0) {
if (getSchedulingPlacementSet(schedulerKey).getUniqueLocationAsks()
<= 1 || allowedLocality.equals(NodeType.OFF_SWITCH)) {
if (LOG.isTraceEnabled()) {
LOG.trace("Assign container on " + node.getNodeName()
+ " node, assignType: OFF_SWITCH" + ", allowedLocality: "
@ -979,7 +983,7 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
+ schedulerKey.getPriority()
+ ", app attempt id: " + this.attemptId);
}
return assignContainer(node, offSwitchRequest, NodeType.OFF_SWITCH,
return assignContainer(node, offswitchAsk, NodeType.OFF_SWITCH,
reserved, schedulerKey);
}
}
@ -1004,30 +1008,35 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
*/
private boolean hasContainerForNode(SchedulerRequestKey key,
FSSchedulerNode node) {
ResourceRequest anyRequest = getResourceRequest(key, ResourceRequest.ANY);
ResourceRequest rackRequest = getResourceRequest(key, node.getRackName());
ResourceRequest nodeRequest = getResourceRequest(key, node.getNodeName());
PendingAsk offswitchAsk = getPendingAsk(key, ResourceRequest.ANY);
Resource resource = offswitchAsk.getPerAllocationResource();
boolean hasRequestForOffswitch =
offswitchAsk.getCount() > 0;
boolean hasRequestForRack = getOutstandingAsksCount(key,
node.getRackName()) > 0;
boolean hasRequestForNode = getOutstandingAsksCount(key,
node.getNodeName()) > 0;
boolean ret = true;
if (!(// There must be outstanding requests at the given priority:
anyRequest != null && anyRequest.getNumContainers() > 0 &&
// If locality relaxation is turned off at *-level, there must be a
// non-zero request for the node's rack:
(anyRequest.getRelaxLocality() ||
(rackRequest != null && rackRequest.getNumContainers() > 0)) &&
// If locality relaxation is turned off at rack-level, there must be a
// non-zero request at the node:
(rackRequest == null || rackRequest.getRelaxLocality() ||
(nodeRequest != null && nodeRequest.getNumContainers() > 0)) &&
// The requested container must be able to fit on the node:
Resources.lessThanOrEqual(RESOURCE_CALCULATOR, null,
anyRequest.getCapability(), node.getRMNode().getTotalCapability()))) {
hasRequestForOffswitch &&
// If locality relaxation is turned off at *-level, there must be a
// non-zero request for the node's rack:
(appSchedulingInfo.canDelayTo(key, ResourceRequest.ANY) ||
(hasRequestForRack)) &&
// If locality relaxation is turned off at rack-level,
// there must be a non-zero request at the node:
(!hasRequestForRack || appSchedulingInfo.canDelayTo(key,
node.getRackName()) || (hasRequestForNode)) &&
// The requested container must be able to fit on the node:
Resources.lessThanOrEqual(RESOURCE_CALCULATOR, null,
resource,
node.getRMNode().getTotalCapability()))) {
ret = false;
} else if (!getQueue().fitsInMaxShare(anyRequest.getCapability())) {
} else if (!getQueue().fitsInMaxShare(resource)) {
// The requested container must fit in queue maximum share
updateAMDiagnosticMsg(anyRequest.getCapability(),
updateAMDiagnosticMsg(resource,
" exceeds current queue or its parents maximum resource allowed).");
ret = false;
}
@ -1117,10 +1126,6 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
scheduler.getClusterResource(), usage, share, true) < 0;
}
ResourceRequest getNextResourceRequest() {
return appSchedulingInfo.getNextResourceRequest();
}
/**
* Helper method that captures if this app is identified to be starved.
* @return true if the app is starved for fairshare, false otherwise
@ -1288,10 +1293,11 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
try {
writeLock.lock();
for (SchedulerRequestKey k : getSchedulerKeys()) {
ResourceRequest r = getResourceRequest(k, ResourceRequest.ANY);
if (r != null) {
Resources.multiplyAndAddTo(demand, r.getCapability(),
r.getNumContainers());
PendingAsk pendingAsk = getPendingAsk(k, ResourceRequest.ANY);
if (pendingAsk.getCount() > 0) {
Resources.multiplyAndAddTo(demand,
pendingAsk.getPerAllocationResource(),
pendingAsk.getCount());
}
}
} finally {
@ -1302,12 +1308,11 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
@Override
public Resource assignContainer(FSSchedulerNode node) {
if (isOverAMShareLimit()) {
Resource amResourceRequest = appSchedulingInfo.getAllResourceRequests()
.get(0).getCapability();
updateAMDiagnosticMsg(amResourceRequest,
PendingAsk amAsk = appSchedulingInfo.getNextPendingAsk();
updateAMDiagnosticMsg(amAsk.getPerAllocationResource(),
" exceeds maximum AM resource allowed).");
if (LOG.isDebugEnabled()) {
LOG.debug("AM resource request: " + amResourceRequest
LOG.debug("AM resource request: " + amAsk.getPerAllocationResource()
+ " exceeds maximum AM resource allowed, "
+ getQueue().dumpState());
}

2
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
View File

@ -26,6 +26,8 @@ import org.apache.hadoop.yarn.api.records.ResourceRequest;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.SchedulingPlacementSet;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.PendingAsk;
import org.apache.hadoop.yarn.util.resource.Resources;
import java.util.ArrayList;

10
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoAppAttempt.java
View File

@ -51,8 +51,7 @@ public class FifoAppAttempt extends FiCaSchedulerApp {
}
public RMContainer allocate(NodeType type, FiCaSchedulerNode node,
SchedulerRequestKey schedulerKey, ResourceRequest request,
Container container) {
SchedulerRequestKey schedulerKey, Container container) {
try {
writeLock.lock();
@ -62,15 +61,14 @@ public class FifoAppAttempt extends FiCaSchedulerApp {
// Required sanity check - AM can call 'allocate' to update resource
// request without locking the scheduler, hence we need to check
if (getTotalRequiredResources(schedulerKey) <= 0) {
if (getOutstandingAsksCount(schedulerKey) <= 0) {
return null;
}
// Create RMContainer
RMContainer rmContainer = new RMContainerImpl(container,
schedulerKey, this.getApplicationAttemptId(), node.getNodeID(),
appSchedulingInfo.getUser(), this.rmContext,
request.getNodeLabelExpression());
appSchedulingInfo.getUser(), this.rmContext, node.getPartition());
((RMContainerImpl) rmContainer).setQueueName(this.getQueueName());
updateAMContainerDiagnostics(AMState.ASSIGNED, null);
@ -83,7 +81,7 @@ public class FifoAppAttempt extends FiCaSchedulerApp {
// Update consumption and track allocations
List<ResourceRequest> resourceRequestList = appSchedulingInfo.allocate(
type, node, schedulerKey, request, container);
type, node, schedulerKey, container);
attemptResourceUsage.incUsed(node.getPartition(),
container.getResource());

102
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java
View File

@ -69,8 +69,8 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedContainerCha
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerAppUtils;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplication;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt;
import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.PendingAsk;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerNode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.AppAddedSchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.AppAttemptAddedSchedulerEvent;
@ -82,6 +82,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeRemoved
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeResourceUpdateSchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeUpdateSchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEvent;
import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
import org.apache.hadoop.yarn.server.utils.BuilderUtils;
import org.apache.hadoop.yarn.server.utils.Lock;
import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator;
@ -544,35 +545,32 @@ public class FifoScheduler extends
private int getMaxAllocatableContainers(FifoAppAttempt application,
SchedulerRequestKey schedulerKey, FiCaSchedulerNode node, NodeType type) {
int maxContainers = 0;
ResourceRequest offSwitchRequest =
application.getResourceRequest(schedulerKey, ResourceRequest.ANY);
if (offSwitchRequest != null) {
maxContainers = offSwitchRequest.getNumContainers();
}
PendingAsk offswitchAsk = application.getPendingAsk(schedulerKey,
ResourceRequest.ANY);
int maxContainers = offswitchAsk.getCount();
if (type == NodeType.OFF_SWITCH) {
return maxContainers;
}
if (type == NodeType.RACK_LOCAL) {
ResourceRequest rackLocalRequest =
application.getResourceRequest(schedulerKey, node.getRMNode()
.getRackName());
if (rackLocalRequest == null) {
PendingAsk rackLocalAsk = application.getPendingAsk(schedulerKey,
node.getRackName());
if (rackLocalAsk.getCount() <= 0) {
return maxContainers;
}
maxContainers = Math.min(maxContainers, rackLocalRequest.getNumContainers());
maxContainers = Math.min(maxContainers,
rackLocalAsk.getCount());
}
if (type == NodeType.NODE_LOCAL) {
ResourceRequest nodeLocalRequest =
application.getResourceRequest(schedulerKey, node.getRMNode()
.getNodeAddress());
if (nodeLocalRequest != null) {
maxContainers = Math.min(maxContainers, nodeLocalRequest.getNumContainers());
PendingAsk nodeLocalAsk = application.getPendingAsk(schedulerKey,
node.getRMNode().getHostName());
if (nodeLocalAsk.getCount() > 0) {
maxContainers = Math.min(maxContainers,
nodeLocalAsk.getCount());
}
}
@ -610,25 +608,21 @@ public class FifoScheduler extends
private int assignNodeLocalContainers(FiCaSchedulerNode node,
FifoAppAttempt application, SchedulerRequestKey schedulerKey) {
int assignedContainers = 0;
ResourceRequest request =
application.getResourceRequest(schedulerKey, node.getNodeName());
if (request != null) {
PendingAsk nodeLocalAsk = application.getPendingAsk(schedulerKey,
node.getNodeName());
if (nodeLocalAsk.getCount() > 0) {
// Don't allocate on this node if we don't need containers on this rack
ResourceRequest rackRequest =
application.getResourceRequest(schedulerKey,
node.getRMNode().getRackName());
if (rackRequest == null || rackRequest.getNumContainers() <= 0) {
if (application.getOutstandingAsksCount(schedulerKey,
node.getRackName()) <= 0) {
return 0;
}
int assignableContainers =
Math.min(
getMaxAllocatableContainers(application, schedulerKey, node,
NodeType.NODE_LOCAL),
request.getNumContainers());
int assignableContainers = Math.min(
getMaxAllocatableContainers(application, schedulerKey, node,
NodeType.NODE_LOCAL), nodeLocalAsk.getCount());
assignedContainers =
assignContainer(node, application, schedulerKey,
assignableContainers, request, NodeType.NODE_LOCAL);
assignContainer(node, application, schedulerKey, assignableContainers,
nodeLocalAsk.getPerAllocationResource(), NodeType.NODE_LOCAL);
}
return assignedContainers;
}
@ -636,25 +630,21 @@ public class FifoScheduler extends
private int assignRackLocalContainers(FiCaSchedulerNode node,
FifoAppAttempt application, SchedulerRequestKey schedulerKey) {
int assignedContainers = 0;
ResourceRequest request =
application.getResourceRequest(schedulerKey, node.getRMNode()
.getRackName());
if (request != null) {
PendingAsk rackAsk = application.getPendingAsk(schedulerKey,
node.getRMNode().getRackName());
if (rackAsk.getCount() > 0) {
// Don't allocate on this rack if the application doens't need containers
ResourceRequest offSwitchRequest =
application.getResourceRequest(schedulerKey, ResourceRequest.ANY);
if (offSwitchRequest.getNumContainers() <= 0) {
if (application.getOutstandingAsksCount(schedulerKey,
ResourceRequest.ANY) <= 0) {
return 0;
}
int assignableContainers =
Math.min(
getMaxAllocatableContainers(application, schedulerKey, node,
NodeType.RACK_LOCAL),
request.getNumContainers());
int assignableContainers =
Math.min(getMaxAllocatableContainers(application, schedulerKey, node,
NodeType.RACK_LOCAL), rackAsk.getCount());
assignedContainers =
assignContainer(node, application, schedulerKey,
assignableContainers, request, NodeType.RACK_LOCAL);
assignContainer(node, application, schedulerKey, assignableContainers,
rackAsk.getPerAllocationResource(), NodeType.RACK_LOCAL);
}
return assignedContainers;
}
@ -662,26 +652,26 @@ public class FifoScheduler extends
private int assignOffSwitchContainers(FiCaSchedulerNode node,
FifoAppAttempt application, SchedulerRequestKey schedulerKey) {
int assignedContainers = 0;
ResourceRequest request =
application.getResourceRequest(schedulerKey, ResourceRequest.ANY);
if (request != null) {
PendingAsk offswitchAsk = application.getPendingAsk(schedulerKey,
ResourceRequest.ANY);
if (offswitchAsk.getCount() > 0) {
assignedContainers =
assignContainer(node, application, schedulerKey,
request.getNumContainers(), request, NodeType.OFF_SWITCH);
offswitchAsk.getCount(),
offswitchAsk.getPerAllocationResource(), NodeType.OFF_SWITCH);
}
return assignedContainers;
}
private int assignContainer(FiCaSchedulerNode node, FifoAppAttempt application,
SchedulerRequestKey schedulerKey, int assignableContainers,
ResourceRequest request, NodeType type) {
Resource capability, NodeType type) {
LOG.debug("assignContainers:" +
" node=" + node.getRMNode().getNodeAddress() +
" application=" + application.getApplicationId().getId() +
" priority=" + schedulerKey.getPriority().getPriority() +
" assignableContainers=" + assignableContainers +
" request=" + request + " type=" + type);
Resource capability = request.getCapability();
" capability=" + capability + " type=" + type);
// TODO: A buggy application with this zero would crash the scheduler.
int availableContainers =
@ -707,7 +697,7 @@ public class FifoScheduler extends
// Inform the application
RMContainer rmContainer = application.allocate(type, node, schedulerKey,
request, container);
container);
// Inform the node
node.allocateContainer(rmContainer);

129
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/LocalitySchedulingPlacementSet.java
View File

@ -19,12 +19,16 @@
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement;
import org.apache.commons.collections.IteratorUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AppSchedulingInfo;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.SchedulingMode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.PendingAsk;
import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
import java.util.ArrayList;
@ -37,9 +41,14 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
public class LocalitySchedulingPlacementSet<N extends SchedulerNode>
implements SchedulingPlacementSet<N> {
private static final Log LOG =
LogFactory.getLog(LocalitySchedulingPlacementSet.class);
private final Map<String, ResourceRequest> resourceRequestMap =
new ConcurrentHashMap<>();
private AppSchedulingInfo appSchedulingInfo;
private volatile String primaryRequestedPartition =
RMNodeLabelsManager.NO_LABEL;
private final ReentrantReadWriteLock.ReadLock readLock;
private final ReentrantReadWriteLock.WriteLock writeLock;
@ -132,11 +141,14 @@ public class LocalitySchedulingPlacementSet<N extends SchedulerNode>
resourceRequestMap.put(resourceName, request);
if (resourceName.equals(ResourceRequest.ANY)) {
String partition = request.getNodeLabelExpression() == null ?
RMNodeLabelsManager.NO_LABEL :
request.getNodeLabelExpression();
this.primaryRequestedPartition = partition;
//update the applications requested labels set
appSchedulingInfo.addRequestedPartition(
request.getNodeLabelExpression() == null ?
RMNodeLabelsManager.NO_LABEL :
request.getNodeLabelExpression());
appSchedulingInfo.addRequestedPartition(partition);
updateResult = new ResourceRequestUpdateResult(lastRequest, request);
}
@ -152,11 +164,43 @@ public class LocalitySchedulingPlacementSet<N extends SchedulerNode>
return resourceRequestMap;
}
@Override
public ResourceRequest getResourceRequest(String resourceName) {
private ResourceRequest getResourceRequest(String resourceName) {
return resourceRequestMap.get(resourceName);
}
@Override
public PendingAsk getPendingAsk(String resourceName) {
try {
readLock.lock();
ResourceRequest request = getResourceRequest(resourceName);
if (null == request) {
return PendingAsk.ZERO;
} else{
return new PendingAsk(request.getCapability(),
request.getNumContainers());
}
} finally {
readLock.unlock();
}
}
@Override
public int getOutstandingAsksCount(String resourceName) {
try {
readLock.lock();
ResourceRequest request = getResourceRequest(resourceName);
if (null == request) {
return 0;
} else{
return request.getNumContainers();
}
} finally {
readLock.unlock();
}
}
private void decrementOutstanding(SchedulerRequestKey schedulerRequestKey,
ResourceRequest offSwitchRequest) {
int numOffSwitchContainers = offSwitchRequest.getNumContainers() - 1;
@ -281,22 +325,67 @@ public class LocalitySchedulingPlacementSet<N extends SchedulerNode>
}
}
@Override
public boolean canDelayTo(String resourceName) {
try {
readLock.lock();
ResourceRequest request = getResourceRequest(resourceName);
return request == null || request.getRelaxLocality();
} finally {
readLock.unlock();
}
}
@Override
public boolean acceptNodePartition(String nodePartition,
SchedulingMode schedulingMode) {
// We will only look at node label = nodeLabelToLookAt according to
// schedulingMode and partition of node.
String nodePartitionToLookAt;
if (schedulingMode == SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY) {
nodePartitionToLookAt = nodePartition;
} else {
nodePartitionToLookAt = RMNodeLabelsManager.NO_LABEL;
}
return primaryRequestedPartition.equals(nodePartitionToLookAt);
}
@Override
public String getPrimaryRequestedNodePartition() {
return primaryRequestedPartition;
}
@Override
public int getUniqueLocationAsks() {
return resourceRequestMap.size();
}
@Override
public void showRequests() {
for (ResourceRequest request : resourceRequestMap.values()) {
if (request.getNumContainers() > 0) {
LOG.debug("\tRequest=" + request);
}
}
}
@Override
public List<ResourceRequest> allocate(SchedulerRequestKey schedulerKey,
NodeType type, SchedulerNode node, ResourceRequest request) {
NodeType type, SchedulerNode node) {
try {
writeLock.lock();
List<ResourceRequest> resourceRequests = new ArrayList<>();
if (null == request) {
if (type == NodeType.NODE_LOCAL) {
request = resourceRequestMap.get(node.getNodeName());
} else if (type == NodeType.RACK_LOCAL) {
request = resourceRequestMap.get(node.getRackName());
} else{
request = resourceRequestMap.get(ResourceRequest.ANY);
}
ResourceRequest request;
if (type == NodeType.NODE_LOCAL) {
request = resourceRequestMap.get(node.getNodeName());
} else if (type == NodeType.RACK_LOCAL) {
request = resourceRequestMap.get(node.getRackName());
} else{
request = resourceRequestMap.get(ResourceRequest.ANY);
}
if (type == NodeType.NODE_LOCAL) {
@ -312,4 +401,14 @@ public class LocalitySchedulingPlacementSet<N extends SchedulerNode>
writeLock.unlock();
}
}
@Override
public Iterator<String> getAcceptedResouceNames() {
try {
readLock.lock();
return resourceRequestMap.keySet().iterator();
} finally {
readLock.unlock();
}
}
}

71
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/SchedulingPlacementSet.java
View File

@ -21,6 +21,8 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.SchedulingMode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.PendingAsk;
import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
import java.util.Collection;
@ -70,22 +72,38 @@ public interface SchedulingPlacementSet<N extends SchedulerNode> {
Map<String, ResourceRequest> getResourceRequests();
/**
* Get ResourceRequest by given schedulerKey and resourceName
* Get pending ask for given resourceName. If there's no such pendingAsk,
* returns {@link PendingAsk#ZERO}
*
* @param resourceName resourceName
* @return ResourceRequest
* @return PendingAsk
*/
ResourceRequest getResourceRequest(String resourceName);
PendingAsk getPendingAsk(String resourceName);
/**
* Get #pending-allocations for given resourceName. If there's no such
* pendingAsk, returns 0
*
* @param resourceName resourceName
* @return #pending-allocations
*/
int getOutstandingAsksCount(String resourceName);
/**
* Notify container allocated.
* @param schedulerKey SchedulerRequestKey for this ResourceRequest
* @param type Type of the allocation
* @param node Which node this container allocated on
* @param request Which resource request to allocate
* @return list of ResourceRequests deducted
*/
List<ResourceRequest> allocate(SchedulerRequestKey schedulerKey,
NodeType type, SchedulerNode node, ResourceRequest request);
NodeType type, SchedulerNode node);
/**
* Returns list of accepted resourceNames.
* @return Iterator of accepted resourceNames
*/
Iterator<String> getAcceptedResouceNames();
/**
* We can still have pending requirement for a given NodeType and node
@ -94,4 +112,47 @@ public interface SchedulingPlacementSet<N extends SchedulerNode> {
* @return true if we has pending requirement
*/
boolean canAllocate(NodeType type, SchedulerNode node);
/**
* Can delay to give locality?
* TODO (wangda): This should be moved out of SchedulingPlacementSet
* and should belong to specific delay scheduling policy impl.
*
* @param resourceName resourceName
* @return can/cannot
*/
boolean canDelayTo(String resourceName);
/**
* Does this {@link SchedulingPlacementSet} accept resources on nodePartition?
*
* @param nodePartition nodePartition
* @param schedulingMode schedulingMode
* @return accepted/not
*/
boolean acceptNodePartition(String nodePartition,
SchedulingMode schedulingMode);
/**
* It is possible that one request can accept multiple node partition,
* So this method returns primary node partition for pending resource /
* headroom calculation.
*
* @return primary requested node partition
*/
String getPrimaryRequestedNodePartition();
/**
* @return number of unique location asks with #pending greater than 0,
* (like /rack1, host1, etc.).
*
* TODO (wangda): This should be moved out of SchedulingPlacementSet
* and should belong to specific delay scheduling policy impl.
*/
int getUniqueLocationAsks();
/**
* Print human-readable requests to LOG debug.
*/
void showRequests();
}

15
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/TestAbstractYarnScheduler.java
View File

@ -75,6 +75,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEv
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEventType;
import org.apache.hadoop.yarn.server.resourcemanager.security.NMTokenSecretManagerInRM;
import org.apache.hadoop.yarn.server.resourcemanager.security.RMContainerTokenSecretManager;
import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
import org.apache.hadoop.yarn.server.utils.BuilderUtils;
import org.apache.hadoop.yarn.util.resource.Resources;
import org.junit.Assert;
@ -702,12 +703,14 @@ public class TestAbstractYarnScheduler extends ParameterizedSchedulerTestBase {
// The core part of this test
// The killed containers' ResourceRequests are recovered back to the
// original app-attempt, not the new one
for (ResourceRequest request : firstSchedulerAppAttempt
.getAppSchedulingInfo().getAllResourceRequests()) {
if (request.getPriority().getPriority() == 0) {
Assert.assertEquals(0, request.getNumContainers());
} else if (request.getPriority().getPriority() == ALLOCATED_CONTAINER_PRIORITY) {
Assert.assertEquals(1, request.getNumContainers());
for (SchedulerRequestKey key : firstSchedulerAppAttempt.getSchedulerKeys()) {
if (key.getPriority().getPriority() == 0) {
Assert.assertEquals(0,
firstSchedulerAppAttempt.getOutstandingAsksCount(key));
} else if (key.getPriority().getPriority() ==
ALLOCATED_CONTAINER_PRIORITY) {
Assert.assertEquals(1,
firstSchedulerAppAttempt.getOutstandingAsksCount(key));
}
}

4
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/TestAppSchedulingInfo.java
View File

@ -141,7 +141,7 @@ public class TestAppSchedulingInfo {
// iterate to verify no ConcurrentModificationException
for (SchedulerRequestKey schedulerKey : info.getSchedulerKeys()) {
info.allocate(NodeType.OFF_SWITCH, null, schedulerKey, req1, null);
info.allocate(NodeType.OFF_SWITCH, null, schedulerKey, null);
}
Assert.assertEquals(1, info.getSchedulerKeys().size());
Assert.assertEquals(SchedulerRequestKey.create(req2),
@ -153,7 +153,7 @@ public class TestAppSchedulingInfo {
reqs.add(req2);
info.updateResourceRequests(reqs, false);
info.allocate(NodeType.OFF_SWITCH, null, SchedulerRequestKey.create(req2),
req2, null);
null);
Assert.assertEquals(0, info.getSchedulerKeys().size());
req1 = ResourceRequest.newInstance(pri1,

3
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/TestSchedulerApplicationAttempt.java
View File

@ -93,8 +93,7 @@ public class TestSchedulerApplicationAttempt {
app.liveContainers.put(container1.getContainerId(), container1);
SchedulerNode node = createNode();
app.appSchedulingInfo.allocate(NodeType.OFF_SWITCH, node,
toSchedulerKey(requestedPriority),
request, container1.getContainer());
toSchedulerKey(requestedPriority), container1.getContainer());
// Reserved container
Priority prio1 = Priority.newInstance(1);

234
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java
View File

@ -188,7 +188,7 @@ public class TestCapacityScheduler {
private ResourceManager resourceManager = null;
private RMContext mockContext;
@Before
public void setUp() throws Exception {
resourceManager = new ResourceManager() {
@ -199,11 +199,11 @@ public class TestCapacityScheduler {
return mgr;
}
};
CapacitySchedulerConfiguration csConf
CapacitySchedulerConfiguration csConf
= new CapacitySchedulerConfiguration();
setupQueueConfiguration(csConf);
YarnConfiguration conf = new YarnConfiguration(csConf);
conf.setClass(YarnConfiguration.RM_SCHEDULER,
conf.setClass(YarnConfiguration.RM_SCHEDULER,
CapacityScheduler.class, ResourceScheduler.class);
resourceManager.init(conf);
resourceManager.getRMContext().getContainerTokenSecretManager().rollMasterKey();
@ -263,7 +263,7 @@ public class TestCapacityScheduler {
new org.apache.hadoop.yarn.server.resourcemanager.NodeManager(
hostName, containerManagerPort, httpPort, rackName, capability,
resourceManager);
NodeAddedSchedulerEvent nodeAddEvent1 =
NodeAddedSchedulerEvent nodeAddEvent1 =
new NodeAddedSchedulerEvent(resourceManager.getRMContext()
.getRMNodes().get(nm.getNodeId()));
resourceManager.getResourceScheduler().handle(nodeAddEvent1);
@ -274,89 +274,89 @@ public class TestCapacityScheduler {
public void testCapacityScheduler() throws Exception {
LOG.info("--- START: testCapacityScheduler ---");
// Register node1
String host_0 = "host_0";
org.apache.hadoop.yarn.server.resourcemanager.NodeManager nm_0 =
registerNode(host_0, 1234, 2345, NetworkTopology.DEFAULT_RACK,
Resources.createResource(4 * GB, 1));
org.apache.hadoop.yarn.server.resourcemanager.NodeManager nm_0 =
registerNode(host_0, 1234, 2345, NetworkTopology.DEFAULT_RACK,
Resources.createResource(4 * GB, 1));
// Register node2
String host_1 = "host_1";
org.apache.hadoop.yarn.server.resourcemanager.NodeManager nm_1 =
registerNode(host_1, 1234, 2345, NetworkTopology.DEFAULT_RACK,
Resources.createResource(2 * GB, 1));
org.apache.hadoop.yarn.server.resourcemanager.NodeManager nm_1 =
registerNode(host_1, 1234, 2345, NetworkTopology.DEFAULT_RACK,
Resources.createResource(2 * GB, 1));
// ResourceRequest priorities
Priority priority_0 = Priority.newInstance(0);
Priority priority_1 = Priority.newInstance(1);
// Submit an application
Application application_0 = new Application("user_0", "a1", resourceManager);
application_0.submit();
application_0.addNodeManager(host_0, 1234, nm_0);
application_0.addNodeManager(host_1, 1234, nm_1);
Resource capability_0_0 = Resources.createResource(1 * GB, 1);
application_0.addResourceRequestSpec(priority_1, capability_0_0);
Resource capability_0_1 = Resources.createResource(2 * GB, 1);
application_0.addResourceRequestSpec(priority_0, capability_0_1);
Task task_0_0 = new Task(application_0, priority_1,
Task task_0_0 = new Task(application_0, priority_1,
new String[] {host_0, host_1});
application_0.addTask(task_0_0);
// Submit another application
Application application_1 = new Application("user_1", "b2", resourceManager);
application_1.submit();
application_1.addNodeManager(host_0, 1234, nm_0);
application_1.addNodeManager(host_1, 1234, nm_1);
Resource capability_1_0 = Resources.createResource(3 * GB, 1);
application_1.addResourceRequestSpec(priority_1, capability_1_0);
Resource capability_1_1 = Resources.createResource(2 * GB, 1);
application_1.addResourceRequestSpec(priority_0, capability_1_1);
Task task_1_0 = new Task(application_1, priority_1,
Task task_1_0 = new Task(application_1, priority_1,
new String[] {host_0, host_1});
application_1.addTask(task_1_0);
// Send resource requests to the scheduler
application_0.schedule();
application_1.schedule();
// Send a heartbeat to kick the tires on the Scheduler
LOG.info("Kick!");
// task_0_0 and task_1_0 allocated, used=4G
nodeUpdate(nm_0);
// nothing allocated
nodeUpdate(nm_1);
// Get allocations from the scheduler
application_0.schedule(); // task_0_0
checkApplicationResourceUsage(1 * GB, application_0);
application_1.schedule(); // task_1_0
checkApplicationResourceUsage(3 * GB, application_1);
checkNodeResourceUsage(4*GB, nm_0); // task_0_0 (1G) and task_1_0 (3G)
checkNodeResourceUsage(0*GB, nm_1); // no tasks, 2G available
LOG.info("Adding new tasks...");
Task task_1_1 = new Task(application_1, priority_0,
Task task_1_1 = new Task(application_1, priority_0,
new String[] {ResourceRequest.ANY});
application_1.addTask(task_1_1);
application_1.schedule();
Task task_0_1 = new Task(application_0, priority_0,
Task task_0_1 = new Task(application_0, priority_0,
new String[] {host_0, host_1});
application_0.addTask(task_0_1);
@ -366,11 +366,11 @@ public class TestCapacityScheduler {
LOG.info("Sending hb from " + nm_0.getHostName());
// nothing new, used=4G
nodeUpdate(nm_0);
LOG.info("Sending hb from " + nm_1.getHostName());
// task_0_1 is prefer as locality, used=2G
nodeUpdate(nm_1);
// Get allocations from the scheduler
LOG.info("Trying to allocate...");
application_0.schedule();
@ -378,10 +378,10 @@ public class TestCapacityScheduler {
application_1.schedule();
checkApplicationResourceUsage(5 * GB, application_1);
nodeUpdate(nm_0);
nodeUpdate(nm_1);
checkNodeResourceUsage(4*GB, nm_0);
checkNodeResourceUsage(2*GB, nm_1);
@ -408,20 +408,20 @@ public class TestCapacityScheduler {
*/
private CapacitySchedulerConfiguration setupQueueConfiguration(
CapacitySchedulerConfiguration conf) {
// Define top-level queues
conf.setQueues(CapacitySchedulerConfiguration.ROOT, new String[] {"a", "b"});
conf.setCapacity(A, A_CAPACITY);
conf.setCapacity(B, B_CAPACITY);
// Define 2nd-level queues
conf.setQueues(A, new String[] {"a1", "a2"});
conf.setCapacity(A1, A1_CAPACITY);
conf.setUserLimitFactor(A1, 100.0f);
conf.setCapacity(A2, A2_CAPACITY);
conf.setUserLimitFactor(A2, 100.0f);
conf.setQueues(B, new String[] {"b1", "b2", "b3"});
conf.setCapacity(B1, B1_CAPACITY);
conf.setUserLimitFactor(B1, 100.0f);
@ -596,8 +596,8 @@ public class TestCapacityScheduler {
conf.setMaximumCapacity(A, -1);
assertEquals(CapacitySchedulerConfiguration.MAXIMUM_CAPACITY_VALUE,conf.getNonLabeledQueueMaximumCapacity(A),delta);
}
@Test
public void testRefreshQueues() throws Exception {
CapacityScheduler cs = new CapacityScheduler();
@ -682,11 +682,11 @@ public class TestCapacityScheduler {
return null;
}
private void checkApplicationResourceUsage(int expected,
private void checkApplicationResourceUsage(int expected,
Application application) {
Assert.assertEquals(expected, application.getUsedResources().getMemorySize());
}
private void checkNodeResourceUsage(int expected,
org.apache.hadoop.yarn.server.resourcemanager.NodeManager node) {
Assert.assertEquals(expected, node.getUsed().getMemorySize());
@ -767,7 +767,7 @@ public class TestCapacityScheduler {
// Add a new queue b4
String B4 = B + ".b4";
float B4_CAPACITY = 10;
B3_CAPACITY -= B4_CAPACITY;
try {
conf.setCapacity(A, 80f);
@ -779,7 +779,7 @@ public class TestCapacityScheduler {
conf.setCapacity(B4, B4_CAPACITY);
cs.reinitialize(conf,mockContext);
checkQueueCapacities(cs, 80f, 20f);
// Verify parent for B4
CSQueue rootQueue = cs.getRootQueue();
CSQueue queueB = findQueue(rootQueue, B);
@ -997,7 +997,7 @@ public class TestCapacityScheduler {
ResourceScheduler.class);
MockRM rm = new MockRM(conf);
rm.start();
MockNM nm1 = rm.registerNode("127.0.0.1:1234", 4 * GB);
RMApp app1 = rm.submitApp(2048);
// kick the scheduling, 2 GB given to AM1, remaining 2GB on nm1
@ -1027,7 +1027,7 @@ public class TestCapacityScheduler {
Assert.assertEquals(1, allocated1.size());
Assert.assertEquals(2 * GB, allocated1.get(0).getResource().getMemorySize());
Assert.assertEquals(nm1.getNodeId(), allocated1.get(0).getNodeId());
report_nm1 = rm.getResourceScheduler().getNodeReport(nm1.getNodeId());
// check node report, 4 GB used and 0 GB available
Assert.assertEquals(0, report_nm1.getAvailableResource().getMemorySize());
@ -1036,13 +1036,13 @@ public class TestCapacityScheduler {
// check container is assigned with 2 GB.
Container c1 = allocated1.get(0);
Assert.assertEquals(2 * GB, c1.getResource().getMemorySize());
// update node resource to 2 GB, so resource is over-consumed.
Map<NodeId, ResourceOption> nodeResourceMap =
Map<NodeId, ResourceOption> nodeResourceMap =
new HashMap<NodeId, ResourceOption>();
nodeResourceMap.put(nm1.getNodeId(),
nodeResourceMap.put(nm1.getNodeId(),
ResourceOption.newInstance(Resource.newInstance(2 * GB, 1), -1));
UpdateNodeResourceRequest request =
UpdateNodeResourceRequest request =
UpdateNodeResourceRequest.newInstance(nodeResourceMap);
AdminService as = ((MockRM)rm).getAdminService();
as.updateNodeResource(request);
@ -1061,7 +1061,7 @@ public class TestCapacityScheduler {
report_nm1 = rm.getResourceScheduler().getNodeReport(nm1.getNodeId());
Assert.assertEquals(4 * GB, report_nm1.getUsedResource().getMemorySize());
Assert.assertEquals(-2 * GB, report_nm1.getAvailableResource().getMemorySize());
// Check container can complete successfully in case of resource over-commitment.
ContainerStatus containerStatus = BuilderUtils.newContainerStatus(
c1.getId(), ContainerState.COMPLETE, "", 0, c1.getResource());
@ -1079,7 +1079,7 @@ public class TestCapacityScheduler {
Assert.assertEquals(2 * GB, report_nm1.getUsedResource().getMemorySize());
// As container return 2 GB back, the available resource becomes 0 again.
Assert.assertEquals(0 * GB, report_nm1.getAvailableResource().getMemorySize());
// Verify no NPE is trigger in schedule after resource is updated.
am1.addRequests(new String[] { "127.0.0.1", "127.0.0.2" }, 3 * GB, 1, 1);
alloc1Response = am1.schedule();
@ -1097,7 +1097,7 @@ public class TestCapacityScheduler {
0, alloc1Response.getAllocatedContainers().size());
rm.stop();
}
@Test
public void testGetAppsInQueue() throws Exception {
Application application_0 = new Application("user_0", "a1", resourceManager);
@ -1145,7 +1145,7 @@ public class TestCapacityScheduler {
cs.getSchedulerApplications(), cs, "a1");
Assert.assertEquals("a1", app.getQueue().getQueueName());
}
@Test
public void testAsyncScheduling() throws Exception {
Configuration conf = new Configuration();
@ -1156,7 +1156,7 @@ public class TestCapacityScheduler {
CapacityScheduler cs = (CapacityScheduler) rm.getResourceScheduler();
final int NODES = 100;
// Register nodes
for (int i=0; i < NODES; ++i) {
String host = "192.168.1." + i;
@ -1164,7 +1164,7 @@ public class TestCapacityScheduler {
MockNodes.newNodeInfo(0, MockNodes.newResource(4 * GB), 1, host);
cs.handle(new NodeAddedSchedulerEvent(node));
}
// Now directly exercise the scheduling loop
for (int i=0; i < NODES; ++i) {
CapacityScheduler.schedule(cs);
@ -1186,7 +1186,7 @@ public class TestCapacityScheduler {
&& attemptPM.getResourcePreempted().equals(currentAttemptPreempted)
&& app.getCurrentAppAttempt().getRMAppAttemptMetrics()
.getIsPreempted() == currentAttemptAMPreempted
&& attemptPM.getNumNonAMContainersPreempted() ==
&& attemptPM.getNumNonAMContainersPreempted() ==
numLatestAttemptTaskPreempted) {
return;
}
@ -1200,7 +1200,7 @@ public class TestCapacityScheduler {
Thread.sleep(500);
}
}
@Test(timeout = 30000)
public void testAllocateDoesNotBlockOnSchedulerLock() throws Exception {
final YarnConfiguration conf = new YarnConfiguration();
@ -1419,7 +1419,7 @@ public class TestCapacityScheduler {
rm1.stop();
}
@Test(timeout = 300000)
public void testRecoverRequestAfterPreemption() throws Exception {
Configuration conf = new Configuration();
@ -1453,8 +1453,9 @@ public class TestCapacityScheduler {
// Already the node local resource request is cleared from RM after
// allocation.
Assert.assertNull(app.getResourceRequest(
SchedulerRequestKey.create(request), request.getResourceName()));
Assert.assertEquals(0,
app.getOutstandingAsksCount(SchedulerRequestKey.create(request),
request.getResourceName()));
}
// Call killContainer to preempt the container
@ -1464,10 +1465,9 @@ public class TestCapacityScheduler {
for (ResourceRequest request : requests) {
// Resource request must have added back in RM after preempt event
// handling.
Assert.assertEquals(
1,
app.getResourceRequest(SchedulerRequestKey.create(request),
request.getResourceName()).getNumContainers());
Assert.assertEquals(1,
app.getOutstandingAsksCount(SchedulerRequestKey.create(request),
request.getResourceName()));
}
// New container will be allocated and will move to ALLOCATED state
@ -2932,7 +2932,7 @@ public class TestCapacityScheduler {
assertEquals("queue B2 max vcores allocation", 12,
((LeafQueue) queueB2).getMaximumAllocation().getVirtualCores());
}
private void waitContainerAllocated(MockAM am, int mem, int nContainer,
int startContainerId, MockRM rm, MockNM nm) throws Exception {
for (int cId = startContainerId; cId < startContainerId + nContainer; cId++) {
@ -2966,44 +2966,44 @@ public class TestCapacityScheduler {
MockNM nm1 =
new MockNM("127.0.0.1:1234", 100 * GB, rm1.getResourceTrackerService());
nm1.registerNode();
RMApp app1 = rm1.submitApp(1 * GB, "app", "user", null, "b1");
MockAM am1 = MockRM.launchAndRegisterAM(app1, rm1, nm1);
waitContainerAllocated(am1, 1 * GB, 1, 2, rm1, nm1);
// Maximum resoure of b1 is 100 * 0.895 * 0.792 = 71 GB
// 2 GBs used by am, so it's 71 - 2 = 69G.
Assert.assertEquals(69 * GB,
am1.doHeartbeat().getAvailableResources().getMemorySize());
RMApp app2 = rm1.submitApp(1 * GB, "app", "user", null, "b2");
MockAM am2 = MockRM.launchAndRegisterAM(app2, rm1, nm1);
// Allocate 5 containers, each one is 8 GB in am2 (40 GB in total)
waitContainerAllocated(am2, 8 * GB, 5, 2, rm1, nm1);
// Allocated one more container with 1 GB resource in b1
waitContainerAllocated(am1, 1 * GB, 1, 3, rm1, nm1);
// Total is 100 GB,
// B2 uses 41 GB (5 * 8GB containers and 1 AM container)
// B1 uses 3 GB (2 * 1GB containers and 1 AM container)
// Available is 100 - 41 - 3 = 56 GB
Assert.assertEquals(56 * GB,
am1.doHeartbeat().getAvailableResources().getMemorySize());
// Now we submit app3 to a1 (in higher level hierarchy), to see if headroom
// of app1 (in queue b1) updated correctly
RMApp app3 = rm1.submitApp(1 * GB, "app", "user", null, "a1");
MockAM am3 = MockRM.launchAndRegisterAM(app3, rm1, nm1);
// Allocate 3 containers, each one is 8 GB in am3 (24 GB in total)
waitContainerAllocated(am3, 8 * GB, 3, 2, rm1, nm1);
// Allocated one more container with 4 GB resource in b1
waitContainerAllocated(am1, 1 * GB, 1, 4, rm1, nm1);
// Total is 100 GB,
// B2 uses 41 GB (5 * 8GB containers and 1 AM container)
// B1 uses 4 GB (3 * 1GB containers and 1 AM container)
@ -3012,7 +3012,7 @@ public class TestCapacityScheduler {
Assert.assertEquals(30 * GB,
am1.doHeartbeat().getAvailableResources().getMemorySize());
}
@Test
public void testParentQueueMaxCapsAreRespected() throws Exception {
/*
@ -3028,7 +3028,7 @@ public class TestCapacityScheduler {
csConf.setCapacity(A, 50);
csConf.setMaximumCapacity(A, 50);
csConf.setCapacity(B, 50);
// Define 2nd-level queues
csConf.setQueues(A, new String[] {"a1", "a2"});
csConf.setCapacity(A1, 50);
@ -3037,7 +3037,7 @@ public class TestCapacityScheduler {
csConf.setUserLimitFactor(A2, 100.0f);
csConf.setCapacity(B1, B1_CAPACITY);
csConf.setUserLimitFactor(B1, 100.0f);
YarnConfiguration conf = new YarnConfiguration(csConf);
conf.setBoolean(CapacitySchedulerConfiguration.ENABLE_USER_METRICS, true);
@ -3048,12 +3048,12 @@ public class TestCapacityScheduler {
MockNM nm1 =
new MockNM("127.0.0.1:1234", 24 * GB, rm1.getResourceTrackerService());
nm1.registerNode();
// Launch app1 in a1, resource usage is 1GB (am) + 4GB * 2 = 9GB
RMApp app1 = rm1.submitApp(1 * GB, "app", "user", null, "a1");
MockAM am1 = MockRM.launchAndRegisterAM(app1, rm1, nm1);
waitContainerAllocated(am1, 4 * GB, 2, 2, rm1, nm1);
// Try to launch app2 in a2, asked 2GB, should success
RMApp app2 = rm1.submitApp(2 * GB, "app", "user", null, "a2");
MockAM am2 = MockRM.launchAndRegisterAM(app2, rm1, nm1);
@ -3070,24 +3070,24 @@ public class TestCapacityScheduler {
Assert.fail("Shouldn't successfully allocate containers for am2, "
+ "queue-a's max capacity will be violated if container allocated");
}
@SuppressWarnings("unchecked")
private <E> Set<E> toSet(E... elements) {
Set<E> set = Sets.newHashSet(elements);
return set;
}
@Test
public void testQueueHierarchyPendingResourceUpdate() throws Exception {
Configuration conf =
TestUtils.getConfigurationWithQueueLabels(new Configuration(false));
conf.setBoolean(YarnConfiguration.NODE_LABELS_ENABLED, true);
final RMNodeLabelsManager mgr = new NullRMNodeLabelsManager();
mgr.init(conf);
mgr.addToCluserNodeLabelsWithDefaultExclusivity(ImmutableSet.of("x", "y"));
mgr.addLabelsToNode(ImmutableMap.of(NodeId.newInstance("h1", 0), toSet("x")));
MemoryRMStateStore memStore = new MemoryRMStateStore();
memStore.init(conf);
MockRM rm = new MockRM(conf, memStore) {
@ -3095,74 +3095,74 @@ public class TestCapacityScheduler {
return mgr;
}
};
rm.start();
MockNM nm1 = // label = x
new MockNM("h1:1234", 200 * GB, rm.getResourceTrackerService());
nm1.registerNode();
MockNM nm2 = // label = ""
new MockNM("h2:1234", 200 * GB, rm.getResourceTrackerService());
nm2.registerNode();
// Launch app1 in queue=a1
RMApp app1 = rm.submitApp(1 * GB, "app", "user", null, "a1");
MockAM am1 = MockRM.launchAndRegisterAM(app1, rm, nm2);
// Launch app2 in queue=b1
RMApp app2 = rm.submitApp(8 * GB, "app", "user", null, "b1");
MockAM am2 = MockRM.launchAndRegisterAM(app2, rm, nm2);
// am1 asks for 8 * 1GB container for no label
am1.allocate(Arrays.asList(ResourceRequest.newInstance(
Priority.newInstance(1), "*", Resources.createResource(1 * GB), 8)),
null);
checkPendingResource(rm, "a1", 8 * GB, null);
checkPendingResource(rm, "a", 8 * GB, null);
checkPendingResource(rm, "root", 8 * GB, null);
// am2 asks for 8 * 1GB container for no label
am2.allocate(Arrays.asList(ResourceRequest.newInstance(
Priority.newInstance(1), "*", Resources.createResource(1 * GB), 8)),
null);
checkPendingResource(rm, "a1", 8 * GB, null);
checkPendingResource(rm, "a", 8 * GB, null);
checkPendingResource(rm, "b1", 8 * GB, null);
checkPendingResource(rm, "b", 8 * GB, null);
// root = a + b
checkPendingResource(rm, "root", 16 * GB, null);
// am2 asks for 8 * 1GB container in another priority for no label
am2.allocate(Arrays.asList(ResourceRequest.newInstance(
Priority.newInstance(2), "*", Resources.createResource(1 * GB), 8)),
null);
checkPendingResource(rm, "a1", 8 * GB, null);
checkPendingResource(rm, "a", 8 * GB, null);
checkPendingResource(rm, "b1", 16 * GB, null);
checkPendingResource(rm, "b", 16 * GB, null);
// root = a + b
checkPendingResource(rm, "root", 24 * GB, null);
// am1 asks 4 GB resource instead of 8 * GB for priority=1
am1.allocate(Arrays.asList(ResourceRequest.newInstance(
Priority.newInstance(1), "*", Resources.createResource(4 * GB), 1)),
null);
checkPendingResource(rm, "a1", 4 * GB, null);
checkPendingResource(rm, "a", 4 * GB, null);
checkPendingResource(rm, "b1", 16 * GB, null);
checkPendingResource(rm, "b", 16 * GB, null);
// root = a + b
checkPendingResource(rm, "root", 20 * GB, null);
// am1 asks 8 * GB resource which label=x
am1.allocate(Arrays.asList(ResourceRequest.newInstance(
Priority.newInstance(2), "*", Resources.createResource(8 * GB), 1,
true, "x")), null);
checkPendingResource(rm, "a1", 4 * GB, null);
checkPendingResource(rm, "a", 4 * GB, null);
checkPendingResource(rm, "a1", 8 * GB, "x");
@ -3172,7 +3172,7 @@ public class TestCapacityScheduler {
// root = a + b
checkPendingResource(rm, "root", 20 * GB, null);
checkPendingResource(rm, "root", 8 * GB, "x");
// some containers allocated for am1, pending resource should decrease
ContainerId containerId =
ContainerId.newContainerId(am1.getApplicationAttemptId(), 2);
@ -3181,7 +3181,7 @@ public class TestCapacityScheduler {
containerId = ContainerId.newContainerId(am1.getApplicationAttemptId(), 3);
Assert.assertTrue(rm.waitForState(nm2, containerId,
RMContainerState.ALLOCATED));
checkPendingResource(rm, "a1", 0 * GB, null);
checkPendingResource(rm, "a", 0 * GB, null);
checkPendingResource(rm, "a1", 0 * GB, "x");
@ -3193,23 +3193,23 @@ public class TestCapacityScheduler {
// root = a + b
checkPendingResourceGreaterThanZero(rm, "root", null);
checkPendingResource(rm, "root", 0 * GB, "x");
// complete am2, pending resource should be 0 now
AppAttemptRemovedSchedulerEvent appRemovedEvent =
new AppAttemptRemovedSchedulerEvent(
am2.getApplicationAttemptId(), RMAppAttemptState.FINISHED, false);
rm.getResourceScheduler().handle(appRemovedEvent);
checkPendingResource(rm, "a1", 0 * GB, null);
checkPendingResource(rm, "a", 0 * GB, null);
checkPendingResource(rm, "a1", 0 * GB, "x");
checkPendingResource(rm, "a", 0 * GB, "x");
checkPendingResource(rm, "a", 0 * GB, "x");
checkPendingResource(rm, "b1", 0 * GB, null);
checkPendingResource(rm, "b", 0 * GB, null);
checkPendingResource(rm, "root", 0 * GB, null);
checkPendingResource(rm, "root", 0 * GB, "x");
}
private void checkPendingResource(MockRM rm, String queueName, int memory,
String label) {
CapacityScheduler cs = (CapacityScheduler) rm.getResourceScheduler();
@ -3247,7 +3247,7 @@ public class TestCapacityScheduler {
Resource minAllocResource = Resource.newInstance(minAllocMb, 1);
String queueName = "a1";
RMApp rmApp = rm.submitApp(amMemory, "app-1", "user_0", null, queueName);
assertEquals("RMApp does not containes minimum allocation",
minAllocResource, rmApp.getAMResourceRequests().get(0).getCapability());
@ -3479,7 +3479,7 @@ public class TestCapacityScheduler {
DominantResourceCalculator.class.getName());
verifyAMLimitForLeafQueue(config);
}
private FiCaSchedulerApp getFiCaSchedulerApp(MockRM rm,
ApplicationId appId) {
CapacityScheduler cs = (CapacityScheduler) rm.getResourceScheduler();
@ -3492,10 +3492,10 @@ public class TestCapacityScheduler {
Configuration conf =
TestUtils.getConfigurationWithQueueLabels(new Configuration(false));
conf.setBoolean(YarnConfiguration.NODE_LABELS_ENABLED, true);
final RMNodeLabelsManager mgr = new NullRMNodeLabelsManager();
mgr.init(conf);
MemoryRMStateStore memStore = new MemoryRMStateStore();
memStore.init(conf);
MockRM rm = new MockRM(conf, memStore) {
@ -3503,17 +3503,17 @@ public class TestCapacityScheduler {
return mgr;
}
};
rm.start();
MockNM nm1 = // label = ""
new MockNM("h1:1234", 200 * GB, rm.getResourceTrackerService());
nm1.registerNode();
// Launch app1 in queue=a1
RMApp app1 = rm.submitApp(1 * GB, "app", "user", null, "a1");
MockAM am1 = MockRM.launchAndRegisterAM(app1, rm, nm1);
// Allocate two more containers
am1.allocate(
Arrays.asList(ResourceRequest.newInstance(Priority.newInstance(1),
@ -3542,15 +3542,15 @@ public class TestCapacityScheduler {
.newInstance(0, containerId1,
ContainerUpdateType.INCREASE_RESOURCE,
Resources.createResource(3 * GB), null)));
FiCaSchedulerApp app = getFiCaSchedulerApp(rm, app1.getApplicationId());
Assert.assertEquals(2 * GB,
app.getAppAttemptResourceUsage().getPending().getMemorySize());
checkPendingResource(rm, "a1", 2 * GB, null);
checkPendingResource(rm, "a", 2 * GB, null);
checkPendingResource(rm, "root", 2 * GB, null);
// am1 asks to change containerId2 (2G -> 3G) and containerId3 (2G -> 5G)
am1.sendContainerResizingRequest(Arrays.asList(
UpdateContainerRequest
@ -3561,13 +3561,13 @@ public class TestCapacityScheduler {
.newInstance(0, containerId3,
ContainerUpdateType.INCREASE_RESOURCE,
Resources.createResource(5 * GB), null)));
Assert.assertEquals(6 * GB,
app.getAppAttemptResourceUsage().getPending().getMemorySize());
checkPendingResource(rm, "a1", 6 * GB, null);
checkPendingResource(rm, "a", 6 * GB, null);
checkPendingResource(rm, "root", 6 * GB, null);
// am1 asks to change containerId1 (1G->3G), containerId2 (2G -> 4G) and
// containerId3 (2G -> 2G)
am1.sendContainerResizingRequest(Arrays.asList(
@ -3650,7 +3650,7 @@ public class TestCapacityScheduler {
+ CapacitySchedulerConfiguration.MAXIMUM_ALLOCATION_VCORES;
conf.setInt(propName, maxAllocVcores);
}
private void sentRMContainerLaunched(MockRM rm, ContainerId containerId) {
CapacityScheduler cs = (CapacityScheduler) rm.getResourceScheduler();
RMContainer rmContainer = cs.getRMContainer(containerId);

90
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestLeafQueue.java
View File

@ -23,7 +23,6 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyBoolean;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
@ -1054,9 +1053,13 @@ public class TestLeafQueue {
//test case 3
qb.finishApplication(app_0.getApplicationId(), user_0);
qb.finishApplication(app_2.getApplicationId(), user_1);
qb.releaseResource(clusterResource, app_0, app_0.getResource(u0SchedKey),
qb.releaseResource(clusterResource, app_0,
app_0.getAppSchedulingInfo().getPendingAsk(u0SchedKey)
.getPerAllocationResource(),
null, null, false);
qb.releaseResource(clusterResource, app_2, app_2.getResource(u1SchedKey),
qb.releaseResource(clusterResource, app_2,
app_2.getAppSchedulingInfo().getPendingAsk(u1SchedKey)
.getPerAllocationResource(),
null, null, false);
qb.setUserLimit(50);
@ -1954,7 +1957,7 @@ public class TestLeafQueue {
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
verifyNoContainerAllocated(assignment);
assertEquals(1, app_0.getSchedulingOpportunities(schedulerKey));
assertEquals(3, app_0.getTotalRequiredResources(schedulerKey));
assertEquals(3, app_0.getOutstandingAsksCount(schedulerKey));
assertEquals(NodeType.NODE_LOCAL, assignment.getType()); // None->NODE_LOCAL
// Another off switch, shouldn't allocate due to delay scheduling
@ -1963,7 +1966,7 @@ public class TestLeafQueue {
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
verifyNoContainerAllocated(assignment);
assertEquals(2, app_0.getSchedulingOpportunities(schedulerKey));
assertEquals(3, app_0.getTotalRequiredResources(schedulerKey));
assertEquals(3, app_0.getOutstandingAsksCount(schedulerKey));
assertEquals(NodeType.NODE_LOCAL, assignment.getType()); // None->NODE_LOCAL
// Another off switch, shouldn't allocate due to delay scheduling
@ -1972,7 +1975,7 @@ public class TestLeafQueue {
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
verifyNoContainerAllocated(assignment);
assertEquals(3, app_0.getSchedulingOpportunities(schedulerKey));
assertEquals(3, app_0.getTotalRequiredResources(schedulerKey));
assertEquals(3, app_0.getOutstandingAsksCount(schedulerKey));
assertEquals(NodeType.NODE_LOCAL, assignment.getType()); // None->NODE_LOCAL
// Another off switch, now we should allocate
@ -1983,7 +1986,7 @@ public class TestLeafQueue {
verifyContainerAllocated(assignment, NodeType.OFF_SWITCH);
// should NOT reset
assertEquals(4, app_0.getSchedulingOpportunities(schedulerKey));
assertEquals(2, app_0.getTotalRequiredResources(schedulerKey));
assertEquals(2, app_0.getOutstandingAsksCount(schedulerKey));
// NODE_LOCAL - node_0
assignment = a.assignContainers(clusterResource, node_0,
@ -1992,7 +1995,7 @@ public class TestLeafQueue {
verifyContainerAllocated(assignment, NodeType.NODE_LOCAL);
// should reset
assertEquals(0, app_0.getSchedulingOpportunities(schedulerKey));
assertEquals(1, app_0.getTotalRequiredResources(schedulerKey));
assertEquals(1, app_0.getOutstandingAsksCount(schedulerKey));
// NODE_LOCAL - node_1
assignment = a.assignContainers(clusterResource, node_1,
@ -2001,7 +2004,7 @@ public class TestLeafQueue {
verifyContainerAllocated(assignment, NodeType.NODE_LOCAL);
// should reset
assertEquals(0, app_0.getSchedulingOpportunities(schedulerKey));
assertEquals(0, app_0.getTotalRequiredResources(schedulerKey));
assertEquals(0, app_0.getOutstandingAsksCount(schedulerKey));
assertEquals(NodeType.NODE_LOCAL, assignment.getType());
// Add 1 more request to check for RACK_LOCAL
@ -2016,7 +2019,7 @@ public class TestLeafQueue {
TestUtils.createResourceRequest(ResourceRequest.ANY, 1*GB, 4, // one extra
true, priority, recordFactory));
app_0.updateResourceRequests(app_0_requests_0);
assertEquals(4, app_0.getTotalRequiredResources(schedulerKey));
assertEquals(4, app_0.getOutstandingAsksCount(schedulerKey));
// Rack-delay
doReturn(true).when(a).getRackLocalityFullReset();
@ -2027,7 +2030,7 @@ public class TestLeafQueue {
new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
assertEquals(1, app_0.getSchedulingOpportunities(schedulerKey));
assertEquals(4, app_0.getTotalRequiredResources(schedulerKey));
assertEquals(4, app_0.getOutstandingAsksCount(schedulerKey));
// Should assign RACK_LOCAL now
assignment = a.assignContainers(clusterResource, node_3,
@ -2036,14 +2039,14 @@ public class TestLeafQueue {
verifyContainerAllocated(assignment, NodeType.RACK_LOCAL);
// should reset
assertEquals(0, app_0.getSchedulingOpportunities(schedulerKey));
assertEquals(3, app_0.getTotalRequiredResources(schedulerKey));
assertEquals(3, app_0.getOutstandingAsksCount(schedulerKey));
// Shouldn't assign RACK_LOCAL because schedulingOpportunities should have gotten reset.
assignment = a.assignContainers(clusterResource, node_3,
new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
assertEquals(1, app_0.getSchedulingOpportunities(schedulerKey));
assertEquals(3, app_0.getTotalRequiredResources(schedulerKey));
assertEquals(3, app_0.getOutstandingAsksCount(schedulerKey));
// Next time we schedule RACK_LOCAL, don't reset
doReturn(false).when(a).getRackLocalityFullReset();
@ -2055,7 +2058,7 @@ public class TestLeafQueue {
verifyContainerAllocated(assignment, NodeType.RACK_LOCAL);
// should NOT reset
assertEquals(2, app_0.getSchedulingOpportunities(schedulerKey));
assertEquals(2, app_0.getTotalRequiredResources(schedulerKey));
assertEquals(2, app_0.getOutstandingAsksCount(schedulerKey));
// Another RACK_LOCAL since schedulingOpportunities not reset
assignment = a.assignContainers(clusterResource, node_3,
@ -2064,7 +2067,7 @@ public class TestLeafQueue {
verifyContainerAllocated(assignment, NodeType.RACK_LOCAL);
// should NOT reset
assertEquals(3, app_0.getSchedulingOpportunities(schedulerKey));
assertEquals(1, app_0.getTotalRequiredResources(schedulerKey));
assertEquals(1, app_0.getOutstandingAsksCount(schedulerKey));
// Add a request larger than cluster size to verify
// OFF_SWITCH delay is capped by cluster size
@ -2170,7 +2173,7 @@ public class TestLeafQueue {
CSAssignment assignment = null;
SchedulerRequestKey schedulerKey = toSchedulerKey(priority);
assertEquals(3, app1.getTotalRequiredResources(schedulerKey));
assertEquals(3, app1.getPendingAsk(schedulerKey, ResourceRequest.ANY).getCount());
// No rack-local yet.
assignment = a.assignContainers(clusterResource, node2,
@ -2179,7 +2182,8 @@ public class TestLeafQueue {
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
verifyNoContainerAllocated(assignment);
assertEquals(1, app1.getSchedulingOpportunities(schedulerKey));
assertEquals(3, app1.getTotalRequiredResources(schedulerKey));
assertEquals(3,
app1.getPendingAsk(schedulerKey, ResourceRequest.ANY).getCount());
assertEquals(NodeType.NODE_LOCAL, assignment.getType()); // None->NODE_LOCAL
// Still no rack-local.
@ -2188,7 +2192,8 @@ public class TestLeafQueue {
SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
assertEquals(2, app1.getSchedulingOpportunities(schedulerKey));
assertEquals(3, app1.getTotalRequiredResources(schedulerKey));
assertEquals(3,
app1.getPendingAsk(schedulerKey, ResourceRequest.ANY).getCount());
assertEquals(NodeType.NODE_LOCAL, assignment.getType()); // None->NODE_LOCAL
// Rack local now.
@ -2197,7 +2202,8 @@ public class TestLeafQueue {
SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
assertEquals(0, app1.getSchedulingOpportunities(schedulerKey));
assertEquals(2, app1.getTotalRequiredResources(schedulerKey));
assertEquals(2,
app1.getPendingAsk(schedulerKey, ResourceRequest.ANY).getCount());
assertEquals(NodeType.RACK_LOCAL, assignment.getType());
// No off-switch until 3 missed opportunities.
@ -2214,7 +2220,8 @@ public class TestLeafQueue {
SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
assertEquals(3, app1.getSchedulingOpportunities(schedulerKey));
assertEquals(2, app1.getTotalRequiredResources(schedulerKey));
assertEquals(2,
app1.getPendingAsk(schedulerKey, ResourceRequest.ANY).getCount());
assertEquals(NodeType.NODE_LOCAL, assignment.getType()); // None->NODE_LOCAL
// Now off-switch should succeed.
@ -2223,7 +2230,8 @@ public class TestLeafQueue {
SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
assertEquals(4, app1.getSchedulingOpportunities(schedulerKey));
assertEquals(1, app1.getTotalRequiredResources(schedulerKey));
assertEquals(1,
app1.getPendingAsk(schedulerKey, ResourceRequest.ANY).getCount());
assertEquals(NodeType.OFF_SWITCH, assignment.getType());
// Check capping by number of cluster nodes.
@ -2235,14 +2243,16 @@ public class TestLeafQueue {
SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
assertEquals(5, app1.getSchedulingOpportunities(schedulerKey));
assertEquals(1, app1.getTotalRequiredResources(schedulerKey));
assertEquals(1,
app1.getPendingAsk(schedulerKey, ResourceRequest.ANY).getCount());
assertEquals(NodeType.NODE_LOCAL, assignment.getType()); // None->NODE_LOCAL
assignment = a.assignContainers(clusterResource, node3,
new ResourceLimits(clusterResource),
SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
assertEquals(6, app1.getSchedulingOpportunities(schedulerKey));
assertEquals(0, app1.getTotalRequiredResources(schedulerKey));
assertEquals(0,
app1.getPendingAsk(schedulerKey, ResourceRequest.ANY).getCount());
assertEquals(NodeType.OFF_SWITCH, assignment.getType());
}
@ -2331,9 +2341,9 @@ public class TestLeafQueue {
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
verifyNoContainerAllocated(assignment);
assertEquals(1, app_0.getSchedulingOpportunities(schedulerKey1));
assertEquals(2, app_0.getTotalRequiredResources(schedulerKey1));
assertEquals(2, app_0.getOutstandingAsksCount(schedulerKey1));
assertEquals(0, app_0.getSchedulingOpportunities(schedulerKey2));
assertEquals(1, app_0.getTotalRequiredResources(schedulerKey2));
assertEquals(1, app_0.getOutstandingAsksCount(schedulerKey2));
// Another off-switch, shouldn't allocate P1 due to delay scheduling
// thus, no P2 either!
@ -2342,9 +2352,9 @@ public class TestLeafQueue {
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
verifyNoContainerAllocated(assignment);
assertEquals(2, app_0.getSchedulingOpportunities(schedulerKey1));
assertEquals(2, app_0.getTotalRequiredResources(schedulerKey1));
assertEquals(2, app_0.getOutstandingAsksCount(schedulerKey1));
assertEquals(0, app_0.getSchedulingOpportunities(schedulerKey2));
assertEquals(1, app_0.getTotalRequiredResources(schedulerKey2));
assertEquals(1, app_0.getOutstandingAsksCount(schedulerKey2));
// Another off-switch, shouldn't allocate OFF_SWITCH P1
assignment = a.assignContainers(clusterResource, node_2,
@ -2352,9 +2362,9 @@ public class TestLeafQueue {
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
verifyContainerAllocated(assignment, NodeType.OFF_SWITCH);
assertEquals(3, app_0.getSchedulingOpportunities(schedulerKey1));
assertEquals(1, app_0.getTotalRequiredResources(schedulerKey1));
assertEquals(1, app_0.getOutstandingAsksCount(schedulerKey1));
assertEquals(0, app_0.getSchedulingOpportunities(schedulerKey2));
assertEquals(1, app_0.getTotalRequiredResources(schedulerKey2));
assertEquals(1, app_0.getOutstandingAsksCount(schedulerKey2));
// Now, DATA_LOCAL for P1
assignment = a.assignContainers(clusterResource, node_0,
@ -2362,9 +2372,9 @@ public class TestLeafQueue {
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
verifyContainerAllocated(assignment, NodeType.NODE_LOCAL);
assertEquals(0, app_0.getSchedulingOpportunities(schedulerKey1));
assertEquals(0, app_0.getTotalRequiredResources(schedulerKey1));
assertEquals(0, app_0.getOutstandingAsksCount(schedulerKey1));
assertEquals(0, app_0.getSchedulingOpportunities(schedulerKey2));
assertEquals(1, app_0.getTotalRequiredResources(schedulerKey2));
assertEquals(1, app_0.getOutstandingAsksCount(schedulerKey2));
// Now, OFF_SWITCH for P2
assignment = a.assignContainers(clusterResource, node_1,
@ -2372,9 +2382,9 @@ public class TestLeafQueue {
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
verifyContainerAllocated(assignment, NodeType.OFF_SWITCH);
assertEquals(0, app_0.getSchedulingOpportunities(schedulerKey1));
assertEquals(0, app_0.getTotalRequiredResources(schedulerKey1));
assertEquals(0, app_0.getOutstandingAsksCount(schedulerKey1));
assertEquals(1, app_0.getSchedulingOpportunities(schedulerKey2));
assertEquals(0, app_0.getTotalRequiredResources(schedulerKey2));
assertEquals(0, app_0.getOutstandingAsksCount(schedulerKey2));
}
@ -2455,7 +2465,7 @@ public class TestLeafQueue {
verifyContainerAllocated(assignment, NodeType.NODE_LOCAL);
assertEquals(0, app_0.getSchedulingOpportunities(schedulerKey));
// should reset
assertEquals(0, app_0.getTotalRequiredResources(schedulerKey));
assertEquals(0, app_0.getOutstandingAsksCount(schedulerKey));
// No allocation on node_1_0 even though it's node/rack local since
// required(ANY) == 0
@ -2466,7 +2476,7 @@ public class TestLeafQueue {
// Still zero
// since #req=0
assertEquals(0, app_0.getSchedulingOpportunities(schedulerKey));
assertEquals(0, app_0.getTotalRequiredResources(schedulerKey));
assertEquals(0, app_0.getOutstandingAsksCount(schedulerKey));
// Add one request
app_0_requests_0.clear();
@ -2482,7 +2492,7 @@ public class TestLeafQueue {
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
verifyNoContainerAllocated(assignment);
assertEquals(1, app_0.getSchedulingOpportunities(schedulerKey));
assertEquals(1, app_0.getTotalRequiredResources(schedulerKey));
assertEquals(1, app_0.getOutstandingAsksCount(schedulerKey));
// NODE_LOCAL - node_1
assignment = a.assignContainers(clusterResource, node_1_0,
@ -2491,7 +2501,7 @@ public class TestLeafQueue {
verifyContainerAllocated(assignment, NodeType.NODE_LOCAL);
// should reset
assertEquals(0, app_0.getSchedulingOpportunities(schedulerKey));
assertEquals(0, app_0.getTotalRequiredResources(schedulerKey));
assertEquals(0, app_0.getOutstandingAsksCount(schedulerKey));
}
@Test (timeout = 30000)
@ -2870,7 +2880,7 @@ public class TestLeafQueue {
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
verifyNoContainerAllocated(assignment);
assertEquals(0, app_0.getSchedulingOpportunities(schedulerKey));
assertEquals(1, app_0.getTotalRequiredResources(schedulerKey));
assertEquals(1, app_0.getOutstandingAsksCount(schedulerKey));
// Now sanity-check node_local
app_0_requests_0.add(
@ -2901,7 +2911,7 @@ public class TestLeafQueue {
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
verifyContainerAllocated(assignment, NodeType.NODE_LOCAL);
assertEquals(0, app_0.getSchedulingOpportunities(schedulerKey));
assertEquals(0, app_0.getTotalRequiredResources(schedulerKey));
assertEquals(0, app_0.getOutstandingAsksCount(schedulerKey));
}
@ -3354,7 +3364,7 @@ public class TestLeafQueue {
applyCSAssignment(clusterResource, assignment, a, nodes, apps);
verifyContainerAllocated(assignment, NodeType.NODE_LOCAL);
assertEquals(1, app_0.getSchedulingOpportunities(schedulerKey));
assertEquals(3, app_0.getTotalRequiredResources(schedulerKey));
assertEquals(3, app_0.getOutstandingAsksCount(schedulerKey));
assertEquals(0, app_0.getLiveContainers().size());
assertEquals(1, app_1.getLiveContainers().size());
}

65
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestNodeLabelContainerAllocation.java
View File

@ -46,6 +46,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptS
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AppSchedulingInfo;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerAppReport;
@ -56,6 +57,8 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaS
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerNode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.AppAttemptRemovedSchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeUpdateSchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.PendingAsk;
import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@ -548,11 +551,12 @@ public class TestNodeLabelContainerAllocation {
ApplicationAttemptId attemptId, int memory) {
CapacityScheduler cs = (CapacityScheduler) rm.getRMContext().getScheduler();
FiCaSchedulerApp app = cs.getApplicationAttempt(attemptId);
ResourceRequest rr =
app.getAppSchedulingInfo().getResourceRequest(
PendingAsk ask =
app.getAppSchedulingInfo().getPendingAsk(
TestUtils.toSchedulerKey(priority), "*");
Assert.assertEquals(memory,
rr.getCapability().getMemorySize() * rr.getNumContainers());
ask.getPerAllocationResource().getMemorySize() * ask
.getCount());
}
private void checkLaunchedContainerNumOnNode(MockRM rm, NodeId nodeId,
@ -607,18 +611,10 @@ public class TestNodeLabelContainerAllocation {
(CapacityScheduler) rm1.getRMContext().getScheduler();
FiCaSchedulerApp app =
cs.getApplicationAttempt(am1.getApplicationAttemptId());
List<ResourceRequest> allResourceRequests =
app.getAppSchedulingInfo().getAllResourceRequests();
for (ResourceRequest changeReq : allResourceRequests) {
if (changeReq.getPriority().getPriority() == 2
|| changeReq.getPriority().getPriority() == 3) {
Assert.assertEquals("Expected label y", "y",
changeReq.getNodeLabelExpression());
} else if (changeReq.getPriority().getPriority() == 4) {
Assert.assertEquals("Expected label EMPTY",
RMNodeLabelsManager.NO_LABEL, changeReq.getNodeLabelExpression());
}
}
checkNodePartitionOfRequestedPriority(app.getAppSchedulingInfo(), 2, "y");
checkNodePartitionOfRequestedPriority(app.getAppSchedulingInfo(), 3, "y");
checkNodePartitionOfRequestedPriority(app.getAppSchedulingInfo(), 4,
RMNodeLabelsManager.NO_LABEL);
// Previous any request was Y trying to update with z and the
// request before ANY label is null
@ -628,17 +624,11 @@ public class TestNodeLabelContainerAllocation {
newReq.add(am1.createResourceReq("h1:1234", 1024, 3, 4, null));
newReq.add(am1.createResourceReq("*", 1024, 4, 5, "z"));
am1.allocate(newReq, new ArrayList<ContainerId>());
allResourceRequests = app.getAppSchedulingInfo().getAllResourceRequests();
for (ResourceRequest changeReq : allResourceRequests) {
if (changeReq.getPriority().getPriority() == 3
|| changeReq.getPriority().getPriority() == 4) {
Assert.assertEquals("Expected label z", "z",
changeReq.getNodeLabelExpression());
} else if (changeReq.getPriority().getPriority() == 2) {
Assert.assertEquals("Expected label y", "y",
changeReq.getNodeLabelExpression());
}
}
checkNodePartitionOfRequestedPriority(app.getAppSchedulingInfo(), 3, "z");
checkNodePartitionOfRequestedPriority(app.getAppSchedulingInfo(), 4, "z");
checkNodePartitionOfRequestedPriority(app.getAppSchedulingInfo(), 2, "y");
// Request before ANY and ANY request is set as NULL. Request should be set
// with Empty Label
List<ResourceRequest> resourceRequest1 = new ArrayList<ResourceRequest>();
@ -653,14 +643,21 @@ public class TestNodeLabelContainerAllocation {
RMNodeLabelsManager.NO_LABEL));
resourceRequest1.add(am1.createResourceReq("h2:1234", 1024, 2, 4, null));
am1.allocate(resourceRequest1, new ArrayList<ContainerId>());
allResourceRequests = app.getAppSchedulingInfo().getAllResourceRequests();
for (ResourceRequest changeReq : allResourceRequests) {
if (changeReq.getPriority().getPriority() == 3) {
Assert.assertEquals("Expected label Empty",
RMNodeLabelsManager.NO_LABEL, changeReq.getNodeLabelExpression());
} else if (changeReq.getPriority().getPriority() == 2) {
Assert.assertEquals("Expected label y", RMNodeLabelsManager.NO_LABEL,
changeReq.getNodeLabelExpression());
checkNodePartitionOfRequestedPriority(app.getAppSchedulingInfo(), 3,
RMNodeLabelsManager.NO_LABEL);
checkNodePartitionOfRequestedPriority(app.getAppSchedulingInfo(), 2,
RMNodeLabelsManager.NO_LABEL);
}
private void checkNodePartitionOfRequestedPriority(AppSchedulingInfo info,
int priority, String expectedPartition) {
for (SchedulerRequestKey key : info.getSchedulerKeys()) {
if (key.getPriority().getPriority() == priority) {
Assert.assertEquals("Expected partition is " + expectedPartition,
expectedPartition,
info.getSchedulingPlacementSet(key)
.getPrimaryRequestedNodePartition());
}
}
}

22
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestReservations.java
View File

@ -327,7 +327,7 @@ public class TestReservations {
assertEquals(5 * GB, node_0.getAllocatedResource().getMemorySize());
assertEquals(3 * GB, node_1.getAllocatedResource().getMemorySize());
assertEquals(0 * GB, node_2.getAllocatedResource().getMemorySize());
assertEquals(2, app_0.getTotalRequiredResources(
assertEquals(2, app_0.getOutstandingAsksCount(
toSchedulerKey(priorityReduce)));
// try to assign reducer (5G on node 0 and should reserve)
@ -346,7 +346,7 @@ public class TestReservations {
assertEquals(5 * GB, node_0.getAllocatedResource().getMemorySize());
assertEquals(3 * GB, node_1.getAllocatedResource().getMemorySize());
assertEquals(0 * GB, node_2.getAllocatedResource().getMemorySize());
assertEquals(2, app_0.getTotalRequiredResources(
assertEquals(2, app_0.getOutstandingAsksCount(
toSchedulerKey(priorityReduce)));
// assign reducer to node 2
@ -365,7 +365,7 @@ public class TestReservations {
assertEquals(5 * GB, node_0.getAllocatedResource().getMemorySize());
assertEquals(3 * GB, node_1.getAllocatedResource().getMemorySize());
assertEquals(5 * GB, node_2.getAllocatedResource().getMemorySize());
assertEquals(1, app_0.getTotalRequiredResources(
assertEquals(1, app_0.getOutstandingAsksCount(
toSchedulerKey(priorityReduce)));
// node_1 heartbeat and unreserves from node_0 in order to allocate
@ -384,7 +384,7 @@ public class TestReservations {
assertEquals(5 * GB, node_0.getAllocatedResource().getMemorySize());
assertEquals(8 * GB, node_1.getAllocatedResource().getMemorySize());
assertEquals(5 * GB, node_2.getAllocatedResource().getMemorySize());
assertEquals(0, app_0.getTotalRequiredResources(
assertEquals(0, app_0.getOutstandingAsksCount(
toSchedulerKey(priorityReduce)));
}
@ -660,7 +660,7 @@ public class TestReservations {
assertEquals(5 * GB, node_0.getAllocatedResource().getMemorySize());
assertEquals(3 * GB, node_1.getAllocatedResource().getMemorySize());
assertEquals(0 * GB, node_2.getAllocatedResource().getMemorySize());
assertEquals(2, app_0.getTotalRequiredResources(
assertEquals(2, app_0.getOutstandingAsksCount(
toSchedulerKey(priorityReduce)));
// try to assign reducer (5G on node 0 and should reserve)
@ -679,7 +679,7 @@ public class TestReservations {
assertEquals(5 * GB, node_0.getAllocatedResource().getMemorySize());
assertEquals(3 * GB, node_1.getAllocatedResource().getMemorySize());
assertEquals(0 * GB, node_2.getAllocatedResource().getMemorySize());
assertEquals(2, app_0.getTotalRequiredResources(
assertEquals(2, app_0.getOutstandingAsksCount(
toSchedulerKey(priorityReduce)));
// assign reducer to node 2
@ -698,7 +698,7 @@ public class TestReservations {
assertEquals(5 * GB, node_0.getAllocatedResource().getMemorySize());
assertEquals(3 * GB, node_1.getAllocatedResource().getMemorySize());
assertEquals(5 * GB, node_2.getAllocatedResource().getMemorySize());
assertEquals(1, app_0.getTotalRequiredResources(
assertEquals(1, app_0.getOutstandingAsksCount(
toSchedulerKey(priorityReduce)));
// node_1 heartbeat and won't unreserve from node_0, potentially stuck
@ -718,7 +718,7 @@ public class TestReservations {
assertEquals(5 * GB, node_0.getAllocatedResource().getMemorySize());
assertEquals(3 * GB, node_1.getAllocatedResource().getMemorySize());
assertEquals(5 * GB, node_2.getAllocatedResource().getMemorySize());
assertEquals(1, app_0.getTotalRequiredResources(
assertEquals(1, app_0.getOutstandingAsksCount(
toSchedulerKey(priorityReduce)));
}
@ -839,7 +839,7 @@ public class TestReservations {
assertEquals(null, node_0.getReservedContainer());
assertEquals(5 * GB, node_0.getAllocatedResource().getMemorySize());
assertEquals(3 * GB, node_1.getAllocatedResource().getMemorySize());
assertEquals(2, app_0.getTotalRequiredResources(
assertEquals(2, app_0.getOutstandingAsksCount(
toSchedulerKey(priorityReduce)));
// try to assign reducer (5G on node 0 and should reserve)
@ -857,7 +857,7 @@ public class TestReservations {
.getMemorySize());
assertEquals(5 * GB, node_0.getAllocatedResource().getMemorySize());
assertEquals(3 * GB, node_1.getAllocatedResource().getMemorySize());
assertEquals(2, app_0.getTotalRequiredResources(
assertEquals(2, app_0.getOutstandingAsksCount(
toSchedulerKey(priorityReduce)));
// could allocate but told need to unreserve first
@ -874,7 +874,7 @@ public class TestReservations {
assertEquals(null, node_0.getReservedContainer());
assertEquals(5 * GB, node_0.getAllocatedResource().getMemorySize());
assertEquals(8 * GB, node_1.getAllocatedResource().getMemorySize());
assertEquals(1, app_0.getTotalRequiredResources(
assertEquals(1, app_0.getOutstandingAsksCount(
toSchedulerKey(priorityReduce)));
}