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

YARN-4963. capacity scheduler: Make number of OFF_SWITCH assignments per heartbeat configurable. Contributed by Nathan Roberts

This commit is contained in:
Jason Lowe 2016-10-28 17:44:56 +00:00
parent 9cc7ab4e96
commit 93165c1ade
4 changed files with 79 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/conf/capacity-scheduler.xml
View File

@ -131,4 +131,16 @@
</description>
</property>
<property>
<name>yarn.scheduler.capacity.per-node-heartbeat.maximum-offswitch-assignments</name>
<value>1</value>
<description>
Controls the number of OFF_SWITCH assignments allowed
during a node's heartbeat. Increasing this value can improve
scheduling rate for OFF_SWITCH containers. Lower values reduce
"clumping" of applications on particular nodes. The default is 1.
Legal values are 1-MAX_INT. This config is refreshable.
</description>
</property>
</configuration>

21
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java
View File

@ -190,6 +190,13 @@ public class CapacitySchedulerConfiguration extends ReservationSchedulerConfigur
public static final String RACK_LOCALITY_FULL_RESET =
PREFIX + "rack-locality-full-reset";
@Private
public static final int DEFAULT_OFFSWITCH_PER_HEARTBEAT_LIMIT = 1;
@Private
public static final String OFFSWITCH_PER_HEARTBEAT_LIMIT =
PREFIX + "per-node-heartbeat.maximum-offswitch-assignments";
@Private
public static final boolean DEFAULT_RACK_LOCALITY_FULL_RESET = true;
@ -707,6 +714,20 @@ public class CapacitySchedulerConfiguration extends ReservationSchedulerConfigur
return getBoolean(ENABLE_USER_METRICS, DEFAULT_ENABLE_USER_METRICS);
}
public int getOffSwitchPerHeartbeatLimit() {
int limit = getInt(OFFSWITCH_PER_HEARTBEAT_LIMIT,
DEFAULT_OFFSWITCH_PER_HEARTBEAT_LIMIT);
if (limit < 1) {
LOG.warn(OFFSWITCH_PER_HEARTBEAT_LIMIT + "(" + limit + ") < 1. Using 1.");
limit = 1;
}
return limit;
}
public void setOffSwitchPerHeartbeatLimit(int limit) {
setInt(OFFSWITCH_PER_HEARTBEAT_LIMIT, limit);
}
public int getNodeLocalityDelay() {
return getInt(NODE_LOCALITY_DELAY, DEFAULT_NODE_LOCALITY_DELAY);
}

33
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/ParentQueue.java
View File

