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

YARN-3485. FairScheduler headroom calculation doesn't consider maxResources for Fifo and FairShare policies. (kasha) 

(cherry picked from commit 8f82970e0c)
This commit is contained in:
Karthik Kambatla 2015-04-28 21:00:35 -07:00
parent fc9e7d7a24
commit 4cc38df7de
7 changed files with 76 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
hadoop-yarn-project/CHANGES.txt
View File

@ -263,6 +263,9 @@ Release 2.7.1 - UNRELEASED
YARN-3464. Race condition in LocalizerRunner kills localizer before
localizing all resources. (Zhihai Xu via kasha)
YARN-3485. FairScheduler headroom calculation doesn't consider
maxResources for Fifo and FairShare policies. (kasha)
Release 2.7.0 - 2015-04-20

20
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

@ -172,6 +172,10 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
+ this.attemptResourceUsage.getReserved());
}
/**
* Headroom depends on resources in the cluster, current usage of the
* queue, queue's fair-share and queue's max-resources.
*/
@Override
public Resource getHeadroom() {
final FSQueue queue = (FSQueue) this.queue;
@ -182,18 +186,22 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
Resource clusterResource = this.scheduler.getClusterResource();
Resource clusterUsage = this.scheduler.getRootQueueMetrics()
.getAllocatedResources();
Resource clusterAvailableResource = Resources.subtract(clusterResource,
clusterUsage);
Resource clusterAvailableResources =
Resources.subtract(clusterResource, clusterUsage);
Resource queueMaxAvailableResources =
Resources.subtract(queue.getMaxShare(), queueUsage);
Resource maxAvailableResource = Resources.componentwiseMin(
clusterAvailableResources, queueMaxAvailableResources);
Resource headroom = policy.getHeadroom(queueFairShare,
queueUsage, clusterAvailableResource);
queueUsage, maxAvailableResource);
if (LOG.isDebugEnabled()) {
LOG.debug("Headroom calculation for " + this.getName() + ":" +
"Min(" +
"(queueFairShare=" + queueFairShare +
" - queueUsage=" + queueUsage + ")," +
" clusterAvailableResource=" + clusterAvailableResource +
"(clusterResource=" + clusterResource +
" - clusterUsage=" + clusterUsage + ")" +
" maxAvailableResource=" + maxAvailableResource +
"Headroom=" + headroom);
}
return headroom;

4
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/SchedulingPolicy.java
View File

@ -185,10 +185,10 @@ public abstract class SchedulingPolicy {
*
* @param queueFairShare fairshare in the queue
* @param queueUsage resources used in the queue
* @param clusterAvailable available resource in cluster
* @param maxAvailable available resource in cluster for this queue
* @return calculated headroom
*/
public abstract Resource getHeadroom(Resource queueFairShare,
Resource queueUsage, Resource clusterAvailable);
Resource queueUsage, Resource maxAvailable);
}

6
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/policies/DominantResourceFairnessPolicy.java
View File

