hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From asur...@apache.org
Subject [4/4] hadoop git commit: YARN-4597. Introduce ContainerScheduler and a SCHEDULED state to NodeManager container lifecycle. (asuresh)
Date Tue, 15 Nov 2016 15:57:09 GMT
YARN-4597. Introduce ContainerScheduler and a SCHEDULED state to NodeManager container lifecycle. (asuresh)


Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/3219b7b4
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/3219b7b4
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/3219b7b4

Branch: refs/heads/trunk
Commit: 3219b7b4ac7d12aee343f6ab2980b3357fc618b6
Parents: 7ffb994
Author: Arun Suresh <asuresh@apache.org>
Authored: Tue Nov 15 07:48:55 2016 -0800
Committer: Arun Suresh <asuresh@apache.org>
Committed: Tue Nov 15 07:56:25 2016 -0800

----------------------------------------------------------------------
 .../hadoop/mapred/TestMROpportunisticMaps.java  |  18 +-
 .../yarn/api/records/ContainerExitStatus.java   |   6 +
 .../hadoop/yarn/api/records/ContainerState.java |   4 +-
 .../hadoop/yarn/conf/YarnConfiguration.java     |  17 +-
 .../src/main/proto/yarn_protos.proto            |   2 +-
 .../api/impl/TestDistributedScheduling.java     |   3 +-
 .../yarn/client/api/impl/TestNMClient.java      |   9 +-
 .../TestOpportunisticContainerAllocation.java   |   2 +
 .../src/main/resources/yarn-default.xml         |   6 +-
 .../hadoop/yarn/server/nodemanager/Context.java |  18 +-
 .../yarn/server/nodemanager/NodeManager.java    |  42 +-
 .../nodemanager/NodeStatusUpdaterImpl.java      |  54 --
 .../containermanager/ContainerManager.java      |   4 +
 .../containermanager/ContainerManagerImpl.java  |  32 +-
 .../containermanager/container/Container.java   |   6 +
 .../container/ContainerImpl.java                |  86 +-
 .../container/ContainerState.java               |   2 +-
 .../launcher/ContainerLaunch.java               |  30 +-
 .../launcher/RecoveredContainerLaunch.java      |   4 +-
 .../monitor/ContainersMonitor.java              |  52 +-
 .../monitor/ContainersMonitorImpl.java          |  79 +-
 .../queuing/QueuingContainerManagerImpl.java    | 686 ---------------
 .../containermanager/queuing/package-info.java  |  23 -
 ...locationBasedResourceUtilizationTracker.java | 137 +++
 .../scheduler/ContainerScheduler.java           | 419 +++++++++
 .../scheduler/ContainerSchedulerEvent.java      |  51 ++
 .../scheduler/ContainerSchedulerEventType.java  |  29 +
 .../scheduler/ResourceUtilizationTracker.java   |  59 ++
 .../scheduler/package-info.java                 |  22 +
 .../nodemanager/metrics/NodeManagerMetrics.java |  35 +
 .../nodemanager/webapp/ContainerLogsUtils.java  |   5 +-
 .../yarn/server/nodemanager/TestEventFlow.java  |   3 +-
 .../nodemanager/TestNodeManagerResync.java      |   8 +
 .../nodemanager/TestNodeManagerShutdown.java    |   5 +-
 .../nodemanager/TestNodeStatusUpdater.java      | 123 ---
 .../amrmproxy/BaseAMRMProxyTest.java            |   8 +-
 .../BaseContainerManagerTest.java               |  24 +-
 .../TestContainerManagerRecovery.java           |  34 +-
 .../TestContainerManagerRegression.java         |  84 --
 .../container/TestContainer.java                |  26 +-
 .../queuing/TestQueuingContainerManager.java    | 596 -------------
 .../TestContainerSchedulerQueuing.java          | 872 +++++++++++++++++++
 .../nodemanager/webapp/MockContainer.java       |  15 +
 ...pportunisticContainerAllocatorAMService.java |   4 +-
 .../rmcontainer/RMContainerImpl.java            |   4 +-
 .../resourcemanager/rmnode/RMNodeImpl.java      |  32 +-
 .../scheduler/AbstractYarnScheduler.java        |   2 +
 .../scheduler/SchedulerNode.java                |   9 +-
 .../hadoop/yarn/server/MiniYARNCluster.java     |  55 +-
 49 files changed, 1988 insertions(+), 1858 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestMROpportunisticMaps.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestMROpportunisticMaps.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestMROpportunisticMaps.java
index 021863b..d975fd0 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestMROpportunisticMaps.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestMROpportunisticMaps.java
@@ -68,15 +68,6 @@ public class TestMROpportunisticMaps {
     doTest(4, 1, 1, 2);
   }
 