@ -76,6 +76,7 @@ public class ParentQueue extends AbstractCSQueue {
volatile int numApplications;
private final CapacitySchedulerContext scheduler;
private boolean needToResortQueuesAtNextAllocation = false;
private int offswitchPerHeartbeatLimit;
private final RecordFactory recordFactory =
RecordFactoryProvider.getRecordFactory(null);
@ -123,6 +124,9 @@ public class ParentQueue extends AbstractCSQueue {
}
}
offswitchPerHeartbeatLimit =
csContext.getConfiguration().getOffSwitchPerHeartbeatLimit();
LOG.info(queueName +
", capacity=" + this.queueCapacities.getCapacity() +
", asboluteCapacity=" + this.queueCapacities.getAbsoluteCapacity() +
@ -130,7 +134,8 @@ public class ParentQueue extends AbstractCSQueue {
", asboluteMaxCapacity=" + this.queueCapacities.getAbsoluteMaximumCapacity() +
", state=" + state +
", acls=" + aclsString +
", labels=" + labelStrBuilder.toString() + "\n" +
", labels=" + labelStrBuilder.toString() +
", offswitchPerHeartbeatLimit = " + getOffSwitchPerHeartbeatLimit() +
", reservationsContinueLooking=" + reservationsContinueLooking);
}
@ -196,6 +201,11 @@ public class ParentQueue extends AbstractCSQueue {
return queueInfo;
}
@Private
public int getOffSwitchPerHeartbeatLimit() {
return offswitchPerHeartbeatLimit;
}
private synchronized QueueUserACLInfo getUserAclInfo(
UserGroupInformation user) {
QueueUserACLInfo userAclInfo =
@ -383,6 +393,8 @@ public class ParentQueue extends AbstractCSQueue {
public synchronized CSAssignment assignContainers(Resource clusterResource,
FiCaSchedulerNode node, ResourceLimits resourceLimits,
SchedulingMode schedulingMode) {
int offswitchCount = 0;
// if our queue cannot access this node, just return
if (schedulingMode == SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY
&& !accessibleToPartition(node.getPartition())) {
@ -478,13 +490,18 @@ public class ParentQueue extends AbstractCSQueue {
+ " absoluteUsedCapacity=" + getAbsoluteUsedCapacity());
}
// Do not assign more than one container if this isn't the root queue
// or if we've already assigned an off-switch container
if (!rootQueue || assignment.getType() == NodeType.OFF_SWITCH) {
if (assignment.getType() == NodeType.OFF_SWITCH) {
offswitchCount++;
}
// Do not assign more containers if this isn't the root queue
// or if we've already assigned enough OFF_SWITCH containers in
// this pass
if (!rootQueue || offswitchCount >= getOffSwitchPerHeartbeatLimit()) {
if (LOG.isDebugEnabled()) {
if (rootQueue && assignment.getType() == NodeType.OFF_SWITCH) {
LOG.debug("Not assigning more than one off-switch container," +
" assignments so far: " + assignment);
if (rootQueue && offswitchCount >= getOffSwitchPerHeartbeatLimit()) {
LOG.debug("Assigned maximum number of off-switch containers: " +
offswitchCount + ", assignments so far: " + assignment);
}
}
break;
@ -821,4 +838,4 @@ public class ParentQueue extends AbstractCSQueue {
public synchronized int getNumApplications() {
return numApplications;
}
}
}

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

@ -29,6 +29,7 @@ import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.when;
import java.util.HashMap;
@ -611,6 +612,8 @@ public class TestParentQueue {
public void testOffSwitchScheduling() throws Exception {
// Setup queue configs
setupSingleLevelQueues(csConf);
csConf.setOffSwitchPerHeartbeatLimit(2);
Map<String, CSQueue> queues = new HashMap<String, CSQueue>();
CSQueue root =
@ -641,12 +644,18 @@ public class TestParentQueue {
queues.get(CapacitySchedulerConfiguration.ROOT).getQueueResourceUsage()
.incPending(Resources.createResource(1 * GB));
// Simulate B returning a container on node_0
stubQueueAllocation(a, clusterResource, node_0, 0*GB, NodeType.OFF_SWITCH);
// Simulate returning 2 containers on node_0 before offswitch limit
stubQueueAllocation(a, clusterResource, node_0, 1*GB, NodeType.OFF_SWITCH);
stubQueueAllocation(b, clusterResource, node_0, 1*GB, NodeType.OFF_SWITCH);
root.assignContainers(clusterResource, node_0,
new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
verifyQueueMetrics(a, 0*GB, clusterResource);
InOrder allocationOrder = inOrder(a, b);
allocationOrder.verify(a, times(1)).assignContainers(eq(clusterResource),
any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class));
allocationOrder.verify(b, times(1)).assignContainers(eq(clusterResource),
any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class));
verifyQueueMetrics(a, 1*GB, clusterResource);
verifyQueueMetrics(b, 1*GB, clusterResource);
// Now, A should get the scheduling opportunity since A=0G/6G, B=1G/14G
@ -655,27 +664,28 @@ public class TestParentQueue {
stubQueueAllocation(b, clusterResource, node_1, 1*GB, NodeType.OFF_SWITCH);
root.assignContainers(clusterResource, node_1,
new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
InOrder allocationOrder = inOrder(a, b);
allocationOrder.verify(a).assignContainers(eq(clusterResource),
allocationOrder = inOrder(a, b);
allocationOrder.verify(a, times(1)).assignContainers(eq(clusterResource),
any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class));
allocationOrder.verify(b).assignContainers(eq(clusterResource),
allocationOrder.verify(b, times(1)).assignContainers(eq(clusterResource),
any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class));
verifyQueueMetrics(a, 2*GB, clusterResource);
verifyQueueMetrics(a, 3*GB, clusterResource);
verifyQueueMetrics(b, 2*GB, clusterResource);
// Now, B should get the scheduling opportunity
// since A has 2/6G while B has 2/14G,
// However, since B returns off-switch, A won't get an opportunity
// A also gets an opportunity because offswitchlimit not reached
stubQueueAllocation(a, clusterResource, node_0, 1*GB, NodeType.NODE_LOCAL);
stubQueueAllocation(b, clusterResource, node_0, 2*GB, NodeType.OFF_SWITCH);
root.assignContainers(clusterResource, node_0,
new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
allocationOrder = inOrder(b, a);
allocationOrder.verify(b).assignContainers(eq(clusterResource),
allocationOrder.verify(b, times(1)).assignContainers(eq(clusterResource),
any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class));
allocationOrder.verify(a).assignContainers(eq(clusterResource),
allocationOrder.verify(a, times(1)).assignContainers(eq(clusterResource),
any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class));
verifyQueueMetrics(a, 2*GB, clusterResource);
verifyQueueMetrics(a, 4*GB, clusterResource);
verifyQueueMetrics(b, 4*GB, clusterResource);
}