@ -90,15 +90,15 @@ public class DominantResourceFairnessPolicy extends SchedulingPolicy {
@Override
public Resource getHeadroom(Resource queueFairShare, Resource queueUsage,
Resource clusterAvailable) {
Resource maxAvailable) {
int queueAvailableMemory =
Math.max(queueFairShare.getMemory() - queueUsage.getMemory(), 0);
int queueAvailableCPU =
Math.max(queueFairShare.getVirtualCores() - queueUsage
.getVirtualCores(), 0);
Resource headroom = Resources.createResource(
Math.min(clusterAvailable.getMemory(), queueAvailableMemory),
Math.min(clusterAvailable.getVirtualCores(),
Math.min(maxAvailable.getMemory(), queueAvailableMemory),
Math.min(maxAvailable.getVirtualCores(),
queueAvailableCPU));
return headroom;
}

6
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/policies/FairSharePolicy.java
View File

@ -116,12 +116,12 @@ public class FairSharePolicy extends SchedulingPolicy {
@Override
public Resource getHeadroom(Resource queueFairShare,
Resource queueUsage, Resource clusterAvailable) {
Resource queueUsage, Resource maxAvailable) {
int queueAvailableMemory = Math.max(
queueFairShare.getMemory() - queueUsage.getMemory(), 0);
Resource headroom = Resources.createResource(
Math.min(clusterAvailable.getMemory(), queueAvailableMemory),
clusterAvailable.getVirtualCores());
Math.min(maxAvailable.getMemory(), queueAvailableMemory),
maxAvailable.getVirtualCores());
return headroom;
}

6
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/policies/FifoPolicy.java
View File

@ -109,12 +109,12 @@ public class FifoPolicy extends SchedulingPolicy {
@Override
public Resource getHeadroom(Resource queueFairShare,
Resource queueUsage, Resource clusterAvailable) {
Resource queueUsage, Resource maxAvailable) {
int queueAvailableMemory = Math.max(
queueFairShare.getMemory() - queueUsage.getMemory(), 0);
Resource headroom = Resources.createResource(
Math.min(clusterAvailable.getMemory(), queueAvailableMemory),
clusterAvailable.getVirtualCores());
Math.min(maxAvailable.getMemory(), queueAvailableMemory),
maxAvailable.getVirtualCores());
return headroom;
}

66
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFSAppAttempt.java
View File

@ -198,18 +198,24 @@ public class TestFSAppAttempt extends FairSchedulerTestBase {
Mockito.when(mockScheduler.getClock()).thenReturn(scheduler.getClock());
final FSLeafQueue mockQueue = Mockito.mock(FSLeafQueue.class);
final Resource queueFairShare = Resources.createResource(4096, 4);
final Resource queueUsage = Resource.newInstance(1024, 1);
final Resource queueMaxResources = Resource.newInstance(5 * 1024, 3);
final Resource queueFairShare = Resources.createResource(4096, 2);
final Resource queueUsage = Resource.newInstance(2048, 2);
final Resource queueStarvation =
Resources.subtract(queueFairShare, queueUsage);
final Resource queueMaxResourcesAvailable =
Resources.subtract(queueMaxResources, queueUsage);
final Resource clusterResource = Resources.createResource(8192, 8);
final Resource clusterUsage = Resources.createResource(6144, 2);
final Resource clusterUsage = Resources.createResource(2048, 2);
final Resource clusterAvailable =
Resources.subtract(clusterResource, clusterUsage);
final QueueMetrics fakeRootQueueMetrics = Mockito.mock(QueueMetrics.class);
ApplicationAttemptId applicationAttemptId = createAppAttemptId(1, 1);
RMContext rmContext = resourceManager.getRMContext();
FSAppAttempt schedulerApp =
new FSAppAttempt(mockScheduler, applicationAttemptId, "user1", mockQueue ,
null, rmContext);
Mockito.when(mockQueue.getMaxShare()).thenReturn(queueMaxResources);
Mockito.when(mockQueue.getFairShare()).thenReturn(queueFairShare);
Mockito.when(mockQueue.getResourceUsage()).thenReturn(queueUsage);
Mockito.when(mockScheduler.getClusterResource()).thenReturn
@ -219,27 +225,51 @@ public class TestFSAppAttempt extends FairSchedulerTestBase {
Mockito.when(mockScheduler.getRootQueueMetrics()).thenReturn
(fakeRootQueueMetrics);
int minClusterAvailableMemory = 2048;
int minClusterAvailableCPU = 6;
int minQueueAvailableCPU = 3;
ApplicationAttemptId applicationAttemptId = createAppAttemptId(1, 1);
RMContext rmContext = resourceManager.getRMContext();
FSAppAttempt schedulerApp =
new FSAppAttempt(mockScheduler, applicationAttemptId, "user1", mockQueue ,
null, rmContext);
// Min of Memory and CPU across cluster and queue is used in
// DominantResourceFairnessPolicy
Mockito.when(mockQueue.getPolicy()).thenReturn(SchedulingPolicy
.getInstance(DominantResourceFairnessPolicy.class));
verifyHeadroom(schedulerApp, minClusterAvailableMemory,
minQueueAvailableCPU);
verifyHeadroom(schedulerApp,
min(queueStarvation.getMemory(),
clusterAvailable.getMemory(),
queueMaxResourcesAvailable.getMemory()),
min(queueStarvation.getVirtualCores(),
clusterAvailable.getVirtualCores(),
queueMaxResourcesAvailable.getVirtualCores())
);
// Fair and Fifo ignore CPU of queue, so use cluster available CPU
Mockito.when(mockQueue.getPolicy()).thenReturn(SchedulingPolicy
.getInstance(FairSharePolicy.class));
verifyHeadroom(schedulerApp, minClusterAvailableMemory,
minClusterAvailableCPU);
verifyHeadroom(schedulerApp,
min(queueStarvation.getMemory(),
clusterAvailable.getMemory(),
queueMaxResourcesAvailable.getMemory()),
Math.min(
clusterAvailable.getVirtualCores(),
queueMaxResourcesAvailable.getVirtualCores())
);
Mockito.when(mockQueue.getPolicy()).thenReturn(SchedulingPolicy
.getInstance(FifoPolicy.class));
verifyHeadroom(schedulerApp, minClusterAvailableMemory,
minClusterAvailableCPU);
verifyHeadroom(schedulerApp,
min(queueStarvation.getMemory(),
clusterAvailable.getMemory(),
queueMaxResourcesAvailable.getMemory()),
Math.min(
clusterAvailable.getVirtualCores(),
queueMaxResourcesAvailable.getVirtualCores())
);
}
private static int min(int value1, int value2, int value3) {
return Math.min(Math.min(value1, value2), value3);
}
protected void verifyHeadroom(FSAppAttempt schedulerApp,