You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by as...@apache.org on 2017/02/24 07:24:23 UTC
[3/3] hadoop git commit: YARN-5959. RM changes to support change of
container ExecutionType. (Arun Suresh via wangda)
YARN-5959. RM changes to support change of container ExecutionType. (Arun Suresh via wangda)
(cherry picked from commit 0a55bd841ec0f2eb89a0383f4c589526e8b138d4)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/650ff95e
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/650ff95e
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/650ff95e
Branch: refs/heads/branch-2
Commit: 650ff95e00abb221c6f6afe64a36cf4d613d5cd2
Parents: 40bc9e7
Author: Wangda Tan <wa...@apache.org>
Authored: Thu Jan 5 10:31:05 2017 -0800
Committer: Arun Suresh <as...@apache.org>
Committed: Thu Feb 23 17:22:25 2017 -0800
----------------------------------------------------------------------
.../v2/app/rm/TestRMContainerAllocator.java | 12 +-
.../sls/scheduler/ResourceSchedulerWrapper.java | 9 +-
.../yarn/api/records/ContainerUpdateType.java | 9 +-
.../yarn/api/records/UpdateContainerError.java | 6 +
.../api/records/UpdateContainerRequest.java | 10 +
.../src/main/proto/yarn_service_protos.proto | 3 +-
.../api/impl/TestAMRMClientOnRMRestart.java | 11 +-
.../OpportunisticContainerAllocator.java | 16 +-
.../OpportunisticContainerContext.java | 8 +-
.../server/scheduler/SchedulerRequestKey.java | 50 +-
.../ApplicationMasterService.java | 57 ++-
...pportunisticContainerAllocatorAMService.java | 19 +-
.../server/resourcemanager/RMServerUtils.java | 142 +++---
.../rmapp/attempt/RMAppAttemptImpl.java | 6 +-
.../rmcontainer/RMContainerImpl.java | 32 +-
.../scheduler/AbstractYarnScheduler.java | 104 ++++-
.../resourcemanager/scheduler/Allocation.java | 26 +-
.../scheduler/AppSchedulingInfo.java | 70 +--
.../scheduler/ContainerUpdateContext.java | 265 +++++++++++
.../scheduler/ContainerUpdates.java | 68 +++
.../scheduler/SchedulerApplicationAttempt.java | 170 ++++++-
.../scheduler/SchedulerUtils.java | 22 +
.../scheduler/YarnScheduler.java | 6 +-
.../scheduler/capacity/CapacityScheduler.java | 16 +-
.../allocator/RegularContainerAllocator.java | 2 +-
.../scheduler/common/fica/FiCaSchedulerApp.java | 13 +-
.../scheduler/fair/FSAppAttempt.java | 4 +-
.../scheduler/fair/FairScheduler.java | 14 +-
.../scheduler/fifo/FifoAppAttempt.java | 4 +-
.../scheduler/fifo/FifoScheduler.java | 5 +-
.../LocalitySchedulingPlacementSet.java | 32 +-
.../placement/SchedulingPlacementSet.java | 5 +-
.../server/resourcemanager/Application.java | 4 +-
.../yarn/server/resourcemanager/MockAM.java | 7 +
.../yarn/server/resourcemanager/MockNM.java | 6 +
.../resourcemanager/TestClientRMService.java | 5 +-
...pportunisticContainerAllocatorAMService.java | 456 ++++++++++++++++++-
.../attempt/TestRMAppAttemptTransitions.java | 19 +-
.../rmcontainer/TestRMContainerImpl.java | 7 +-
.../capacity/TestCapacityScheduler.java | 37 +-
.../scheduler/capacity/TestChildQueueOrder.java | 5 +-
.../scheduler/capacity/TestReservations.java | 9 +-
.../scheduler/fair/FairSchedulerTestBase.java | 14 +-
.../fair/TestContinuousScheduling.java | 9 +-
.../scheduler/fair/TestFairScheduler.java | 33 +-
.../scheduler/fifo/TestFifoScheduler.java | 46 +-
46 files changed, 1586 insertions(+), 287 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/650ff95e/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/rm/TestRMContainerAllocator.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/rm/TestRMContainerAllocator.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/rm/TestRMContainerAllocator.java
index ce8d777..78b389a 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/rm/TestRMContainerAllocator.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/rm/TestRMContainerAllocator.java
@@ -108,7 +108,6 @@ import org.apache.hadoop.yarn.api.records.NodeReport;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
-import org.apache.hadoop.yarn.api.records.UpdateContainerRequest;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.event.Dispatcher;
import org.apache.hadoop.yarn.event.DrainDispatcher;
@@ -130,6 +129,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Allocation;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ContainerUpdates;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEvent;
@@ -1705,8 +1705,7 @@ public class TestRMContainerAllocator {
ApplicationAttemptId applicationAttemptId, List<ResourceRequest> ask,
List<ContainerId> release, List<String> blacklistAdditions,
List<String> blacklistRemovals,
- List<UpdateContainerRequest> increaseRequests,
- List<UpdateContainerRequest> decreaseRequests) {
+ ContainerUpdates updateRequests) {
List<ResourceRequest> askCopy = new ArrayList<ResourceRequest>();
for (ResourceRequest req : ask) {
ResourceRequest reqCopy = ResourceRequest.newInstance(req
@@ -1721,7 +1720,7 @@ public class TestRMContainerAllocator {
lastBlacklistRemovals = blacklistRemovals;
Allocation allocation = super.allocate(
applicationAttemptId, askCopy, release, blacklistAdditions,
- blacklistRemovals, increaseRequests, decreaseRequests);
+ blacklistRemovals, updateRequests);
if (forceResourceLimit != null) {
// Test wants to force the non-default resource limit
allocation.setResourceLimit(forceResourceLimit);
@@ -1752,8 +1751,7 @@ public class TestRMContainerAllocator {
ApplicationAttemptId applicationAttemptId, List<ResourceRequest> ask,
List<ContainerId> release, List<String> blacklistAdditions,
List<String> blacklistRemovals,
- List<UpdateContainerRequest> increaseRequest,
- List<UpdateContainerRequest> decreaseRequests) {
+ ContainerUpdates updateRequests) {
List<ResourceRequest> askCopy = new ArrayList<ResourceRequest>();
for (ResourceRequest req : ask) {
ResourceRequest reqCopy = ResourceRequest.newInstance(req
@@ -1764,7 +1762,7 @@ public class TestRMContainerAllocator {
SecurityUtil.setTokenServiceUseIp(false);
Allocation normalAlloc = super.allocate(
applicationAttemptId, askCopy, release,
- blacklistAdditions, blacklistRemovals, null, null);
+ blacklistAdditions, blacklistRemovals, updateRequests);
List<Container> containers = normalAlloc.getContainers();
if(containers.size() > 0) {
// allocate excess container
http://git-wip-us.apache.org/repos/asf/hadoop/blob/650ff95e/hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/scheduler/ResourceSchedulerWrapper.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/scheduler/ResourceSchedulerWrapper.java b/hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/scheduler/ResourceSchedulerWrapper.java
index 5d7e471..4c5b1f5 100644
--- a/hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/scheduler/ResourceSchedulerWrapper.java
+++ b/hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/scheduler/ResourceSchedulerWrapper.java
@@ -59,7 +59,6 @@ import org.apache.hadoop.yarn.api.records.QueueInfo;
import org.apache.hadoop.yarn.api.records.QueueUserACLInfo;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
-import org.apache.hadoop.yarn.api.records.UpdateContainerRequest;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore;
@@ -68,6 +67,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEven
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.UpdatedContainerInfo;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AbstractYarnScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Allocation;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ContainerUpdates;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedContainerChangeRequest;
@@ -207,14 +207,13 @@ public class ResourceSchedulerWrapper
public Allocation allocate(ApplicationAttemptId attemptId,
List<ResourceRequest> resourceRequests, List<ContainerId> containerIds,
List<String> strings, List<String> strings2,
- List<UpdateContainerRequest> increaseRequests,
- List<UpdateContainerRequest> decreaseRequests) {
+ ContainerUpdates updateRequests) {
if (metricsON) {
final Timer.Context context = schedulerAllocateTimer.time();
Allocation allocation = null;
try {
allocation = scheduler.allocate(attemptId, resourceRequests,
- containerIds, strings, strings2, null, null);
+ containerIds, strings, strings2, updateRequests);
return allocation;
} finally {
context.stop();
@@ -228,7 +227,7 @@ public class ResourceSchedulerWrapper
}
} else {
return scheduler.allocate(attemptId,
- resourceRequests, containerIds, strings, strings2, null, null);
+ resourceRequests, containerIds, strings, strings2, updateRequests);
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/650ff95e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerUpdateType.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerUpdateType.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerUpdateType.java
index 978ea09..6109cdb 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerUpdateType.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerUpdateType.java
@@ -39,7 +39,12 @@ public enum ContainerUpdateType {
DECREASE_RESOURCE,
/**
- * Execution Type change.
+ * Execution Type promotion.
*/
- UPDATE_EXECUTION_TYPE
+ PROMOTE_EXECUTION_TYPE,
+
+ /**
+ * Execution Type demotion.
+ */
+ DEMOTE_EXECUTION_TYPE
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/650ff95e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/UpdateContainerError.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/UpdateContainerError.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/UpdateContainerError.java
index 7102f7b..e7458cf 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/UpdateContainerError.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/UpdateContainerError.java
@@ -87,6 +87,12 @@ public abstract class UpdateContainerError {
}
@Override
+ public String toString() {
+ return "UpdateContainerError{reason=" + getReason() + ", "
+ + "req=" + getUpdateContainerRequest() + "}";
+ }
+
+ @Override
public boolean equals(Object obj) {
if (this == obj) {
return true;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/650ff95e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/UpdateContainerRequest.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/UpdateContainerRequest.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/UpdateContainerRequest.java
index 2a8bb79..2060370 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/UpdateContainerRequest.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/UpdateContainerRequest.java
@@ -166,11 +166,13 @@ public abstract class UpdateContainerRequest {
ContainerId cId = getContainerId();
ExecutionType execType = getExecutionType();
Resource capability = getCapability();
+ ContainerUpdateType updateType = getContainerUpdateType();
result =
prime * result + ((capability == null) ? 0 : capability.hashCode());
result = prime * result + ((cId == null) ? 0 : cId.hashCode());
result = prime * result + getContainerVersion();
result = prime * result + ((execType == null) ? 0 : execType.hashCode());
+ result = prime * result + ((updateType== null) ? 0 : updateType.hashCode());
return result;
}
@@ -224,6 +226,14 @@ public abstract class UpdateContainerRequest {
} else if (!execType.equals(other.getExecutionType())) {
return false;
}
+ ContainerUpdateType updateType = getContainerUpdateType();
+ if (updateType == null) {
+ if (other.getContainerUpdateType() != null) {
+ return false;
+ }
+ } else if (!updateType.equals(other.getContainerUpdateType())) {
+ return false;
+ }
return true;
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/650ff95e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto
index c5c9e9e..a95b846 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto
@@ -63,7 +63,8 @@ message FinishApplicationMasterResponseProto {
enum ContainerUpdateTypeProto {
INCREASE_RESOURCE = 0;
DECREASE_RESOURCE = 1;
- UPDATE_EXECUTION_TYPE = 2;
+ PROMOTE_EXECUTION_TYPE = 2;
+ DEMOTE_EXECUTION_TYPE = 3;
}
message UpdateContainerRequestProto {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/650ff95e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClientOnRMRestart.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClientOnRMRestart.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClientOnRMRestart.java
index f1c49f2..ac77446 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClientOnRMRestart.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClientOnRMRestart.java
@@ -66,6 +66,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Allocation;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ContainerUpdates;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fifo.FifoScheduler;
@@ -574,8 +575,7 @@ public class TestAMRMClientOnRMRestart {
ApplicationAttemptId applicationAttemptId, List<ResourceRequest> ask,
List<ContainerId> release, List<String> blacklistAdditions,
List<String> blacklistRemovals,
- List<UpdateContainerRequest> increaseRequests,
- List<UpdateContainerRequest> decreaseRequests) {
+ ContainerUpdates updateRequests) {
List<ResourceRequest> askCopy = new ArrayList<ResourceRequest>();
for (ResourceRequest req : ask) {
ResourceRequest reqCopy =
@@ -586,13 +586,12 @@ public class TestAMRMClientOnRMRestart {
}
lastAsk = ask;
lastRelease = release;
- lastIncrease = increaseRequests;
- lastDecrease = decreaseRequests;
+ lastIncrease = updateRequests.getIncreaseRequests();
+ lastDecrease = updateRequests.getDecreaseRequests();
lastBlacklistAdditions = blacklistAdditions;
lastBlacklistRemovals = blacklistRemovals;
return super.allocate(applicationAttemptId, askCopy, release,
- blacklistAdditions, blacklistRemovals, increaseRequests,
- decreaseRequests);
+ blacklistAdditions, blacklistRemovals, updateRequests);
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/650ff95e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/OpportunisticContainerAllocator.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/OpportunisticContainerAllocator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/OpportunisticContainerAllocator.java
index c1300b2..6fd5228 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/OpportunisticContainerAllocator.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/OpportunisticContainerAllocator.java
@@ -321,13 +321,21 @@ public class OpportunisticContainerAllocator {
// before accepting an ask)
Resource capability = normalizeCapability(appParams, rr);
+ return createContainer(
+ rmIdentifier, appParams.getContainerTokenExpiryInterval(),
+ SchedulerRequestKey.create(rr), userName, node, cId, capability);
+ }
+
+ private Container createContainer(long rmIdentifier, long tokenExpiry,
+ SchedulerRequestKey schedulerKey, String userName, RemoteNode node,
+ ContainerId cId, Resource capability) {
long currTime = System.currentTimeMillis();
ContainerTokenIdentifier containerTokenIdentifier =
new ContainerTokenIdentifier(
cId, 0, node.getNodeId().toString(), userName,
- capability, currTime + appParams.containerTokenExpiryInterval,
+ capability, currTime + tokenExpiry,
tokenSecretManager.getCurrentKey().getKeyId(), rmIdentifier,
- rr.getPriority(), currTime,
+ schedulerKey.getPriority(), currTime,
null, CommonNodeLabelsManager.NO_LABEL, ContainerType.TASK,
ExecutionType.OPPORTUNISTIC);
byte[] pwd =
@@ -336,9 +344,9 @@ public class OpportunisticContainerAllocator {
containerTokenIdentifier);
Container container = BuilderUtils.newContainer(
cId, node.getNodeId(), node.getHttpAddress(),
- capability, rr.getPriority(), containerToken,
+ capability, schedulerKey.getPriority(), containerToken,
containerTokenIdentifier.getExecutionType(),
- rr.getAllocationRequestId());
+ schedulerKey.getAllocationRequestId());
return container;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/650ff95e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/OpportunisticContainerContext.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/OpportunisticContainerContext.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/OpportunisticContainerContext.java
index a2f9f4d..1b1c5b9 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/OpportunisticContainerContext.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/OpportunisticContainerContext.java
@@ -150,8 +150,9 @@ public class OpportunisticContainerContext {
resourceRequest.getNumContainers() + request.getNumContainers());
}
if (ResourceRequest.isAnyLocation(request.getResourceName())) {
- LOG.info("# of outstandingOpReqs in ANY (at" +
- "priority = "+ schedulerKey.getPriority()
+ LOG.info("# of outstandingOpReqs in ANY (at "
+ + "priority = " + schedulerKey.getPriority()
+ + ", allocationReqId = " + schedulerKey.getAllocationRequestId()
+ ", with capability = " + request.getCapability() + " ) : "
+ resourceRequest.getNumContainers());
}
@@ -167,7 +168,8 @@ public class OpportunisticContainerContext {
public void matchAllocationToOutstandingRequest(Resource capability,
List<Container> allocatedContainers) {
for (Container c : allocatedContainers) {
- SchedulerRequestKey schedulerKey = SchedulerRequestKey.extractFrom(c);
+ SchedulerRequestKey schedulerKey =
+ SchedulerRequestKey.extractFrom(c);
Map<Resource, ResourceRequest> asks =
outstandingOpReqs.get(schedulerKey);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/650ff95e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/SchedulerRequestKey.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/SchedulerRequestKey.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/SchedulerRequestKey.java
index 9b7edbe..02539ba 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/SchedulerRequestKey.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/SchedulerRequestKey.java
@@ -19,8 +19,10 @@
package org.apache.hadoop.yarn.server.scheduler;
import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
+import org.apache.hadoop.yarn.api.records.UpdateContainerRequest;
/**
* Composite key for outstanding scheduler requests for any schedulable entity.
@@ -31,6 +33,7 @@ public final class SchedulerRequestKey implements
private final Priority priority;
private final long allocationRequestId;
+ private final ContainerId containerToUpdate;
/**
* Factory method to generate a SchedulerRequestKey from a ResourceRequest.
@@ -39,7 +42,13 @@ public final class SchedulerRequestKey implements
*/
public static SchedulerRequestKey create(ResourceRequest req) {
return new SchedulerRequestKey(req.getPriority(),
- req.getAllocationRequestId());
+ req.getAllocationRequestId(), null);
+ }
+
+ public static SchedulerRequestKey create(UpdateContainerRequest req,
+ SchedulerRequestKey schedulerRequestKey) {
+ return new SchedulerRequestKey(schedulerRequestKey.getPriority(),
+ schedulerRequestKey.getAllocationRequestId(), req.getContainerId());
}
/**
@@ -50,12 +59,16 @@ public final class SchedulerRequestKey implements
*/
public static SchedulerRequestKey extractFrom(Container container) {
return new SchedulerRequestKey(container.getPriority(),
- container.getAllocationRequestId());
+ container.getAllocationRequestId(), null);
}
- SchedulerRequestKey(Priority priority, long allocationRequestId) {
+
+
+ public SchedulerRequestKey(Priority priority, long allocationRequestId,
+ ContainerId containerToUpdate) {
this.priority = priority;
this.allocationRequestId = allocationRequestId;
+ this.containerToUpdate = containerToUpdate;
}
/**
@@ -76,6 +89,10 @@ public final class SchedulerRequestKey implements
return allocationRequestId;
}
+ public ContainerId getContainerToUpdate() {
+ return containerToUpdate;
+ }
+
@Override
public int compareTo(SchedulerRequestKey o) {
if (o == null) {
@@ -85,6 +102,15 @@ public final class SchedulerRequestKey implements
return 1;
}
}
+
+ // Ensure updates are ranked higher
+ if (this.containerToUpdate == null && o.containerToUpdate != null) {
+ return -1;
+ }
+ if (this.containerToUpdate != null && o.containerToUpdate == null) {
+ return 1;
+ }
+
int priorityCompare = o.getPriority().compareTo(priority);
// we first sort by priority and then by allocationRequestId
if (priorityCompare != 0) {
@@ -107,16 +133,21 @@ public final class SchedulerRequestKey implements
if (getAllocationRequestId() != that.getAllocationRequestId()) {
return false;
}
- return getPriority() != null ?
- getPriority().equals(that.getPriority()) :
- that.getPriority() == null;
+ if (!getPriority().equals(that.getPriority())) {
+ return false;
+ }
+ return containerToUpdate != null ?
+ containerToUpdate.equals(that.containerToUpdate) :
+ that.containerToUpdate == null;
}
@Override
public int hashCode() {
- int result = getPriority() != null ? getPriority().hashCode() : 0;
- result = 31 * result + (int) (getAllocationRequestId() ^ (
- getAllocationRequestId() >>> 32));
+ int result = priority != null ? priority.hashCode() : 0;
+ result = 31 * result + (int) (allocationRequestId ^ (allocationRequestId
+ >>> 32));
+ result = 31 * result + (containerToUpdate != null ? containerToUpdate
+ .hashCode() : 0);
return result;
}
@@ -125,6 +156,7 @@ public final class SchedulerRequestKey implements
return "SchedulerRequestKey{" +
"priority=" + priority +
", allocationRequestId=" + allocationRequestId +
+ ", containerToUpdate=" + containerToUpdate +
'}';
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/650ff95e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ApplicationMasterService.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ApplicationMasterService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ApplicationMasterService.java
index c2b0012..a76cd0a 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ApplicationMasterService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ApplicationMasterService.java
@@ -68,7 +68,6 @@ import org.apache.hadoop.yarn.api.records.ResourceBlacklistRequest;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
import org.apache.hadoop.yarn.api.records.StrictPreemptionContract;
import org.apache.hadoop.yarn.api.records.UpdateContainerError;
-import org.apache.hadoop.yarn.api.records.UpdateContainerRequest;
import org.apache.hadoop.yarn.api.records.UpdatedContainer;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.ApplicationAttemptNotFoundException;
@@ -93,7 +92,12 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAt
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptUnregistrationEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler
+ .AbstractYarnScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Allocation;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ContainerUpdates;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler
+ .SchedulerApplicationAttempt;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNodeReport;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.YarnScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.security
@@ -554,12 +558,10 @@ public class ApplicationMasterService extends AbstractService implements
// Split Update Resource Requests into increase and decrease.
// No Exceptions are thrown here. All update errors are aggregated
// and returned to the AM.
- List<UpdateContainerRequest> increaseResourceReqs = new ArrayList<>();
- List<UpdateContainerRequest> decreaseResourceReqs = new ArrayList<>();
- List<UpdateContainerError> updateContainerErrors =
+ List<UpdateContainerError> updateErrors = new ArrayList<>();
+ ContainerUpdates containerUpdateRequests =
RMServerUtils.validateAndSplitUpdateResourceRequests(
- rmContext, request, maximumCapacity,
- increaseResourceReqs, decreaseResourceReqs);
+ rmContext, request, maximumCapacity, updateErrors);
// Send new requests to appAttempt.
Allocation allocation;
@@ -575,7 +577,7 @@ public class ApplicationMasterService extends AbstractService implements
allocation =
this.rScheduler.allocate(appAttemptId, ask, release,
blacklistAdditions, blacklistRemovals,
- increaseResourceReqs, decreaseResourceReqs);
+ containerUpdateRequests);
}
if (!blacklistAdditions.isEmpty() || !blacklistRemovals.isEmpty()) {
@@ -591,7 +593,7 @@ public class ApplicationMasterService extends AbstractService implements
}
// Notify the AM of container update errors
- addToUpdateContainerErrors(allocateResponse, updateContainerErrors);
+ addToUpdateContainerErrors(allocateResponse, updateErrors);
// update the response with the deltas of node status changes
List<RMNode> updatedNodes = new ArrayList<RMNode>();
@@ -625,6 +627,21 @@ public class ApplicationMasterService extends AbstractService implements
.pullJustFinishedContainers());
allocateResponse.setAvailableResources(allocation.getResourceLimit());
+ addToContainerUpdates(appAttemptId, allocateResponse, allocation);
+
+ allocateResponse.setNumClusterNodes(this.rScheduler.getNumClusterNodes());
+
+ // add preemption to the allocateResponse message (if any)
+ allocateResponse
+ .setPreemptionMessage(generatePreemptionMessage(allocation));
+
+ // Set application priority
+ allocateResponse.setApplicationPriority(app
+ .getApplicationPriority());
+ }
+
+ private void addToContainerUpdates(ApplicationAttemptId appAttemptId,
+ AllocateResponse allocateResponse, Allocation allocation) {
// Handling increased containers
addToUpdatedContainers(
allocateResponse, ContainerUpdateType.INCREASE_RESOURCE,
@@ -635,15 +652,23 @@ public class ApplicationMasterService extends AbstractService implements
allocateResponse, ContainerUpdateType.DECREASE_RESOURCE,
allocation.getDecreasedContainers());
- allocateResponse.setNumClusterNodes(this.rScheduler.getNumClusterNodes());
-
- // add preemption to the allocateResponse message (if any)
- allocateResponse
- .setPreemptionMessage(generatePreemptionMessage(allocation));
+ // Handling promoted containers
+ addToUpdatedContainers(
+ allocateResponse, ContainerUpdateType.PROMOTE_EXECUTION_TYPE,
+ allocation.getPromotedContainers());
- // Set application priority
- allocateResponse.setApplicationPriority(app
- .getApplicationPriority());
+ // Handling demoted containers
+ addToUpdatedContainers(
+ allocateResponse, ContainerUpdateType.DEMOTE_EXECUTION_TYPE,
+ allocation.getDemotedContainers());
+
+ SchedulerApplicationAttempt applicationAttempt = ((AbstractYarnScheduler)
+ rScheduler).getApplicationAttempt(appAttemptId);
+ if (applicationAttempt != null) {
+ addToUpdateContainerErrors(allocateResponse,
+ ((AbstractYarnScheduler)rScheduler)
+ .getApplicationAttempt(appAttemptId).pullUpdateContainerErrors());
+ }
}
protected void addToUpdateContainerErrors(AllocateResponse allocateResponse,
http://git-wip-us.apache.org/repos/asf/hadoop/blob/650ff95e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/OpportunisticContainerAllocatorAMService.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/OpportunisticContainerAllocatorAMService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/OpportunisticContainerAllocatorAMService.java
index 708b481..c8ee0c6 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/OpportunisticContainerAllocatorAMService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/OpportunisticContainerAllocatorAMService.java
@@ -32,7 +32,6 @@ import org.apache.hadoop.yarn.security.AMRMTokenIdentifier;
import org.apache.hadoop.yarn.server.api.DistributedSchedulingAMProtocol;
import org.apache.hadoop.yarn.api.impl.pb.service.ApplicationMasterProtocolPBServiceImpl;
-
import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest;
import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
import org.apache.hadoop.yarn.server.api.protocolrecords.DistributedSchedulingAllocateRequest;
@@ -48,15 +47,14 @@ import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.ipc.YarnRPC;
import org.apache.hadoop.yarn.proto.ApplicationMasterProtocol.ApplicationMasterProtocolService;
-
import org.apache.hadoop.yarn.server.api.protocolrecords.RemoteNode;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType;
-import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerImpl;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AbstractYarnScheduler;
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.YarnScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.distributed.NodeQueueLoadMonitor;
@@ -69,9 +67,9 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeUpdateS
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.security.AMRMTokenSecretManager;
-
import org.apache.hadoop.yarn.server.scheduler.OpportunisticContainerAllocator;
import org.apache.hadoop.yarn.server.scheduler.OpportunisticContainerContext;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
import org.apache.hadoop.yarn.server.utils.YarnServerSecurityUtils;
import java.io.IOException;
@@ -251,6 +249,7 @@ public class OpportunisticContainerAllocatorAMService
// Allocate GUARANTEED containers.
request.setAskList(partitionedAsks.getGuaranteed());
+
super.allocateInternal(appAttemptId, request, allocateResponse);
}
@@ -298,15 +297,9 @@ public class OpportunisticContainerAllocatorAMService
boolean isRemotelyAllocated) {
for (Container container : allocContainers) {
// Create RMContainer
- SchedulerApplicationAttempt appAttempt =
- ((AbstractYarnScheduler) rmContext.getScheduler())
- .getCurrentAttemptForContainer(container.getId());
- RMContainer rmContainer = new RMContainerImpl(container,
- appAttempt.getApplicationAttemptId(), container.getNodeId(),
- appAttempt.getUser(), rmContext, isRemotelyAllocated);
- appAttempt.addRMContainer(container.getId(), rmContainer);
- ((AbstractYarnScheduler) rmContext.getScheduler()).getNode(
- container.getNodeId()).allocateContainer(rmContainer);
+ RMContainer rmContainer =
+ SchedulerUtils.createOpportunisticRmContainer(
+ rmContext, container, isRemotelyAllocated);
rmContainer.handle(
new RMContainerEvent(container.getId(),
RMContainerEventType.ACQUIRED));
http://git-wip-us.apache.org/repos/asf/hadoop/blob/650ff95e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMServerUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMServerUtils.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMServerUtils.java
index f91259e..49aa3b0 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMServerUtils.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMServerUtils.java
@@ -39,6 +39,8 @@ import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationResourceUsageReport;
import org.apache.hadoop.yarn.api.records.ApplicationTimeoutType;
import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.ContainerUpdateType;
+import org.apache.hadoop.yarn.api.records.ExecutionType;
import org.apache.hadoop.yarn.api.records.NodeState;
import org.apache.hadoop.yarn.api.records.QueueInfo;
import org.apache.hadoop.yarn.api.records.Resource;
@@ -64,6 +66,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt
.RMAppAttemptState;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ContainerUpdates;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler
.ResourceScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler
@@ -81,7 +84,7 @@ import org.apache.hadoop.yarn.util.resource.Resources;
*/
public class RMServerUtils {
- private static final String UPDATE_OUTSTANDING_ERROR =
+ public static final String UPDATE_OUTSTANDING_ERROR =
"UPDATE_OUTSTANDING_ERROR";
private static final String INCORRECT_CONTAINER_VERSION_ERROR =
"INCORRECT_CONTAINER_VERSION_ERROR";
@@ -125,74 +128,105 @@ public class RMServerUtils {
/**
* Check if we have:
- * - Request for same containerId and different target resource
- * - If targetResources violates maximum/minimumAllocation
- * @param rmContext RM context
- * @param request Allocate Request
- * @param maximumAllocation Maximum Allocation
- * @param increaseResourceReqs Increase Resource Request
- * @param decreaseResourceReqs Decrease Resource Request
- * @return List of container Errors
+ * - Request for same containerId and different target resource.
+ * - If targetResources violates maximum/minimumAllocation.
+ * @param rmContext RM context.
+ * @param request Allocate Request.
+ * @param maximumAllocation Maximum Allocation.
+ * @param updateErrors Container update errors.
+ * @return ContainerUpdateRequests.
*/
- public static List<UpdateContainerError>
+ public static ContainerUpdates
validateAndSplitUpdateResourceRequests(RMContext rmContext,
AllocateRequest request, Resource maximumAllocation,
- List<UpdateContainerRequest> increaseResourceReqs,
- List<UpdateContainerRequest> decreaseResourceReqs) {
- List<UpdateContainerError> errors = new ArrayList<>();
+ List<UpdateContainerError> updateErrors) {
+ ContainerUpdates updateRequests =
+ new ContainerUpdates();
Set<ContainerId> outstandingUpdate = new HashSet<>();
for (UpdateContainerRequest updateReq : request.getUpdateRequests()) {
RMContainer rmContainer = rmContext.getScheduler().getRMContainer(
updateReq.getContainerId());
- String msg = null;
- if (rmContainer == null) {
- msg = INVALID_CONTAINER_ID;
- }
- // Only allow updates if the requested version matches the current
- // version
- if (msg == null && updateReq.getContainerVersion() !=
- rmContainer.getContainer().getVersion()) {
- msg = INCORRECT_CONTAINER_VERSION_ERROR + "|"
- + updateReq.getContainerVersion() + "|"
- + rmContainer.getContainer().getVersion();
- }
- // No more than 1 container update per request.
- if (msg == null &&
- outstandingUpdate.contains(updateReq.getContainerId())) {
- msg = UPDATE_OUTSTANDING_ERROR;
- }
+ String msg = validateContainerIdAndVersion(outstandingUpdate,
+ updateReq, rmContainer);
+ ContainerUpdateType updateType = updateReq.getContainerUpdateType();
if (msg == null) {
- Resource original = rmContainer.getContainer().getResource();
- Resource target = updateReq.getCapability();
- if (Resources.fitsIn(target, original)) {
- // This is a decrease request
- if (validateIncreaseDecreaseRequest(rmContext, updateReq,
- maximumAllocation, false)) {
- decreaseResourceReqs.add(updateReq);
- outstandingUpdate.add(updateReq.getContainerId());
+ if ((updateType != ContainerUpdateType.PROMOTE_EXECUTION_TYPE) &&
+ (updateType !=ContainerUpdateType.DEMOTE_EXECUTION_TYPE)) {
+ Resource original = rmContainer.getContainer().getResource();
+ Resource target = updateReq.getCapability();
+ if (Resources.fitsIn(target, original)) {
+ // This is a decrease request
+ if (validateIncreaseDecreaseRequest(rmContext, updateReq,
+ maximumAllocation, false)) {
+ updateRequests.getDecreaseRequests().add(updateReq);
+ outstandingUpdate.add(updateReq.getContainerId());
+ } else {
+ msg = RESOURCE_OUTSIDE_ALLOWED_RANGE;
+ }
} else {
- msg = RESOURCE_OUTSIDE_ALLOWED_RANGE;
+ // This is an increase request
+ if (validateIncreaseDecreaseRequest(rmContext, updateReq,
+ maximumAllocation, true)) {
+ updateRequests.getIncreaseRequests().add(updateReq);
+ outstandingUpdate.add(updateReq.getContainerId());
+ } else {
+ msg = RESOURCE_OUTSIDE_ALLOWED_RANGE;
+ }
}
} else {
- // This is an increase request
- if (validateIncreaseDecreaseRequest(rmContext, updateReq,
- maximumAllocation, true)) {
- increaseResourceReqs.add(updateReq);
- outstandingUpdate.add(updateReq.getContainerId());
- } else {
- msg = RESOURCE_OUTSIDE_ALLOWED_RANGE;
+ ExecutionType original = rmContainer.getExecutionType();
+ ExecutionType target = updateReq.getExecutionType();
+ if (target != original) {
+ if (target == ExecutionType.GUARANTEED &&
+ original == ExecutionType.OPPORTUNISTIC) {
+ updateRequests.getPromotionRequests().add(updateReq);
+ outstandingUpdate.add(updateReq.getContainerId());
+ } else if (target == ExecutionType.OPPORTUNISTIC &&
+ original == ExecutionType.GUARANTEED) {
+ updateRequests.getDemotionRequests().add(updateReq);
+ outstandingUpdate.add(updateReq.getContainerId());
+ }
}
}
}
- if (msg != null) {
- UpdateContainerError updateError = RECORD_FACTORY
- .newRecordInstance(UpdateContainerError.class);
- updateError.setReason(msg);
- updateError.setUpdateContainerRequest(updateReq);
- errors.add(updateError);
- }
+ checkAndcreateUpdateError(updateErrors, updateReq, msg);
+ }
+ return updateRequests;
+ }
+
+ private static void checkAndcreateUpdateError(
+ List<UpdateContainerError> errors, UpdateContainerRequest updateReq,
+ String msg) {
+ if (msg != null) {
+ UpdateContainerError updateError = RECORD_FACTORY
+ .newRecordInstance(UpdateContainerError.class);
+ updateError.setReason(msg);
+ updateError.setUpdateContainerRequest(updateReq);
+ errors.add(updateError);
+ }
+ }
+
+ private static String validateContainerIdAndVersion(
+ Set<ContainerId> outstandingUpdate, UpdateContainerRequest updateReq,
+ RMContainer rmContainer) {
+ String msg = null;
+ if (rmContainer == null) {
+ msg = INVALID_CONTAINER_ID;
+ }
+ // Only allow updates if the requested version matches the current
+ // version
+ if (msg == null && updateReq.getContainerVersion() !=
+ rmContainer.getContainer().getVersion()) {
+ msg = INCORRECT_CONTAINER_VERSION_ERROR + "|"
+ + updateReq.getContainerVersion() + "|"
+ + rmContainer.getContainer().getVersion();
+ }
+ // No more than 1 container update per request.
+ if (msg == null &&
+ outstandingUpdate.contains(updateReq.getContainerId())) {
+ msg = UPDATE_OUTSTANDING_ERROR;
}
- return errors;
+ return msg;
}
/**
http://git-wip-us.apache.org/repos/asf/hadoop/blob/650ff95e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
index 185f5b2..da42ab6 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
@@ -96,6 +96,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerImpl
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeFinishedContainersPulledByAMEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AbstractYarnScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Allocation;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ContainerUpdates;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt.AMState;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
@@ -1123,7 +1124,8 @@ public class RMAppAttemptImpl implements RMAppAttempt, Recoverable {
Collections.singletonList(appAttempt.amReq),
EMPTY_CONTAINER_RELEASE_LIST,
amBlacklist.getBlacklistAdditions(),
- amBlacklist.getBlacklistRemovals(), null, null);
+ amBlacklist.getBlacklistRemovals(),
+ new ContainerUpdates());
if (amContainerAllocation != null
&& amContainerAllocation.getContainers() != null) {
assert (amContainerAllocation.getContainers().size() == 0);
@@ -1147,7 +1149,7 @@ public class RMAppAttemptImpl implements RMAppAttempt, Recoverable {
Allocation amContainerAllocation =
appAttempt.scheduler.allocate(appAttempt.applicationAttemptId,
EMPTY_CONTAINER_REQUEST_LIST, EMPTY_CONTAINER_RELEASE_LIST, null,
- null, null, null);
+ null, new ContainerUpdates());
// There must be at least one container allocated, because a
// CONTAINER_ALLOCATED is emitted after an RMContainer is constructed,
// and is put in SchedulerApplication#newlyAllocatedContainers.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/650ff95e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmcontainer/RMContainerImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmcontainer/RMContainerImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmcontainer/RMContainerImpl.java
index b647224..e5948de 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmcontainer/RMContainerImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmcontainer/RMContainerImpl.java
@@ -108,6 +108,8 @@ public class RMContainerImpl implements RMContainer, Comparable<RMContainer> {
// Transitions from ACQUIRED state
.addTransition(RMContainerState.ACQUIRED, RMContainerState.RUNNING,
RMContainerEventType.LAUNCHED)
+ .addTransition(RMContainerState.ACQUIRED, RMContainerState.ACQUIRED,
+ RMContainerEventType.ACQUIRED)
.addTransition(RMContainerState.ACQUIRED, RMContainerState.COMPLETED,
RMContainerEventType.FINISHED, new FinishedTransition())
.addTransition(RMContainerState.ACQUIRED, RMContainerState.RELEASED,
@@ -125,6 +127,8 @@ public class RMContainerImpl implements RMContainer, Comparable<RMContainer> {
.addTransition(RMContainerState.RUNNING, RMContainerState.RELEASED,
RMContainerEventType.RELEASED, new KillTransition())
.addTransition(RMContainerState.RUNNING, RMContainerState.RUNNING,
+ RMContainerEventType.ACQUIRED)
+ .addTransition(RMContainerState.RUNNING, RMContainerState.RUNNING,
RMContainerEventType.RESERVED, new ContainerReservedTransition())
.addTransition(RMContainerState.RUNNING, RMContainerState.RUNNING,
RMContainerEventType.CHANGE_RESOURCE, new ChangeResourceTransition())
@@ -163,13 +167,13 @@ public class RMContainerImpl implements RMContainer, Comparable<RMContainer> {
private final WriteLock writeLock;
private final ApplicationAttemptId appAttemptId;
private final NodeId nodeId;
- private final Container container;
private final RMContext rmContext;
private final EventHandler eventHandler;
private final ContainerAllocationExpirer containerAllocationExpirer;
private final String user;
private final String nodeLabelExpression;
+ private volatile Container container;
private Resource reservedResource;
private NodeId reservedNode;
private SchedulerRequestKey reservedSchedulerKey;
@@ -188,44 +192,44 @@ public class RMContainerImpl implements RMContainer, Comparable<RMContainer> {
private boolean isExternallyAllocated;
private SchedulerRequestKey allocatedSchedulerKey;
- public RMContainerImpl(Container container,
+ public RMContainerImpl(Container container, SchedulerRequestKey schedulerKey,
ApplicationAttemptId appAttemptId, NodeId nodeId, String user,
RMContext rmContext) {
- this(container, appAttemptId, nodeId, user, rmContext, System
+ this(container, schedulerKey, appAttemptId, nodeId, user, rmContext, System
.currentTimeMillis(), "");
}
- public RMContainerImpl(Container container,
+ public RMContainerImpl(Container container, SchedulerRequestKey schedulerKey,
ApplicationAttemptId appAttemptId, NodeId nodeId, String user,
RMContext rmContext, boolean isExternallyAllocated) {
- this(container, appAttemptId, nodeId, user, rmContext, System
+ this(container, schedulerKey, appAttemptId, nodeId, user, rmContext, System
.currentTimeMillis(), "", isExternallyAllocated);
}
private boolean saveNonAMContainerMetaInfo;
- public RMContainerImpl(Container container,
+ public RMContainerImpl(Container container, SchedulerRequestKey schedulerKey,
ApplicationAttemptId appAttemptId, NodeId nodeId, String user,
RMContext rmContext, String nodeLabelExpression) {
- this(container, appAttemptId, nodeId, user, rmContext, System
+ this(container, schedulerKey, appAttemptId, nodeId, user, rmContext, System
.currentTimeMillis(), nodeLabelExpression);
}
- public RMContainerImpl(Container container,
+ public RMContainerImpl(Container container, SchedulerRequestKey schedulerKey,
ApplicationAttemptId appAttemptId, NodeId nodeId, String user,
RMContext rmContext, long creationTime, String nodeLabelExpression) {
- this(container, appAttemptId, nodeId, user, rmContext, creationTime,
- nodeLabelExpression, false);
+ this(container, schedulerKey, appAttemptId, nodeId, user, rmContext,
+ creationTime, nodeLabelExpression, false);
}
- public RMContainerImpl(Container container,
+ public RMContainerImpl(Container container, SchedulerRequestKey schedulerKey,
ApplicationAttemptId appAttemptId, NodeId nodeId, String user,
RMContext rmContext, long creationTime, String nodeLabelExpression,
boolean isExternallyAllocated) {
this.stateMachine = stateMachineFactory.make(this);
this.nodeId = nodeId;
this.container = container;
- this.allocatedSchedulerKey = SchedulerRequestKey.extractFrom(container);
+ this.allocatedSchedulerKey = schedulerKey;
this.appAttemptId = appAttemptId;
this.user = user;
this.creationTime = creationTime;
@@ -276,6 +280,10 @@ public class RMContainerImpl implements RMContainer, Comparable<RMContainer> {
return this.container;
}
+ public void setContainer(Container container) {
+ this.container = container;
+ }
+
@Override
public RMContainerState getState() {
this.readLock.lock();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/650ff95e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java
index 3851d20..ce6d2a2 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java
@@ -51,6 +51,7 @@ import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceOption;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
+import org.apache.hadoop.yarn.api.records.UpdateContainerError;
import org.apache.hadoop.yarn.api.records.UpdateContainerRequest;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.InvalidResourceRequestException;
@@ -62,6 +63,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.RMAppManagerEventType;
import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger;
import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
+import org.apache.hadoop.yarn.server.resourcemanager.RMServerUtils;
import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEvent;
@@ -81,6 +83,11 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeResourceUpdate
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.UpdatedContainerInfo;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivitiesManager;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.QueueEntitlement;
+
+
+import org.apache.hadoop.yarn.server.scheduler.OpportunisticContainerContext;
+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.Clock;
import org.apache.hadoop.yarn.util.SystemClock;
@@ -509,9 +516,11 @@ public abstract class AbstractYarnScheduler
ApplicationAttemptId attemptId =
container.getId().getApplicationAttemptId();
RMContainer rmContainer =
- new RMContainerImpl(container, attemptId, node.getNodeID(),
- applications.get(attemptId.getApplicationId()).getUser(), rmContext,
- status.getCreationTime(), status.getNodeLabelExpression());
+ new RMContainerImpl(container,
+ SchedulerRequestKey.extractFrom(container), attemptId,
+ node.getNodeID(), applications.get(
+ attemptId.getApplicationId()).getUser(), rmContext,
+ status.getCreationTime(), status.getNodeLabelExpression());
return rmContainer;
}
@@ -1064,4 +1073,93 @@ public abstract class AbstractYarnScheduler
ask.setCapability(getNormalizedResource(ask.getCapability()));
}
}
+
+ protected void handleExecutionTypeUpdates(
+ SchedulerApplicationAttempt appAttempt,
+ List<UpdateContainerRequest> promotionRequests,
+ List<UpdateContainerRequest> demotionRequests) {
+ if (promotionRequests != null && !promotionRequests.isEmpty()) {
+ LOG.info("Promotion Update requests : " + promotionRequests);
+ handlePromotionRequests(appAttempt, promotionRequests);
+ }
+ if (demotionRequests != null && !demotionRequests.isEmpty()) {
+ LOG.info("Demotion Update requests : " + demotionRequests);
+ handleDemotionRequests(appAttempt, demotionRequests);
+ }
+ }
+
+ private void handlePromotionRequests(
+ SchedulerApplicationAttempt applicationAttempt,
+ List<UpdateContainerRequest> updateContainerRequests) {
+ for (UpdateContainerRequest uReq : updateContainerRequests) {
+ RMContainer rmContainer =
+ rmContext.getScheduler().getRMContainer(uReq.getContainerId());
+ // Check if this is a container update
+ // And not in the middle of a Demotion
+ if (rmContainer != null) {
+ // Check if this is an executionType change request
+ // If so, fix the rr to make it look like a normal rr
+ // with relaxLocality=false and numContainers=1
+ SchedulerNode schedulerNode = rmContext.getScheduler()
+ .getSchedulerNode(rmContainer.getContainer().getNodeId());
+
+ // Add only if no outstanding promote requests exist.
+ if (!applicationAttempt.getUpdateContext()
+ .checkAndAddToOutstandingIncreases(
+ rmContainer, schedulerNode, uReq)) {
+ applicationAttempt.addToUpdateContainerErrors(
+ UpdateContainerError.newInstance(
+ RMServerUtils.UPDATE_OUTSTANDING_ERROR, uReq));
+ }
+ } else {
+ LOG.warn("Cannot promote non-existent (or completed) Container ["
+ + uReq.getContainerId() + "]");
+ }
+ }
+ }
+
+ private void handleDemotionRequests(SchedulerApplicationAttempt appAttempt,
+ List<UpdateContainerRequest> demotionRequests) {
+ OpportunisticContainerContext oppCntxt =
+ appAttempt.getOpportunisticContainerContext();
+ for (UpdateContainerRequest uReq : demotionRequests) {
+ RMContainer rmContainer =
+ rmContext.getScheduler().getRMContainer(uReq.getContainerId());
+ if (rmContainer != null) {
+ if (appAttempt.getUpdateContext().checkAndAddToOutstandingDecreases(
+ rmContainer.getContainer())) {
+ RMContainer demotedRMContainer =
+ createDemotedRMContainer(appAttempt, oppCntxt, rmContainer);
+ appAttempt.addToNewlyDemotedContainers(
+ uReq.getContainerId(), demotedRMContainer);
+ } else {
+ appAttempt.addToUpdateContainerErrors(
+ UpdateContainerError.newInstance(
+ RMServerUtils.UPDATE_OUTSTANDING_ERROR, uReq));
+ }
+ } else {
+ LOG.warn("Cannot demote non-existent (or completed) Container ["
+ + uReq.getContainerId() + "]");
+ }
+ }
+ }
+
+ private RMContainer createDemotedRMContainer(
+ SchedulerApplicationAttempt appAttempt,
+ OpportunisticContainerContext oppCntxt,
+ RMContainer rmContainer) {
+ SchedulerRequestKey sk =
+ SchedulerRequestKey.extractFrom(rmContainer.getContainer());
+ Container demotedContainer = BuilderUtils.newContainer(
+ ContainerId.newContainerId(appAttempt.getApplicationAttemptId(),
+ oppCntxt.getContainerIdGenerator().generateContainerId()),
+ rmContainer.getContainer().getNodeId(),
+ rmContainer.getContainer().getNodeHttpAddress(),
+ rmContainer.getContainer().getResource(),
+ sk.getPriority(), null, ExecutionType.OPPORTUNISTIC,
+ sk.getAllocationRequestId());
+ demotedContainer.setVersion(rmContainer.getContainer().getVersion());
+ return SchedulerUtils.createOpportunisticRmContainer(
+ rmContext, demotedContainer, false);
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/650ff95e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/Allocation.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/Allocation.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/Allocation.java
index 2edaf9a..4d904d8 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/Allocation.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/Allocation.java
@@ -37,6 +37,8 @@ public class Allocation {
final List<NMToken> nmTokens;
final List<Container> increasedContainers;
final List<Container> decreasedContainers;
+ final List<Container> promotedContainers;
+ final List<Container> demotedContainers;
private Resource resourceLimit;
@@ -51,13 +53,23 @@ public class Allocation {
Set<ContainerId> strictContainers, Set<ContainerId> fungibleContainers,
List<ResourceRequest> fungibleResources, List<NMToken> nmTokens) {
this(containers, resourceLimit,strictContainers, fungibleContainers,
- fungibleResources, nmTokens, null, null);
+ fungibleResources, nmTokens, null, null, null, null);
}
-
+
public Allocation(List<Container> containers, Resource resourceLimit,
Set<ContainerId> strictContainers, Set<ContainerId> fungibleContainers,
List<ResourceRequest> fungibleResources, List<NMToken> nmTokens,
List<Container> increasedContainers, List<Container> decreasedContainer) {
+ this(containers, resourceLimit,strictContainers, fungibleContainers,
+ fungibleResources, nmTokens, increasedContainers, decreasedContainer,
+ null, null);
+ }
+
+ public Allocation(List<Container> containers, Resource resourceLimit,
+ Set<ContainerId> strictContainers, Set<ContainerId> fungibleContainers,
+ List<ResourceRequest> fungibleResources, List<NMToken> nmTokens,
+ List<Container> increasedContainers, List<Container> decreasedContainer,
+ List<Container> promotedContainers, List<Container> demotedContainer) {
this.containers = containers;
this.resourceLimit = resourceLimit;
this.strictContainers = strictContainers;
@@ -66,6 +78,8 @@ public class Allocation {
this.nmTokens = nmTokens;
this.increasedContainers = increasedContainers;
this.decreasedContainers = decreasedContainer;
+ this.promotedContainers = promotedContainers;
+ this.demotedContainers = demotedContainer;
}
public List<Container> getContainers() {
@@ -100,6 +114,14 @@ public class Allocation {
return decreasedContainers;
}
+ public List<Container> getPromotedContainers() {
+ return promotedContainers;
+ }
+
+ public List<Container> getDemotedContainers() {
+ return demotedContainers;
+ }
+
@VisibleForTesting
public void setResourceLimit(Resource resource) {
this.resourceLimit = resource;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/650ff95e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.java
index fff4493..2cc3742 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.java
@@ -55,7 +55,6 @@ import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;
-
/**
* This class keeps track of all the consumption of an application. This also
* keeps track of current running/completed containers for the application.
@@ -92,10 +91,11 @@ public class AppSchedulingInfo {
final Map<NodeId, Map<SchedulerRequestKey, Map<ContainerId,
SchedContainerChangeRequest>>> containerIncreaseRequestMap =
new ConcurrentHashMap<>();
-
private final ReentrantReadWriteLock.ReadLock readLock;
private final ReentrantReadWriteLock.WriteLock writeLock;
+ public final ContainerUpdateContext updateContext;
+
public AppSchedulingInfo(ApplicationAttemptId appAttemptId,
String user, Queue queue, ActiveUsersManager activeUsersManager,
long epoch, ResourceUsage appResourceUsage) {
@@ -109,6 +109,7 @@ public class AppSchedulingInfo {
this.appResourceUsage = appResourceUsage;
ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+ updateContext = new ContainerUpdateContext(this);
readLock = lock.readLock();
writeLock = lock.writeLock();
}
@@ -376,6 +377,10 @@ public class AppSchedulingInfo {
}
}
+ public ContainerUpdateContext getUpdateContext() {
+ return updateContext;
+ }
+
/**
* The ApplicationMaster is updating resource requirements for the
* application, by asking for more resources and releasing resources acquired
@@ -414,29 +419,9 @@ public class AppSchedulingInfo {
}
// Update scheduling placement set
- for (Map.Entry<SchedulerRequestKey, Map<String, ResourceRequest>> entry : dedupRequests.entrySet()) {
- SchedulerRequestKey schedulerRequestKey = entry.getKey();
-
- if (!schedulerKeyToPlacementSets.containsKey(schedulerRequestKey)) {
- schedulerKeyToPlacementSets.put(schedulerRequestKey,
- new LocalitySchedulingPlacementSet<>(this));
- }
-
- // Update placement set
- ResourceRequestUpdateResult pendingAmountChanges =
- schedulerKeyToPlacementSets.get(schedulerRequestKey)
- .updateResourceRequests(
- entry.getValue().values(),
- recoverPreemptedRequestForAContainer);
-
- if (null != pendingAmountChanges) {
- updatePendingResources(
- pendingAmountChanges.getLastAnyResourceRequest(),
- pendingAmountChanges.getNewResourceRequest(), schedulerRequestKey,
- queue.getMetrics());
- offswitchResourcesUpdated = true;
- }
- }
+ offswitchResourcesUpdated =
+ addToPlacementSets(
+ recoverPreemptedRequestForAContainer, dedupRequests);
return offswitchResourcesUpdated;
} finally {
@@ -444,6 +429,37 @@ public class AppSchedulingInfo {
}
}
+ boolean addToPlacementSets(
+ boolean recoverPreemptedRequestForAContainer,
+ Map<SchedulerRequestKey, Map<String, ResourceRequest>> dedupRequests) {
+ boolean offswitchResourcesUpdated = false;
+ for (Map.Entry<SchedulerRequestKey, Map<String, ResourceRequest>> entry :
+ dedupRequests.entrySet()) {
+ SchedulerRequestKey schedulerRequestKey = entry.getKey();
+
+ if (!schedulerKeyToPlacementSets.containsKey(schedulerRequestKey)) {
+ schedulerKeyToPlacementSets.put(schedulerRequestKey,
+ new LocalitySchedulingPlacementSet<>(this));
+ }
+
+ // Update placement set
+ ResourceRequestUpdateResult pendingAmountChanges =
+ schedulerKeyToPlacementSets.get(schedulerRequestKey)
+ .updateResourceRequests(
+ entry.getValue().values(),
+ recoverPreemptedRequestForAContainer);
+
+ if (null != pendingAmountChanges) {
+ updatePendingResources(
+ pendingAmountChanges.getLastAnyResourceRequest(),
+ pendingAmountChanges.getNewResourceRequest(), schedulerRequestKey,
+ queue.getMetrics());
+ offswitchResourcesUpdated = true;
+ }
+ }
+ return offswitchResourcesUpdated;
+ }
+
private void updatePendingResources(ResourceRequest lastRequest,
ResourceRequest request, SchedulerRequestKey schedulerKey,
QueueMetrics metrics) {
@@ -717,8 +733,8 @@ public class AppSchedulingInfo {
updateMetricsForAllocatedContainer(type, containerAllocated);
}
- return schedulerKeyToPlacementSets.get(schedulerKey).allocate(type, node,
- request);
+ return schedulerKeyToPlacementSets.get(schedulerKey)
+ .allocate(schedulerKey, type, node, request);
} finally {
writeLock.unlock();
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/650ff95e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/ContainerUpdateContext.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/ContainerUpdateContext.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/ContainerUpdateContext.java
new file mode 100644
index 0000000..91cc3b6
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/ContainerUpdateContext.java
@@ -0,0 +1,265 @@
+/**
+ * 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;
+
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.ContainerUpdateType;
+import org.apache.hadoop.yarn.api.records.ExecutionType;
+import org.apache.hadoop.yarn.api.records.ExecutionTypeRequest;
+import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.api.records.ResourceRequest;
+import org.apache.hadoop.yarn.api.records.UpdateContainerRequest;
+import org.apache.hadoop.yarn.factories.RecordFactory;
+import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
+import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Class encapsulates all outstanding container increase and decrease
+ * requests for an application.
+ */
+public class ContainerUpdateContext {
+
+ public static final ContainerId UNDEFINED =
+ ContainerId.newContainerId(ApplicationAttemptId.newInstance(
+ ApplicationId.newInstance(-1, -1), -1), -1);
+ protected static final RecordFactory RECORD_FACTORY =
+ RecordFactoryProvider.getRecordFactory(null);
+
+ // Keep track of containers that are undergoing promotion
+ private final Map<SchedulerRequestKey, Map<Resource,
+ Map<NodeId, Set<ContainerId>>>> outstandingIncreases = new HashMap<>();
+
+ private final Set<ContainerId> outstandingDecreases = new HashSet<>();
+ private final AppSchedulingInfo appSchedulingInfo;
+
+ ContainerUpdateContext(AppSchedulingInfo appSchedulingInfo) {
+ this.appSchedulingInfo = appSchedulingInfo;
+ }
+
+ private synchronized boolean isBeingIncreased(Container container) {
+ Map<Resource, Map<NodeId, Set<ContainerId>>> resourceMap =
+ outstandingIncreases.get(
+ new SchedulerRequestKey(container.getPriority(),
+ container.getAllocationRequestId(), container.getId()));
+ if (resourceMap != null) {
+ Map<NodeId, Set<ContainerId>> locationMap =
+ resourceMap.get(container.getResource());
+ if (locationMap != null) {
+ Set<ContainerId> containerIds = locationMap.get(container.getNodeId());
+ if (containerIds != null && !containerIds.isEmpty()) {
+ return containerIds.contains(container.getId());
+ }
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Add the container to outstanding decreases.
+ * @param container Container.
+ * @return true if updated to outstanding decreases was successful.
+ */
+ public synchronized boolean checkAndAddToOutstandingDecreases(
+ Container container) {
+ if (isBeingIncreased(container)
+ || outstandingDecreases.contains(container.getId())) {
+ return false;
+ }
+ outstandingDecreases.add(container.getId());
+ return true;
+ }
+
+ /**
+ * Add the container to outstanding increases.
+ * @param rmContainer RMContainer.
+ * @param schedulerNode SchedulerNode.
+ * @param updateRequest UpdateContainerRequest.
+ * @return true if updated to outstanding increases was successful.
+ */
+ public synchronized boolean checkAndAddToOutstandingIncreases(
+ RMContainer rmContainer, SchedulerNode schedulerNode,
+ UpdateContainerRequest updateRequest) {
+ Container container = rmContainer.getContainer();
+ SchedulerRequestKey schedulerKey =
+ SchedulerRequestKey.create(updateRequest,
+ rmContainer.getAllocatedSchedulerKey());
+ Map<Resource, Map<NodeId, Set<ContainerId>>> resourceMap =
+ outstandingIncreases.get(schedulerKey);
+ if (resourceMap == null) {
+ resourceMap = new HashMap<>();
+ outstandingIncreases.put(schedulerKey, resourceMap);
+ }
+ Map<NodeId, Set<ContainerId>> locationMap =
+ resourceMap.get(container.getResource());
+ if (locationMap == null) {
+ locationMap = new HashMap<>();
+ resourceMap.put(container.getResource(), locationMap);
+ }
+ Set<ContainerId> containerIds = locationMap.get(container.getNodeId());
+ if (containerIds == null) {
+ containerIds = new HashSet<>();
+ locationMap.put(container.getNodeId(), containerIds);
+ }
+ if (containerIds.contains(container.getId())
+ || outstandingDecreases.contains(container.getId())) {
+ return false;
+ }
+ containerIds.add(container.getId());
+
+ Map<SchedulerRequestKey, Map<String, ResourceRequest>> updateResReqs =
+ new HashMap<>();
+ Resource resToIncrease = getResourceToIncrease(updateRequest, rmContainer);
+ Map<String, ResourceRequest> resMap =
+ createResourceRequests(rmContainer, schedulerNode,
+ schedulerKey, resToIncrease);
+ updateResReqs.put(schedulerKey, resMap);
+ appSchedulingInfo.addToPlacementSets(false, updateResReqs);
+ return true;
+ }
+
+ private Map<String, ResourceRequest> createResourceRequests(
+ RMContainer rmContainer, SchedulerNode schedulerNode,
+ SchedulerRequestKey schedulerKey, Resource resToIncrease) {
+ Map<String, ResourceRequest> resMap = new HashMap<>();
+ resMap.put(rmContainer.getContainer().getNodeId().getHost(),
+ createResourceReqForIncrease(schedulerKey, resToIncrease,
+ RECORD_FACTORY.newRecordInstance(ResourceRequest.class),
+ rmContainer, rmContainer.getContainer().getNodeId().getHost()));
+ resMap.put(schedulerNode.getRackName(),
+ createResourceReqForIncrease(schedulerKey, resToIncrease,
+ RECORD_FACTORY.newRecordInstance(ResourceRequest.class),
+ rmContainer, schedulerNode.getRackName()));
+ resMap.put(ResourceRequest.ANY,
+ createResourceReqForIncrease(schedulerKey, resToIncrease,
+ RECORD_FACTORY.newRecordInstance(ResourceRequest.class),
+ rmContainer, ResourceRequest.ANY));
+ return resMap;
+ }
+
+ private Resource getResourceToIncrease(UpdateContainerRequest updateReq,
+ RMContainer rmContainer) {
+ if (updateReq.getContainerUpdateType() ==
+ ContainerUpdateType.PROMOTE_EXECUTION_TYPE) {
+ return rmContainer.getContainer().getResource();
+ }
+ // TODO: Fix this for container increase..
+ // This has to equal the Resources in excess of fitsIn()
+ // for container increase and is equal to the container total
+ // resource for Promotion.
+ return null;
+ }
+
+ private static ResourceRequest createResourceReqForIncrease(
+ SchedulerRequestKey schedulerRequestKey, Resource resToIncrease,
+ ResourceRequest rr, RMContainer rmContainer, String resourceName) {
+ rr.setResourceName(resourceName);
+ rr.setNumContainers(1);
+ rr.setRelaxLocality(false);
+ rr.setPriority(rmContainer.getContainer().getPriority());
+ rr.setAllocationRequestId(schedulerRequestKey.getAllocationRequestId());
+ rr.setCapability(resToIncrease);
+ rr.setNodeLabelExpression(rmContainer.getNodeLabelExpression());
+ rr.setExecutionTypeRequest(ExecutionTypeRequest.newInstance(
+ ExecutionType.GUARANTEED, true));
+ return rr;
+ }
+
+ /**
+ * Remove Container from outstanding increases / decreases. Calling this
+ * method essentially completes the update process.
+ * @param schedulerKey SchedulerRequestKey.
+ * @param container Container.
+ */
+ public synchronized void removeFromOutstandingUpdate(
+ SchedulerRequestKey schedulerKey, Container container) {
+ Map<Resource, Map<NodeId, Set<ContainerId>>> resourceMap =
+ outstandingIncreases.get(schedulerKey);
+ if (resourceMap != null) {
+ Map<NodeId, Set<ContainerId>> locationMap =
+ resourceMap.get(container.getResource());
+ if (locationMap != null) {
+ Set<ContainerId> containerIds = locationMap.get(container.getNodeId());
+ if (containerIds != null && !containerIds.isEmpty()) {
+ containerIds.remove(container.getId());
+ if (containerIds.isEmpty()) {
+ locationMap.remove(container.getNodeId());
+ }
+ }
+ if (locationMap.isEmpty()) {
+ resourceMap.remove(container.getResource());
+ }
+ }
+ if (resourceMap.isEmpty()) {
+ outstandingIncreases.remove(schedulerKey);
+ }
+ }
+ outstandingDecreases.remove(container.getId());
+ }
+
+ /**
+ * Check if a new container is to be matched up against an outstanding
+ * Container increase request.
+ * @param node SchedulerNode.
+ * @param schedulerKey SchedulerRequestKey.
+ * @param rmContainer RMContainer.
+ * @return ContainerId.
+ */
+ public ContainerId matchContainerToOutstandingIncreaseReq(
+ SchedulerNode node, SchedulerRequestKey schedulerKey,
+ RMContainer rmContainer) {
+ ContainerId retVal = null;
+ Container container = rmContainer.getContainer();
+ Map<Resource, Map<NodeId, Set<ContainerId>>> resourceMap =
+ outstandingIncreases.get(schedulerKey);
+ if (resourceMap != null) {
+ Map<NodeId, Set<ContainerId>> locationMap =
+ resourceMap.get(container.getResource());
+ if (locationMap != null) {
+ Set<ContainerId> containerIds = locationMap.get(container.getNodeId());
+ if (containerIds != null && !containerIds.isEmpty()) {
+ retVal = containerIds.iterator().next();
+ }
+ }
+ }
+ // Allocation happened on NM on the same host, but not on the NM
+ // we need.. We need to signal that this container has to be released.
+ // We also need to add these requests back.. to be reallocated.
+ if (resourceMap != null && retVal == null) {
+ Map<SchedulerRequestKey, Map<String, ResourceRequest>> reqsToUpdate =
+ new HashMap<>();
+ Map<String, ResourceRequest> resMap = createResourceRequests
+ (rmContainer, node, schedulerKey,
+ rmContainer.getContainer().getResource());
+ reqsToUpdate.put(schedulerKey, resMap);
+ appSchedulingInfo.addToPlacementSets(true, reqsToUpdate);
+ return UNDEFINED;
+ }
+ return retVal;
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org