-  /**
-   * Test will run with 6 Maps and 2 Reducers. All the Maps are OPPORTUNISTIC.
-   * @throws Exception
-   */
-  @Test
-  public void testMultipleReducers() throws Exception {
-    doTest(6, 2, 1, 6);
-  }
-
   public void doTest(int numMappers, int numReducers, int numNodes,
       int percent) throws Exception {
     doTest(numMappers, numReducers, numNodes, 1000, percent);
@@ -94,7 +85,8 @@ public class TestMROpportunisticMaps {
       conf.setBoolean(YarnConfiguration.
           OPPORTUNISTIC_CONTAINER_ALLOCATION_ENABLED, true);
       conf.setBoolean(YarnConfiguration.DIST_SCHEDULING_ENABLED, true);
-      conf.setBoolean(YarnConfiguration.NM_CONTAINER_QUEUING_ENABLED, true);
+      conf.setInt(
+          YarnConfiguration.NM_OPPORTUNISTIC_CONTAINERS_MAX_QUEUE_LENGTH, 10);
       dfsCluster = new MiniDFSCluster.Builder(conf)
           .numDataNodes(numNodes).build();
       fileSystem = dfsCluster.getFileSystem();
@@ -104,11 +96,7 @@ public class TestMROpportunisticMaps {
       createInput(fileSystem, numMappers, numLines);
       // Run the test.
 
-      Configuration jobConf = mrCluster.getConfig();
-      jobConf.set(YarnConfiguration.RM_SCHEDULER_ADDRESS,
-          YarnConfiguration.DEFAULT_AMRM_PROXY_ADDRESS);
-
-      runMergeTest(new JobConf(jobConf), fileSystem,
+      runMergeTest(new JobConf(conf), fileSystem,
           numMappers, numReducers, numLines, percent);
     } finally {
       if (dfsCluster != null) {

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerExitStatus.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerExitStatus.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerExitStatus.java
index f88fa3b..0207010 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerExitStatus.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerExitStatus.java
@@ -72,4 +72,10 @@ public class ContainerExitStatus {
    */
   public static final int KILLED_AFTER_APP_COMPLETION = -107;
 
+  /**
+   * Container was terminated by the ContainerScheduler to make room
+   * for another container...
+   */
+  public static final int KILLED_BY_CONTAINER_SCHEDULER = -108;
+
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerState.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerState.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerState.java
index 582389f..4efd8c1 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerState.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerState.java
@@ -36,6 +36,6 @@ public enum ContainerState {
   /** Completed container */
   COMPLETE,
 
-  /** Queued at the NM. */
-  QUEUED
+  /** Scheduled (awaiting resources) at the NM. */
+  SCHEDULED
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
index b95bd1a..a54104d 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
@@ -390,12 +390,16 @@ public class YarnConfiguration extends Configuration {
   public static final float NM_CONTAINER_QUEUING_LIMIT_STDEV_DEFAULT =
       1.0f;
 
-  /** Min length of container queue at NodeManager. */
+  /** Min length of container queue at NodeManager. This is a cluster-wide
+   * configuration that acts as the lower-bound of optimal queue length
+   * calculated by the NodeQueueLoadMonitor */
   public static final String NM_CONTAINER_QUEUING_MIN_QUEUE_LENGTH =
       YARN_PREFIX + "nm-container-queuing.min-queue-length";
   public static final int NM_CONTAINER_QUEUING_MIN_QUEUE_LENGTH_DEFAULT = 1;
 
-  /** Max length of container queue at NodeManager. */
+  /** Max length of container queue at NodeManager. This is a cluster-wide
+   * configuration that acts as the upper-bound of optimal queue length
+   * calculated by the NodeQueueLoadMonitor */
   public static final String NM_CONTAINER_QUEUING_MAX_QUEUE_LENGTH =
       YARN_PREFIX + "nm-container-queuing.max-queue-length";
   public static final int NM_CONTAINER_QUEUING_MAX_QUEUE_LENGTH_DEFAULT = 10;
@@ -834,10 +838,11 @@ public class YarnConfiguration extends Configuration {
   /** Prefix for all node manager configs.*/
   public static final String NM_PREFIX = "yarn.nodemanager.";
 
-  /** Enable Queuing of <code>OPPORTUNISTIC</code> containers. */
-  public static final String NM_CONTAINER_QUEUING_ENABLED = NM_PREFIX
-      + "container-queuing-enabled";
-  public static final boolean NM_CONTAINER_QUEUING_ENABLED_DEFAULT = false;
+  /** Max Queue length of <code>OPPORTUNISTIC</code> containers on the NM. */
+  public static final String NM_OPPORTUNISTIC_CONTAINERS_MAX_QUEUE_LENGTH =
+      NM_PREFIX + "opportunistic-containers-max-queue-length";
+  public static final int NM_OPPORTUNISTIC_CONTAINERS_MAX_QUEUE_LENGTH_DEFAULT =
+      0;
 
   /** Environment variables that will be sent to containers.*/
   public static final String NM_ADMIN_USER_ENV = NM_PREFIX + "admin-env";

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto
index 1022a38..cb37126 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto
@@ -82,7 +82,7 @@ enum ContainerStateProto {
   C_NEW = 1;
   C_RUNNING = 2;
   C_COMPLETE = 3;
-  C_QUEUED = 4;
+  C_SCHEDULED = 4;
 }
 
 message ContainerProto {

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestDistributedScheduling.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestDistributedScheduling.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestDistributedScheduling.java
index 4cfc4eb..b552d19 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestDistributedScheduling.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestDistributedScheduling.java
@@ -108,7 +108,8 @@ public class TestDistributedScheduling extends BaseAMRMProxyE2ETest {
     conf.setBoolean(YarnConfiguration.
         OPPORTUNISTIC_CONTAINER_ALLOCATION_ENABLED, true);
     conf.setBoolean(YarnConfiguration.DIST_SCHEDULING_ENABLED, true);
-    conf.setBoolean(YarnConfiguration.NM_CONTAINER_QUEUING_ENABLED, true);
+    conf.setInt(YarnConfiguration.NM_OPPORTUNISTIC_CONTAINERS_MAX_QUEUE_LENGTH,
+        10);
     cluster.init(conf);
     cluster.start();
     yarnConf = cluster.getConfig();

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestNMClient.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestNMClient.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestNMClient.java
index 3640883..d211d6d 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestNMClient.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestNMClient.java
@@ -36,6 +36,7 @@ import org.apache.hadoop.io.DataOutputBuffer;
 import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.service.Service.STATE;
+import org.apache.hadoop.util.Shell;
 import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
 import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationRequest;
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
@@ -330,6 +331,12 @@ public class TestNMClient {
           ByteBuffer.wrap(dob.getData(), 0, dob.getLength());
       ContainerLaunchContext clc =
           Records.newRecord(ContainerLaunchContext.class);
+      if (Shell.WINDOWS) {
+        clc.setCommands(
+            Arrays.asList("ping", "-n", "100", "127.0.0.1", ">nul"));
+      } else {
+        clc.setCommands(Arrays.asList("sleep", "10"));
+      }
       clc.setTokens(securityTokens);
       try {
         nmClient.startContainer(container, clc);
@@ -415,7 +422,7 @@ public class TestNMClient {
     try {
       nmClient.increaseContainerResource(container);
     } catch (YarnException e) {
-      // NM container will only be in LOCALIZED state, so expect the increase
+      // NM container will only be in SCHEDULED state, so expect the increase
       // action to fail.
       if (!e.getMessage().contains(
           "can only be changed when a container is in RUNNING state")) {

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestOpportunisticContainerAllocation.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestOpportunisticContainerAllocation.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestOpportunisticContainerAllocation.java
index ace145d..802c207 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestOpportunisticContainerAllocation.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestOpportunisticContainerAllocation.java
@@ -111,6 +111,8 @@ public class TestOpportunisticContainerAllocation {
     conf.setInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB, 512);
     conf.setBoolean(
         YarnConfiguration.OPPORTUNISTIC_CONTAINER_ALLOCATION_ENABLED, true);
+    conf.setInt(
+        YarnConfiguration.NM_OPPORTUNISTIC_CONTAINERS_MAX_QUEUE_LENGTH, 10);
     conf.setLong(YarnConfiguration.NM_LOG_RETAIN_SECONDS, 1);
     yarnCluster =
         new MiniYARNCluster(TestAMRMClient.class.getName(), nodeCount, 1, 1);

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
index 019166b..c436289 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
@@ -1000,10 +1000,10 @@
   </property>
 
   <property>
-    <description>Enable Queuing of OPPORTUNISTIC containers on the
+    <description>Max number of OPPORTUNISTIC containers to queue at the
       nodemanager.</description>
-    <name>yarn.nodemanager.container-queuing-enabled</name>
-    <value>false</value>
+    <name>yarn.nodemanager.opportunistic-containers-max-queue-length</name>
+    <value>0</value>
   </property>
 
   <property>

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/Context.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/Context.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/Context.java
index e888393..16a8497 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/Context.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/Context.java
@@ -27,12 +27,12 @@ import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.NodeId;
-import org.apache.hadoop.yarn.security.ContainerTokenIdentifier;
 import org.apache.hadoop.yarn.server.api.protocolrecords.LogAggregationReport;
 import org.apache.hadoop.yarn.server.api.records.NodeHealthStatus;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.ContainerManager;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.Application;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
+
 import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService;
 import org.apache.hadoop.yarn.server.scheduler.OpportunisticContainerAllocator;
 import org.apache.hadoop.yarn.server.nodemanager.security.NMContainerTokenSecretManager;
@@ -47,15 +47,6 @@ import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
 public interface Context {
 
   /**
-   * Interface exposing methods related to the queuing of containers in the NM.
-   */
-  interface QueuingContext {
-    ConcurrentMap<ContainerId, ContainerTokenIdentifier> getQueuedContainers();
-
-    ConcurrentMap<ContainerTokenIdentifier, String> getKilledQueuedContainers();
-  }
-
-  /**
    * Return the nodeId. Usable only when the ContainerManager is started.
    * 
    * @return the NodeId
@@ -112,13 +103,6 @@ public interface Context {
 
   NodeStatusUpdater getNodeStatusUpdater();
 
-  /**
-   * Returns a <code>QueuingContext</code> that provides information about the
-   * number of Containers Queued as well as the number of Containers that were
-   * queued and killed.
-   */
-  QueuingContext getQueuingContext();
-
   boolean isDistributedSchedulingEnabled();
 
   OpportunisticContainerAllocator getContainerAllocator();

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeManager.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeManager.java
index 0f0a081..72875a9 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeManager.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeManager.java
@@ -56,7 +56,6 @@ import org.apache.hadoop.yarn.event.Dispatcher;
 import org.apache.hadoop.yarn.event.EventHandler;
 import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
 import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
-import org.apache.hadoop.yarn.security.ContainerTokenIdentifier;
 import org.apache.hadoop.yarn.server.api.protocolrecords.LogAggregationReport;
 import org.apache.hadoop.yarn.server.api.records.NodeHealthStatus;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.ContainerManager;
@@ -64,7 +63,6 @@ import org.apache.hadoop.yarn.server.nodemanager.collectormanager.NMCollectorSer
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.ContainerManagerImpl;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.Application;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
-import org.apache.hadoop.yarn.server.nodemanager.containermanager.queuing.QueuingContainerManagerImpl;
 import org.apache.hadoop.yarn.server.nodemanager.metrics.NodeManagerMetrics;
 import org.apache.hadoop.yarn.server.nodemanager.nodelabels.ConfigurationNodeLabelsProvider;
 import org.apache.hadoop.yarn.server.nodemanager.nodelabels.NodeLabelsProvider;
@@ -177,14 +175,8 @@ public class NodeManager extends CompositeService
       ContainerExecutor exec, DeletionService del,
       NodeStatusUpdater nodeStatusUpdater, ApplicationACLsManager aclsManager,
       LocalDirsHandlerService dirsHandler) {
-    if (getConfig().getBoolean(YarnConfiguration.NM_CONTAINER_QUEUING_ENABLED,
-        YarnConfiguration.NM_CONTAINER_QUEUING_ENABLED_DEFAULT)) {
-      return new QueuingContainerManagerImpl(context, exec, del,
-          nodeStatusUpdater, metrics, dirsHandler);
-    } else {
-      return new ContainerManagerImpl(context, exec, del, nodeStatusUpdater,
-          metrics, dirsHandler);
-    }
+    return new ContainerManagerImpl(context, exec, del, nodeStatusUpdater,
+        metrics, dirsHandler);
   }
 
   protected NMCollectorService createNMCollectorService(Context ctxt) {
@@ -510,7 +502,6 @@ public class NodeManager extends CompositeService
 
     private OpportunisticContainerAllocator containerAllocator;
 
-    private final QueuingContext queuingContext;
     private ContainerExecutor executor;
 
     private NMTimelinePublisher nmTimelinePublisher;
@@ -533,7 +524,6 @@ public class NodeManager extends CompositeService
       this.stateStore = stateStore;
       this.logAggregationReportForApps = new ConcurrentLinkedQueue<
           LogAggregationReport>();
-      this.queuingContext = new QueuingNMContext();
       this.isDistSchedulingEnabled = isDistSchedulingEnabled;
       this.conf = conf;
     }
@@ -662,11 +652,6 @@ public class NodeManager extends CompositeService
       this.nodeStatusUpdater = nodeStatusUpdater;
     }
 
-    @Override
-    public QueuingContext getQueuingContext() {
-      return this.queuingContext;
-    }
-
     public boolean isDistributedSchedulingEnabled() {
       return isDistSchedulingEnabled;
     }
@@ -716,29 +701,6 @@ public class NodeManager extends CompositeService
   }
 
   /**
-   * Class that keeps the context for containers queued at the NM.
-   */
-  public static class QueuingNMContext implements Context.QueuingContext {
-    protected final ConcurrentMap<ContainerId, ContainerTokenIdentifier>
-        queuedContainers = new ConcurrentSkipListMap<>();
-
-    protected final ConcurrentMap<ContainerTokenIdentifier, String>
-        killedQueuedContainers = new ConcurrentHashMap<>();
-
-    @Override
-    public ConcurrentMap<ContainerId, ContainerTokenIdentifier>
-        getQueuedContainers() {
-      return this.queuedContainers;
-    }
-
-    @Override
-    public ConcurrentMap<ContainerTokenIdentifier, String>
-        getKilledQueuedContainers() {
-      return this.killedQueuedContainers;
-    }
-  }
-
-  /**
    * @return the node health checker
    */
   public NodeHealthCheckerService getNodeHealthChecker() {

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdaterImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdaterImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdaterImpl.java
index 6e50d24..7f74ed8 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdaterImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdaterImpl.java
@@ -47,7 +47,6 @@ import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.VersionUtil;
 import org.apache.hadoop.yarn.api.protocolrecords.SignalContainerRequest;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
-import org.apache.hadoop.yarn.api.records.ContainerExitStatus;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.ContainerState;
 import org.apache.hadoop.yarn.api.records.ContainerStatus;
@@ -62,7 +61,6 @@ import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
 import org.apache.hadoop.yarn.factories.RecordFactory;
 import org.apache.hadoop.yarn.factories.impl.pb.RecordFactoryPBImpl;
 import org.apache.hadoop.yarn.nodelabels.CommonNodeLabelsManager;
-import org.apache.hadoop.yarn.security.ContainerTokenIdentifier;
 import org.apache.hadoop.yarn.server.api.ResourceManagerConstants;
 import org.apache.hadoop.yarn.server.api.ResourceTracker;
 import org.apache.hadoop.yarn.server.api.ServerRMProxy;
@@ -89,7 +87,6 @@ import org.apache.hadoop.yarn.server.nodemanager.metrics.NodeManagerMetrics;
 import org.apache.hadoop.yarn.server.nodemanager.nodelabels.NodeLabelsProvider;
 import org.apache.hadoop.yarn.server.nodemanager.timelineservice.NMTimelinePublisher;
 import org.apache.hadoop.yarn.server.nodemanager.util.NodeManagerHardwareUtils;
-import org.apache.hadoop.yarn.server.utils.BuilderUtils;
 import org.apache.hadoop.yarn.util.resource.Resources;
 import org.apache.hadoop.yarn.util.ResourceCalculatorPlugin;
 import org.apache.hadoop.yarn.util.YarnVersionInfo;
@@ -570,9 +567,6 @@ public class NodeStatusUpdaterImpl extends AbstractService implements
       }
     }
 
-    // Account for all containers that got killed while they were still queued.
-    pendingCompletedContainers.putAll(getKilledQueuedContainerStatuses());
-
     containerStatuses.addAll(pendingCompletedContainers.values());
 
     if (LOG.isDebugEnabled()) {
@@ -582,43 +576,6 @@ public class NodeStatusUpdaterImpl extends AbstractService implements
     return containerStatuses;
   }
 
-  /**
-   * Add to the container statuses the status of the containers that got killed
-   * while they were queued.
-   */
-  private Map<ContainerId, ContainerStatus> getKilledQueuedContainerStatuses() {
-    Map<ContainerId, ContainerStatus> killedQueuedContainerStatuses =
-        new HashMap<>();
-    for (Map.Entry<ContainerTokenIdentifier, String> killedQueuedContainer :
-        this.context.getQueuingContext().
-            getKilledQueuedContainers().entrySet()) {
-      ContainerTokenIdentifier containerTokenId = killedQueuedContainer
-          .getKey();
-      ContainerId containerId = containerTokenId.getContainerID();
-      ContainerStatus containerStatus = BuilderUtils.newContainerStatus(
-          containerId, ContainerState.COMPLETE,
-          killedQueuedContainer.getValue(), ContainerExitStatus.ABORTED,
-          containerTokenId.getResource(), containerTokenId.getExecutionType());
-      ApplicationId applicationId = containerId.getApplicationAttemptId()
-          .getApplicationId();
-      if (isApplicationStopped(applicationId)) {
-        if (LOG.isDebugEnabled()) {
-          LOG.debug(applicationId + " is completing, " + " remove "
-              + containerId + " from NM context.");
-        }
-        this.context.getQueuingContext().getKilledQueuedContainers()
-            .remove(containerTokenId);
-        killedQueuedContainerStatuses.put(containerId, containerStatus);
-      } else {
-        if (!isContainerRecentlyStopped(containerId)) {
-          killedQueuedContainerStatuses.put(containerId, containerStatus);
-        }
-      }
-      addCompletedContainer(containerId);
-    }
-    return killedQueuedContainerStatuses;
-  }
-
   private List<ApplicationId> getRunningApplications() {
     List<ApplicationId> runningApplications = new ArrayList<ApplicationId>();
     runningApplications.addAll(this.context.getApplications().keySet());
@@ -703,17 +660,6 @@ public class NodeStatusUpdaterImpl extends AbstractService implements
       }
     }
 
-    // Remove null containers from queuing context for killed queued containers.
-    Iterator<ContainerTokenIdentifier> killedQueuedContIter =
-        context.getQueuingContext().getKilledQueuedContainers().keySet().
-            iterator();
-    while (killedQueuedContIter.hasNext()) {
-      if (removedNullContainers.contains(
-          killedQueuedContIter.next().getContainerID())) {
-        killedQueuedContIter.remove();
-      }
-    }
-
     if (!removedContainers.isEmpty()) {
       LOG.info("Removed completed containers from NM context: "
           + removedContainers);

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManager.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManager.java
index 1cbb8c7..066d987 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManager.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManager.java
@@ -26,6 +26,8 @@ import org.apache.hadoop.yarn.server.api.records.OpportunisticContainersStatus;
 import org.apache.hadoop.yarn.server.nodemanager.ContainerManagerEvent;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor
     .ContainersMonitor;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.scheduler
+    .ContainerScheduler;
 
 /**
  * The ContainerManager is an entity that manages the life cycle of Containers.
@@ -42,4 +44,6 @@ public interface ContainerManager extends ServiceStateChangeListener,
 
   void setBlockNewContainerRequests(boolean blockNewContainerRequests);
 
+  ContainerScheduler getContainerScheduler();
+
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java
index c7810f9..e8de8b7 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java
@@ -136,6 +136,9 @@ import org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor.Change
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor.ContainersMonitor;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor.ContainersMonitorEventType;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor.ContainersMonitorImpl;
+
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.scheduler.ContainerScheduler;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.scheduler.ContainerSchedulerEventType;
 import org.apache.hadoop.yarn.server.nodemanager.metrics.NodeManagerMetrics;
 import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService;
 import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredApplicationsState;
@@ -205,6 +208,7 @@ public class ContainerManagerImpl extends CompositeService implements
   private final WriteLock writeLock;
   private AMRMProxyService amrmProxyService;
   protected boolean amrmProxyEnabled = false;
+  private final ContainerScheduler containerScheduler;
 
   private long waitForContainersOnShutdownMillis;
 
@@ -231,6 +235,8 @@ public class ContainerManagerImpl extends CompositeService implements
     addService(containersLauncher);
 
     this.nodeStatusUpdater = nodeStatusUpdater;
+    this.containerScheduler = createContainerScheduler(context);
+    addService(containerScheduler);
 
     // Start configurable services
     auxiliaryServices = new AuxServices();
@@ -259,7 +265,8 @@ public class ContainerManagerImpl extends CompositeService implements
     dispatcher.register(AuxServicesEventType.class, auxiliaryServices);
     dispatcher.register(ContainersMonitorEventType.class, containersMonitor);
     dispatcher.register(ContainersLauncherEventType.class, containersLauncher);
-    
+    dispatcher.register(ContainerSchedulerEventType.class, containerScheduler);
+
     addService(dispatcher);
 
     ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
@@ -311,6 +318,14 @@ public class ContainerManagerImpl extends CompositeService implements
     }
   }
 
+  @VisibleForTesting
+  protected ContainerScheduler createContainerScheduler(Context cntxt) {
+    // Currently, this dispatcher is shared by the ContainerManager,
+    // all the containers, the container monitor and all the container.
+    // The ContainerScheduler may use its own dispatcher.
+    return new ContainerScheduler(cntxt, dispatcher, metrics);
+  }
+
   protected ContainersMonitor createContainersMonitor(ContainerExecutor exec) {
     return new ContainersMonitorImpl(exec, dispatcher, this.context);
   }
@@ -1263,10 +1278,8 @@ public class ContainerManagerImpl extends CompositeService implements
       }
     } else {
       context.getNMStateStore().storeContainerKilled(containerID);
-      dispatcher.getEventHandler().handle(
-        new ContainerKillEvent(containerID,
-            ContainerExitStatus.KILLED_BY_APPMASTER,
-            "Container killed by the ApplicationMaster."));
+      container.sendKillEvent(ContainerExitStatus.KILLED_BY_APPMASTER,
+          "Container killed by the ApplicationMaster.");
 
       NMAuditLogger.logSuccess(container.getUser(),    
         AuditConstants.STOP_CONTAINER, "ContainerManageImpl", containerID
@@ -1521,12 +1534,12 @@ public class ContainerManagerImpl extends CompositeService implements
 
   @Override
   public OpportunisticContainersStatus getOpportunisticContainersStatus() {
-    return OpportunisticContainersStatus.newInstance();
+    return this.containerScheduler.getOpportunisticContainersStatus();
   }
 
   @Override
   public void updateQueuingLimit(ContainerQueuingLimit queuingLimit) {
-    LOG.trace("Implementation does not support queuing of Containers!!");
+    this.containerScheduler.updateQueuingLimit(queuingLimit);
   }
 
   @SuppressWarnings("unchecked")
@@ -1687,4 +1700,9 @@ public class ContainerManagerImpl extends CompositeService implements
       LOG.info("Container " + containerId + " no longer exists");
     }
   }
+
+  @Override
+  public ContainerScheduler getContainerScheduler() {
+    return this.containerScheduler;
+  }
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/Container.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/Container.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/Container.java
index 78c240a..8004f33 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/Container.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/Container.java
@@ -83,7 +83,13 @@ public interface Container extends EventHandler<ContainerEvent> {
 
   boolean isReInitializing();
 
+  boolean isMarkedForKilling();
+
   boolean canRollback();
 
   void commitUpgrade();
+
+  void sendLaunchEvent();
+
+  void sendKillEvent(int exitStatus, String description);
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/ContainerImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/ContainerImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/ContainerImpl.java
index 307cf5d..4a6be32 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/ContainerImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/ContainerImpl.java
@@ -74,6 +74,8 @@ import org.apache.hadoop.yarn.server.nodemanager.containermanager.loghandler.eve
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor.ContainerMetrics;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor.ContainerStartMonitoringEvent;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor.ContainerStopMonitoringEvent;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.scheduler.ContainerSchedulerEvent;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.scheduler.ContainerSchedulerEventType;
 import org.apache.hadoop.yarn.server.nodemanager.metrics.NodeManagerMetrics;
 import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService;
 import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredContainerState;
@@ -164,6 +166,7 @@ public class ContainerImpl implements Container {
   private String ips;
   private volatile ReInitializationContext reInitContext;
   private volatile boolean isReInitializing = false;
+  private volatile boolean isMarkeForKilling = false;
 
   /** The NM-wide configuration - not specific to this container */
   private final Configuration daemonConf;
@@ -286,7 +289,7 @@ public class ContainerImpl implements Container {
     // From NEW State
     .addTransition(ContainerState.NEW,
         EnumSet.of(ContainerState.LOCALIZING,
-            ContainerState.LOCALIZED,
+            ContainerState.SCHEDULED,
             ContainerState.LOCALIZATION_FAILED,
             ContainerState.DONE),
         ContainerEventType.INIT_CONTAINER, new RequestResourcesTransition())
@@ -298,7 +301,7 @@ public class ContainerImpl implements Container {
 
     // From LOCALIZING State
     .addTransition(ContainerState.LOCALIZING,
-        EnumSet.of(ContainerState.LOCALIZING, ContainerState.LOCALIZED),
+        EnumSet.of(ContainerState.LOCALIZING, ContainerState.SCHEDULED),
         ContainerEventType.RESOURCE_LOCALIZED, new LocalizedTransition())
     .addTransition(ContainerState.LOCALIZING,
         ContainerState.LOCALIZATION_FAILED,
@@ -309,7 +312,7 @@ public class ContainerImpl implements Container {
         UPDATE_DIAGNOSTICS_TRANSITION)
     .addTransition(ContainerState.LOCALIZING, ContainerState.KILLING,
         ContainerEventType.KILL_CONTAINER,
-        new KillDuringLocalizationTransition())
+        new KillBeforeRunningTransition())
 
     // From LOCALIZATION_FAILED State
     .addTransition(ContainerState.LOCALIZATION_FAILED,
@@ -334,17 +337,18 @@ public class ContainerImpl implements Container {
         ContainerState.LOCALIZATION_FAILED,
         ContainerEventType.RESOURCE_FAILED)
 
-    // From LOCALIZED State
-    .addTransition(ContainerState.LOCALIZED, ContainerState.RUNNING,
+    // From SCHEDULED State
+    .addTransition(ContainerState.SCHEDULED, ContainerState.RUNNING,
         ContainerEventType.CONTAINER_LAUNCHED, new LaunchTransition())
-    .addTransition(ContainerState.LOCALIZED, ContainerState.EXITED_WITH_FAILURE,
+    .addTransition(ContainerState.SCHEDULED, ContainerState.EXITED_WITH_FAILURE,
         ContainerEventType.CONTAINER_EXITED_WITH_FAILURE,
         new ExitedWithFailureTransition(true))
-    .addTransition(ContainerState.LOCALIZED, ContainerState.LOCALIZED,
+    .addTransition(ContainerState.SCHEDULED, ContainerState.SCHEDULED,
        ContainerEventType.UPDATE_DIAGNOSTICS_MSG,
        UPDATE_DIAGNOSTICS_TRANSITION)
-    .addTransition(ContainerState.LOCALIZED, ContainerState.KILLING,
-        ContainerEventType.KILL_CONTAINER, new KillTransition())
+    .addTransition(ContainerState.SCHEDULED, ContainerState.KILLING,
+        ContainerEventType.KILL_CONTAINER,
+        new KillBeforeRunningTransition())
 
     // From RUNNING State
     .addTransition(ContainerState.RUNNING,
@@ -353,7 +357,7 @@ public class ContainerImpl implements Container {
         new ExitedWithSuccessTransition(true))
     .addTransition(ContainerState.RUNNING,
         EnumSet.of(ContainerState.RELAUNCHING,
-            ContainerState.LOCALIZED,
+            ContainerState.SCHEDULED,
             ContainerState.EXITED_WITH_FAILURE),
         ContainerEventType.CONTAINER_EXITED_WITH_FAILURE,
         new RetryFailureTransition())
@@ -402,7 +406,7 @@ public class ContainerImpl implements Container {
     .addTransition(ContainerState.REINITIALIZING, ContainerState.KILLING,
         ContainerEventType.KILL_CONTAINER, new KillTransition())
     .addTransition(ContainerState.REINITIALIZING,
-        ContainerState.LOCALIZED,
+        ContainerState.SCHEDULED,
         ContainerEventType.CONTAINER_KILLED_ON_REQUEST,
         new KilledForReInitializationTransition())
 
@@ -520,9 +524,11 @@ public class ContainerImpl implements Container {
     case NEW:
     case LOCALIZING:
     case LOCALIZATION_FAILED:
-    case LOCALIZED:
+    case SCHEDULED:
+      return org.apache.hadoop.yarn.api.records.ContainerState.SCHEDULED;
     case RUNNING:
     case RELAUNCHING:
+    case REINITIALIZING:
     case EXITED_WITH_SUCCESS:
     case EXITED_WITH_FAILURE:
     case KILLING:
@@ -553,7 +559,7 @@ public class ContainerImpl implements Container {
   public Map<Path, List<String>> getLocalizedResources() {
     this.readLock.lock();
     try {
-      if (ContainerState.LOCALIZED == getContainerState()
+      if (ContainerState.SCHEDULED == getContainerState()
           || ContainerState.RELAUNCHING == getContainerState()) {
         return resourceSet.getLocalizedResources();
       } else {
@@ -690,6 +696,9 @@ public class ContainerImpl implements Container {
     ContainerStatus containerStatus = cloneAndGetContainerStatus();
     eventHandler.handle(new ApplicationContainerFinishedEvent(containerStatus));
 
+    // Tell the scheduler the container is Done
+    eventHandler.handle(new ContainerSchedulerEvent(this,
+        ContainerSchedulerEventType.CONTAINER_COMPLETED));
     // Remove the container from the resource-monitor
     eventHandler.handle(new ContainerStopMonitoringEvent(containerId));
     // Tell the logService too
@@ -698,7 +707,8 @@ public class ContainerImpl implements Container {
   }
 
   @SuppressWarnings("unchecked") // dispatcher not typed
-  private void sendLaunchEvent() {
+  @Override
+  public void sendLaunchEvent() {
     ContainersLauncherEventType launcherEvent =
         ContainersLauncherEventType.LAUNCH_CONTAINER;
     if (recoveredStatus == RecoveredContainerStatus.LAUNCHED) {
@@ -711,6 +721,22 @@ public class ContainerImpl implements Container {
   }
 
   @SuppressWarnings("unchecked") // dispatcher not typed
+  private void sendScheduleEvent() {
+    dispatcher.getEventHandler().handle(
+        new ContainerSchedulerEvent(this,
+            ContainerSchedulerEventType.SCHEDULE_CONTAINER)
+    );
+  }
+
+  @SuppressWarnings("unchecked") // dispatcher not typed
+  @Override
+  public void sendKillEvent(int exitStatus, String description) {
+    this.isMarkeForKilling = true;
+    dispatcher.getEventHandler().handle(
+        new ContainerKillEvent(containerId, exitStatus, description));
+  }
+
+  @SuppressWarnings("unchecked") // dispatcher not typed
   private void sendRelaunchEvent() {
     ContainersLauncherEventType launcherEvent =
         ContainersLauncherEventType.RELAUNCH_CONTAINER;
@@ -781,7 +807,7 @@ public class ContainerImpl implements Container {
    * to the ResourceLocalizationManager and enters LOCALIZING state.
    * 
    * If there are no resources to localize, sends LAUNCH_CONTAINER event
-   * and enters LOCALIZED state directly.
+   * and enters SCHEDULED state directly.
    * 
    * If there are any invalid resources specified, enters LOCALIZATION_FAILED
    * directly.
@@ -847,9 +873,9 @@ public class ContainerImpl implements Container {
         }
         return ContainerState.LOCALIZING;
       } else {
-        container.sendLaunchEvent();
+        container.sendScheduleEvent();
         container.metrics.endInitingContainer();
-        return ContainerState.LOCALIZED;
+        return ContainerState.SCHEDULED;
       }
     }
   }
@@ -889,7 +915,7 @@ public class ContainerImpl implements Container {
           new ContainerLocalizationEvent(LocalizationEventType.
               CONTAINER_RESOURCES_LOCALIZED, container));
 
-      container.sendLaunchEvent();
+      container.sendScheduleEvent();
       container.metrics.endInitingContainer();
 
       // If this is a recovered container that has already launched, skip
@@ -909,7 +935,7 @@ public class ContainerImpl implements Container {
                 SharedCacheUploadEventType.UPLOAD));
       }
 
-      return ContainerState.LOCALIZED;
+      return ContainerState.SCHEDULED;
     }
   }
 
@@ -1099,7 +1125,7 @@ public class ContainerImpl implements Container {
   }
 
   /**
-   * Transition from LOCALIZED state to RUNNING state upon receiving
+   * Transition from SCHEDULED state to RUNNING state upon receiving
    * a CONTAINER_LAUNCHED event.
    */
   static class LaunchTransition extends ContainerTransition {
@@ -1257,7 +1283,7 @@ public class ContainerImpl implements Container {
             container.containerId.getApplicationAttemptId().getApplicationId(),
             container.containerId);
         new KilledForReInitializationTransition().transition(container, event);
-        return ContainerState.LOCALIZED;
+        return ContainerState.SCHEDULED;
       } else {
         new ExitedWithFailureTransition(true).transition(container, event);
         return ContainerState.EXITED_WITH_FAILURE;
@@ -1339,7 +1365,7 @@ public class ContainerImpl implements Container {
   }
 
   /**
-   * Transition to LOCALIZED and wait for RE-LAUNCH
+   * Transition to SCHEDULED and wait for RE-LAUNCH
    */
   static class KilledForReInitializationTransition extends ContainerTransition {
 
@@ -1363,8 +1389,8 @@ public class ContainerImpl implements Container {
 
       container.resourceSet =
           container.reInitContext.mergedResourceSet(container.resourceSet);
-
-      container.sendLaunchEvent();
+      container.isMarkeForKilling = false;
+      container.sendScheduleEvent();
     }
   }
 
@@ -1392,7 +1418,7 @@ public class ContainerImpl implements Container {
    * Transition from LOCALIZING to KILLING upon receiving
    * KILL_CONTAINER event.
    */
-  static class KillDuringLocalizationTransition implements
+  static class KillBeforeRunningTransition implements
       SingleArcTransition<ContainerImpl, ContainerEvent> {
     @Override
     public void transition(ContainerImpl container, ContainerEvent event) {
@@ -1424,7 +1450,7 @@ public class ContainerImpl implements Container {
 
   /**
    * Transitions upon receiving KILL_CONTAINER.
-   * - LOCALIZED -> KILLING.
+   * - SCHEDULED -> KILLING.
    * - RUNNING -> KILLING.
    * - REINITIALIZING -> KILLING.
    */
@@ -1651,7 +1677,8 @@ public class ContainerImpl implements Container {
             stateMachine.doTransition(event.getType(), event);
       } catch (InvalidStateTransitionException e) {
         LOG.warn("Can't handle this event at current state: Current: ["
-            + oldState + "], eventType: [" + event.getType() + "]", e);
+            + oldState + "], eventType: [" + event.getType() + "]," +
+            " container: [" + containerID + "]", e);
       }
       if (oldState != newState) {
         LOG.info("Container " + containerID + " transitioned from "
@@ -1715,6 +1742,11 @@ public class ContainerImpl implements Container {
   }
 
   @Override
+  public boolean isMarkedForKilling() {
+    return this.isMarkeForKilling;
+  }
+
+  @Override
   public boolean canRollback() {
     return (this.reInitContext != null)
         && (this.reInitContext.canRollback());

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/ContainerState.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/ContainerState.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/ContainerState.java
index 70de90c..91d1356 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/ContainerState.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/ContainerState.java
@@ -19,7 +19,7 @@
 package org.apache.hadoop.yarn.server.nodemanager.containermanager.container;
 
 public enum ContainerState {
-  NEW, LOCALIZING, LOCALIZATION_FAILED, LOCALIZED, RUNNING, RELAUNCHING,
+  NEW, LOCALIZING, LOCALIZATION_FAILED, SCHEDULED, RUNNING, RELAUNCHING,
   REINITIALIZING, EXITED_WITH_SUCCESS, EXITED_WITH_FAILURE, KILLING,
   CONTAINER_CLEANEDUP_AFTER_KILL, CONTAINER_RESOURCES_CLEANINGUP, DONE
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerLaunch.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerLaunch.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerLaunch.java
index d774030..823457f 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerLaunch.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerLaunch.java
@@ -104,9 +104,10 @@ public class ContainerLaunch implements Callable<Integer> {
   private final Context context;
   private final ContainerManagerImpl containerManager;
   
-  protected AtomicBoolean shouldLaunchContainer = new AtomicBoolean(false);
+  protected AtomicBoolean containerAlreadyLaunched = new AtomicBoolean(false);
   protected AtomicBoolean completed = new AtomicBoolean(false);
 
+  private volatile boolean killedBeforeStart = false;
   private long sleepDelayBeforeSigKill = 250;
   private long maxKillWaitTime = 2000;
 
@@ -401,7 +402,12 @@ public class ContainerLaunch implements Callable<Integer> {
   @SuppressWarnings("unchecked")
   protected int launchContainer(ContainerStartContext ctx) throws IOException {
     ContainerId containerId = container.getContainerId();
-
+    if (container.isMarkedForKilling()) {
+      LOG.info("Container " + containerId + " not launched as it has already "
+          + "been marked for Killing");
+      this.killedBeforeStart = true;
+      return ExitCode.TERMINATED.getExitCode();
+    }
     // LaunchContainer is a blocking call. We are here almost means the
     // container is launched, so send out the event.
     dispatcher.getEventHandler().handle(new ContainerEvent(
@@ -410,7 +416,7 @@ public class ContainerLaunch implements Callable<Integer> {
     context.getNMStateStore().storeContainerLaunched(containerId);
 
     // Check if the container is signalled to be killed.
-    if (!shouldLaunchContainer.compareAndSet(false, true)) {
+    if (!containerAlreadyLaunched.compareAndSet(false, true)) {
       LOG.info("Container " + containerId + " not launched as "
           + "cleanup already called");
       return ExitCode.TERMINATED.getExitCode();
@@ -451,10 +457,14 @@ public class ContainerLaunch implements Callable<Integer> {
         || exitCode == ExitCode.TERMINATED.getExitCode()) {
       // If the process was killed, Send container_cleanedup_after_kill and
       // just break out of this method.
-      dispatcher.getEventHandler().handle(
-          new ContainerExitEvent(containerId,
-              ContainerEventType.CONTAINER_KILLED_ON_REQUEST, exitCode,
-              diagnosticInfo.toString()));
+
+      // If Container was killed before starting... NO need to do this.
+      if (!killedBeforeStart) {
+        dispatcher.getEventHandler().handle(
+            new ContainerExitEvent(containerId,
+                ContainerEventType.CONTAINER_KILLED_ON_REQUEST, exitCode,
+                diagnosticInfo.toString()));
+      }
     } else if (exitCode != 0) {
       handleContainerExitWithFailure(containerId, exitCode, containerLogDir,
           diagnosticInfo);
@@ -565,7 +575,8 @@ public class ContainerLaunch implements Callable<Integer> {
     }
 
     // launch flag will be set to true if process already launched
-    boolean alreadyLaunched = !shouldLaunchContainer.compareAndSet(false, true);
+    boolean alreadyLaunched =
+        !containerAlreadyLaunched.compareAndSet(false, true);
     if (!alreadyLaunched) {
       LOG.info("Container " + containerIdStr + " not launched."
           + " No cleanup needed to be done");
@@ -660,7 +671,8 @@ public class ContainerLaunch implements Callable<Integer> {
 
     LOG.info("Sending signal " + command + " to container " + containerIdStr);
 
-    boolean alreadyLaunched = !shouldLaunchContainer.compareAndSet(false, true);
+    boolean alreadyLaunched =
+        !containerAlreadyLaunched.compareAndSet(false, true);
     if (!alreadyLaunched) {
       LOG.info("Container " + containerIdStr + " not launched."
           + " Not sending the signal");

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/RecoveredContainerLaunch.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/RecoveredContainerLaunch.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/RecoveredContainerLaunch.java
index 3cd31b7..a04a23f 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/RecoveredContainerLaunch.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/RecoveredContainerLaunch.java
@@ -39,7 +39,7 @@ import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Cont
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerExitEvent;
 import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerReacquisitionContext;
-import org.apache.hadoop.yarn.util.ConverterUtils;
+
 
 /**
  * This is a ContainerLaunch which has been recovered after an NM restart (for
@@ -57,7 +57,7 @@ public class RecoveredContainerLaunch extends ContainerLaunch {
   {
     super(context, configuration, dispatcher, exec, app, container, dirsHandler,
       containerManager);
-    this.shouldLaunchContainer.set(true);
+    this.containerAlreadyLaunched.set(true);
   }
 
   /**

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainersMonitor.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainersMonitor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainersMonitor.java
index 1069b4f..64831e9 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainersMonitor.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainersMonitor.java
@@ -19,29 +19,51 @@
 package org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor;
 
 import org.apache.hadoop.service.Service;
+import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.api.records.ResourceUtilization;
 import org.apache.hadoop.yarn.event.EventHandler;
 import org.apache.hadoop.yarn.server.nodemanager.ResourceView;
-import org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor.ContainersMonitorImpl.ProcessTreeInfo;
 
 public interface ContainersMonitor extends Service,
     EventHandler<ContainersMonitorEvent>, ResourceView {
-  public ResourceUtilization getContainersUtilization();
+  ResourceUtilization getContainersUtilization();
 
-  ResourceUtilization getContainersAllocation();
-
-  boolean hasResourcesAvailable(ProcessTreeInfo pti);
-
-  void increaseContainersAllocation(ProcessTreeInfo pti);
-
-  void decreaseContainersAllocation(ProcessTreeInfo pti);
-
-  void increaseResourceUtilization(ResourceUtilization resourceUtil,
-      ProcessTreeInfo pti);
-
-  void decreaseResourceUtilization(ResourceUtilization resourceUtil,
-      ProcessTreeInfo pti);
+  float getVmemRatio();
 
   void subtractNodeResourcesFromResourceUtilization(
       ResourceUtilization resourceUtil);
+
+  /**
+   * Utility method to add a {@link Resource} to the
+   * {@link ResourceUtilization}.
+   * @param containersMonitor Containers Monitor.
+   * @param resourceUtil Resource Utilization.
+   * @param resource Resource.
+   */
+  static void increaseResourceUtilization(
+      ContainersMonitor containersMonitor, ResourceUtilization resourceUtil,
+      Resource resource) {
+    float vCores = (float) resource.getVirtualCores() /
+        containersMonitor.getVCoresAllocatedForContainers();
+    int vmem = (int) (resource.getMemorySize()
+        * containersMonitor.getVmemRatio());
+    resourceUtil.addTo((int)resource.getMemorySize(), vmem, vCores);
+  }
+
+  /**
+   * Utility method to subtract a {@link Resource} from the
+   * {@link ResourceUtilization}.
+   * @param containersMonitor Containers Monitor.
+   * @param resourceUtil Resource Utilization.
+   * @param resource Resource.
+   */
+  static void decreaseResourceUtilization(
+      ContainersMonitor containersMonitor, ResourceUtilization resourceUtil,
+      Resource resource) {
+    float vCores = (float) resource.getVirtualCores() /
+        containersMonitor.getVCoresAllocatedForContainers();
+    int vmem = (int) (resource.getMemorySize()
+        * containersMonitor.getVmemRatio());
+    resourceUtil.subtractFrom((int)resource.getMemorySize(), vmem, vCores);
+  }
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3219b7b4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainersMonitorImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainersMonitorImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainersMonitorImpl.java
index fad2b6a..1914b42 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainersMonitorImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainersMonitorImpl.java
@@ -97,9 +97,6 @@ public class ContainersMonitorImpl extends AbstractService implements
   }
 
   private ResourceUtilization containersUtilization;
-  // Tracks the aggregated allocation of the currently allocated containers
-  // when queuing of containers at the NMs is enabled.
-  private final ResourceUtilization containersAllocation;
 
   private volatile boolean stopped = false;
 
@@ -114,7 +111,6 @@ public class ContainersMonitorImpl extends AbstractService implements
     this.monitoringThread = new MonitoringThread();
 
     this.containersUtilization = ResourceUtilization.newInstance(0, 0, 0.0f);
-    this.containersAllocation = ResourceUtilization.newInstance(0, 0, 0.0f);
   }
 
   @Override
@@ -743,6 +739,8 @@ public class ContainersMonitorImpl extends AbstractService implements
       LOG.warn("Container " + containerId.toString() + "does not exist");
       return;
     }
+    // YARN-5860: Route this through the ContainerScheduler to
+    //       fix containerAllocation
     container.setResource(resource);
   }
 
@@ -842,67 +840,6 @@ public class ContainersMonitorImpl extends AbstractService implements
     this.containersUtilization = utilization;
   }
 
-  public ResourceUtilization getContainersAllocation() {
-    return this.containersAllocation;
-  }
-
-  /**
-   * @return true if there are available allocated resources for the given
-   *         container to start.
-   */
-  @Override
-  public boolean hasResourcesAvailable(ProcessTreeInfo pti) {
-    synchronized (this.containersAllocation) {
-      // Check physical memory.
-      if (this.containersAllocation.getPhysicalMemory() +
-          (int) (pti.getPmemLimit() >> 20) >
-          (int) (getPmemAllocatedForContainers() >> 20)) {
-        return false;
-      }
-      // Check virtual memory.
-      if (isVmemCheckEnabled() &&
-          this.containersAllocation.getVirtualMemory() +
-          (int) (pti.getVmemLimit() >> 20) >
-          (int) (getVmemAllocatedForContainers() >> 20)) {
-        return false;
-      }
-      // Check CPU.
-      if (this.containersAllocation.getCPU()
-          + allocatedCpuUsage(pti) > 1.0f) {
-        return false;
-      }
-    }
-    return true;
-  }
-
-  @Override
-  public void increaseContainersAllocation(ProcessTreeInfo pti) {
-    synchronized (this.containersAllocation) {
-      increaseResourceUtilization(this.containersAllocation, pti);
-    }
-  }
-
-  @Override
-  public void decreaseContainersAllocation(ProcessTreeInfo pti) {
-    synchronized (this.containersAllocation) {
-      decreaseResourceUtilization(this.containersAllocation, pti);
-    }
-  }
-
-  @Override
-  public void increaseResourceUtilization(ResourceUtilization resourceUtil,
-      ProcessTreeInfo pti) {
-    resourceUtil.addTo((int) (pti.getPmemLimit() >> 20),
-        (int) (pti.getVmemLimit() >> 20), allocatedCpuUsage(pti));
-  }
-
-  @Override
-  public void decreaseResourceUtilization(ResourceUtilization resourceUtil,
-      ProcessTreeInfo pti) {
-    resourceUtil.subtractFrom((int) (pti.getPmemLimit() >> 20),
-        (int) (pti.getVmemLimit() >> 20), allocatedCpuUsage(pti));
-  }
-
   @Override
   public void subtractNodeResourcesFromResourceUtilization(
       ResourceUtilization resourceUtil) {
@@ -910,14 +847,9 @@ public class ContainersMonitorImpl extends AbstractService implements
         (int) (getVmemAllocatedForContainers() >> 20), 1.0f);
   }
 
-  /**
-   * Calculates the vCores CPU usage that is assigned to the given
-   * {@link ProcessTreeInfo}. In particular, it takes into account the number of
-   * vCores that are allowed to be used by the NM and returns the CPU usage
-   * as a normalized value between {@literal >=} 0 and {@literal <=} 1.
-   */
-  private float allocatedCpuUsage(ProcessTreeInfo pti) {
-    return (float) pti.getCpuVcores() / getVCoresAllocatedForContainers();
+  @Override
+  public float getVmemRatio() {
+    return vmemRatio;
   }
 
   @Override
@@ -988,5 +920,4 @@ public class ContainersMonitorImpl extends AbstractService implements
             startEvent.getVmemLimit(), startEvent.getPmemLimit(),
             startEvent.getCpuVcores()));
   }
-
 }


---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org


Mime
View raw message