YARN-11200 Backport numa to branch-2.10

Author: Samrat002

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java

@@ -3524,6 +3524,22 @@ public static boolean areNodeLabelsEnabled(
   public static final String DEFAULT_YARN_WORKFLOW_ID_TAG_PREFIX =
       "workflowid:";
 
+  /**
+   * Settings for NUMA awareness.
+   */
+  public static final String NM_NUMA_AWARENESS_ENABLED = NM_PREFIX
+      + "numa-awareness.enabled";
+  public static final boolean DEFAULT_NM_NUMA_AWARENESS_ENABLED = false;
+  public static final String NM_NUMA_AWARENESS_READ_TOPOLOGY = NM_PREFIX
+      + "numa-awareness.read-topology";
+  public static final boolean DEFAULT_NM_NUMA_AWARENESS_READ_TOPOLOGY = false;
+  public static final String NM_NUMA_AWARENESS_NODE_IDS = NM_PREFIX
+      + "numa-awareness.node-ids";
+  public static final String NM_NUMA_AWARENESS_NUMACTL_CMD = NM_PREFIX
+      + "numa-awareness.numactl.cmd";
+  public static final String DEFAULT_NM_NUMA_AWARENESS_NUMACTL_CMD =
+      "/usr/bin/numactl";
+
   public YarnConfiguration() {
     super();
   }
@@ -3535,6 +3551,17 @@ public YarnConfiguration(Configuration conf) {
     }
   }
 
+  /**
+   * Returns whether the NUMA awareness is enabled.
+   *
+   * @param conf the configuration
+   * @return whether the NUMA awareness is enabled.
+   */
+  public static boolean numaAwarenessEnabled(Configuration conf) {
+    return conf.getBoolean(NM_NUMA_AWARENESS_ENABLED,
+        DEFAULT_NM_NUMA_AWARENESS_ENABLED);
+  }
+
   @Private
   public static List<String> getServiceAddressConfKeys(Configuration conf) {
     return useHttps(conf) ? RM_SERVICES_ADDRESS_CONF_KEYS_HTTPS

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml

@@ -3680,4 +3680,56 @@
     <name>yarn.client.load.resource-types.from-server</name>
     <value>false</value>
   </property>
+
+  <property>
+    <description>
+      Whether to enable the NUMA awareness for containers in Node Manager.
+    </description>
+    <name>yarn.nodemanager.numa-awareness.enabled</name>
+    <value>false</value>
+  </property>
+
+  <property>
+    <description>
+      Whether to read the NUMA topology from the system or from the
+      configurations. If the value is true then NM reads the NUMA topology from
+      system using the command 'numactl --hardware'. If the value is false then NM
+      reads the topology from the configurations
+      'yarn.nodemanager.numa-awareness.node-ids'(for node id's),
+      'yarn.nodemanager.numa-awareness.&lt;NODE_ID&gt;.memory'(for each node memory),
+      'yarn.nodemanager.numa-awareness.&lt;NODE_ID&gt;.cpus'(for each node cpus).
+    </description>
+    <name>yarn.nodemanager.numa-awareness.read-topology</name>
+    <value>false</value>
+  </property>
+
+  <property>
+    <description>
+      NUMA node id's in the form of comma separated list. Memory and No of CPUs
+      will be read using the properties
+      'yarn.nodemanager.numa-awareness.&lt;NODE_ID&gt;.memory' and
+      'yarn.nodemanager.numa-awareness.&lt;NODE_ID&gt;.cpus' for each id specified
+      in this value. This property value will be read only when
+      'yarn.nodemanager.numa-awareness.read-topology=false'.
+
+      For example, if yarn.nodemanager.numa-awareness.node-ids=0,1
+      then need to specify memory and cpus for node id's '0' and '1' like below,
+      yarn.nodemanager.numa-awareness.0.memory=73717
+      yarn.nodemanager.numa-awareness.0.cpus=4
+      yarn.nodemanager.numa-awareness.1.memory=73727
+      yarn.nodemanager.numa-awareness.1.cpus=4
+    </description>
+    <name>yarn.nodemanager.numa-awareness.node-ids</name>
+    <value></value>
+  </property>
+
+  <property>
+    <description>
+      The numactl command path which controls NUMA policy for processes or
+      shared memory.
+    </description>
+    <name>yarn.nodemanager.numa-awareness.numactl.cmd</name>
+    <value>/usr/bin/numactl</value>
+  </property>
+
 </configuration>

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/LinuxContainerExecutor.java

@@ -111,6 +111,7 @@ public class LinuxContainerExecutor extends ContainerExecutor {
   private boolean containerLimitUsers;
   private ResourceHandler resourceHandlerChain;
   private LinuxContainerRuntime linuxContainerRuntime;
+  private Context nmContext;
 
   /**
    * The container exit code.
@@ -262,6 +263,13 @@ protected String getContainerExecutorExecutablePath(Configuration conf) {
         defaultPath);
   }
 
+  private void addNumaArgsToCommand(List<String> prefixCommands,
+      List<String> numaArgs) {
+    if (numaArgs != null) {
+      prefixCommands.addAll(numaArgs);
+    }
+  }
+
   /**
    * Add a niceness level to the process that will be executed.  Adds
    * {@code -n <nice>} to the given command. The niceness level will be
@@ -283,6 +291,7 @@ protected PrivilegedOperationExecutor getPrivilegedOperationExecutor() {
 
   @Override
   public void init(Context nmContext) throws IOException {
+    this.nmContext = nmContext;
     Configuration conf = super.getConf();
 
     // Send command to executor which will just start up,
@@ -475,6 +484,7 @@ public int launchContainer(ContainerStartContext ctx)
             container.getResource());
     String resourcesOptions = resourcesHandler.getResourcesOption(containerId);
     String tcCommandFile = null;
+    List<String> numaArgs = null;
 
     try {
       if (resourceHandlerChain != null) {
@@ -496,6 +506,9 @@ public int launchContainer(ContainerStartContext ctx)
             case TC_MODIFY_STATE:
               tcCommandFile = op.getArguments().get(0);
               break;
+            case ADD_NUMA_PARAMS:
+              numaArgs = op.getArguments();
+              break;
             default:
               LOG.warn("PrivilegedOperation type unsupported in launch: "
                   + op.getOperationType());
@@ -529,6 +542,7 @@ public int launchContainer(ContainerStartContext ctx)
             .Builder(container);
 
         addSchedPriorityCommand(prefixCommands);
+        addNumaArgsToCommand(prefixCommands, numaArgs);
         if (prefixCommands.size() > 0) {
           builder.setExecutionAttribute(CONTAINER_LAUNCH_PREFIX_COMMANDS,
               prefixCommands);

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/privileged/PrivilegedOperation.java

@@ -52,7 +52,8 @@ public enum OperationType {
     ADD_PID_TO_CGROUP(""), //no CLI switch supported yet.
     RUN_DOCKER_CMD("--run-docker"),
     GPU("--module-gpu"),
-    LIST_AS_USER(""); //no CLI switch supported yet.
+    LIST_AS_USER(""), //no CLI switch supported yet.
+    ADD_NUMA_PARAMS(""); // no CLI switch supported yet.
 
     private final String option;
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/resources/ResourceHandlerModule.java

@@ -27,6 +27,7 @@
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.server.nodemanager.Context;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.privileged.PrivilegedOperationExecutor;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.numa.NumaResourceHandlerImpl;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.resourceplugin.ResourcePlugin;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.resourceplugin.ResourcePluginManager;
 import org.apache.hadoop.yarn.server.nodemanager.util.CgroupsLCEResourcesHandler;
@@ -208,6 +209,14 @@ private static void addHandlerIfNotNull(List<ResourceHandler> handlerList,
     }
   }
 
+  private static ResourceHandler getNumaResourceHandler(Configuration conf,
+      Context nmContext) {
+    if (YarnConfiguration.numaAwarenessEnabled(conf)) {
+      return new NumaResourceHandlerImpl(conf, nmContext);
+    }
+    return null;
+  }
+
   private static void initializeConfiguredResourceHandlerChain(
       Configuration conf, Context nmContext)
       throws ResourceHandlerException {
@@ -218,6 +227,7 @@ private static void initializeConfiguredResourceHandlerChain(
     addHandlerIfNotNull(handlerList, getMemoryResourceHandler(conf));
     addHandlerIfNotNull(handlerList, getCGroupsCpuResourceHandler(conf));
     addHandlersFromConfiguredResourcePlugins(handlerList, conf, nmContext);
+    addHandlerIfNotNull(handlerList, getNumaResourceHandler(conf, nmContext));
     resourceHandlerChain = new ResourceHandlerChain(handlerList);
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/resources/numa/NumaNodeResource.java

@@ -0,0 +1,203 @@
+
+/**
+ * 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.nodemanager.containermanager.linux.resources.numa;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.Resource;
+
+/**
+ * NumaNodeResource class holds the NUMA node topology with the total and used
+ * resources.
+ */
+public class NumaNodeResource {
+  private String nodeId;
+  private long totalMemory;
+  private int totalCpus;
+  private long usedMemory;
+  private int usedCpus;
+
+  private static final Logger LOG = LoggerFactory.
+      getLogger(NumaNodeResource.class);
+
+  private Map<ContainerId, Long> containerVsMemUsage =
+      new ConcurrentHashMap<>();
+  private Map<ContainerId, Integer> containerVsCpusUsage =
+      new ConcurrentHashMap<>();
+
+  public NumaNodeResource(String nodeId, long totalMemory, int totalCpus) {
+    this.nodeId = nodeId;
+    this.totalMemory = totalMemory;
+    this.totalCpus = totalCpus;
+  }
+
+  /**
+   * Checks whether the specified resources available or not.
+   *
+   * @param resource resource
+   * @return whether the specified resources available or not
+   */
+  public boolean isResourcesAvailable(Resource resource) {
+    LOG.debug(
+        "Memory available:" + (totalMemory - usedMemory) + ", CPUs available:"
+            + (totalCpus - usedCpus) + ", requested:" + resource);
+    if ((totalMemory - usedMemory) >= resource.getMemorySize()
+        && (totalCpus - usedCpus) >= resource.getVirtualCores()) {
+      return true;
+    }
+    return false;
+  }
+
+  /**
+   * Assigns available memory and returns the remaining needed memory.
+   *
+   * @param memreq required memory
+   * @param containerId which container memory to assign
+   * @return remaining needed memory
+   */
+  public long assignAvailableMemory(long memreq, ContainerId containerId) {
+    long memAvailable = totalMemory - usedMemory;
+    if (memAvailable >= memreq) {
+      containerVsMemUsage.put(containerId, memreq);
+      usedMemory += memreq;
+      return 0;
+    } else {
+      usedMemory += memAvailable;
+      containerVsMemUsage.put(containerId, memAvailable);
+      return memreq - memAvailable;
+    }
+  }
+
+  /**
+   * Assigns available cpu's and returns the remaining needed cpu's.
+   *
+   * @param cpusreq required cpu's
+   * @param containerId which container cpu's to assign
+   * @return remaining needed cpu's
+   */
+  public int assignAvailableCpus(int cpusreq, ContainerId containerId) {
+    int cpusAvailable = totalCpus - usedCpus;
+    if (cpusAvailable >= cpusreq) {
+      containerVsCpusUsage.put(containerId, cpusreq);
+      usedCpus += cpusreq;
+      return 0;
+    } else {
+      usedCpus += cpusAvailable;
+      containerVsCpusUsage.put(containerId, cpusAvailable);
+      return cpusreq - cpusAvailable;
+    }
+  }
+
+  /**
+   * Assigns the requested resources for Container.
+   *
+   * @param resource resource to assign
+   * @param containerId to which container the resources to assign
+   */
+  public void assignResources(Resource resource, ContainerId containerId) {
+    containerVsMemUsage.put(containerId, resource.getMemorySize());
+    containerVsCpusUsage.put(containerId, resource.getVirtualCores());
+    usedMemory += resource.getMemorySize();
+    usedCpus += resource.getVirtualCores();
+  }
+
+  /**
+   * Releases the assigned resources for Container.
+   *
+   * @param containerId to which container the assigned resources to release
+   */
+  public void releaseResources(ContainerId containerId) {
+    if (containerVsMemUsage.containsKey(containerId)) {
+      usedMemory -= containerVsMemUsage.get(containerId);
+      containerVsMemUsage.remove(containerId);
+    }
+    if (containerVsCpusUsage.containsKey(containerId)) {
+      usedCpus -= containerVsCpusUsage.get(containerId);
+      containerVsCpusUsage.remove(containerId);
+    }
+  }
+
+  /**
+   * Recovers the memory resources for Container.
+   *
+   * @param containerId recover the memory resources for the Container
+   * @param memory memory to recover
+   */
+  public void recoverMemory(ContainerId containerId, long memory) {
+    containerVsMemUsage.put(containerId, memory);
+    usedMemory += memory;
+  }
+
+  /**
+   * Recovers the cpu's resources for Container.
+   *
+   * @param containerId recover the cpu's resources for the Container
+   * @param cpus cpu's to recover
+   */
+  public void recoverCpus(ContainerId containerId, int cpus) {
+    containerVsCpusUsage.put(containerId, cpus);
+    usedCpus += cpus;
+  }
+
+  @Override
+  public String toString() {
+    return "Node Id:" + nodeId + "\tMemory:" + totalMemory + "\tCPus:"
+        + totalCpus;
+  }
+
+  @Override
+  public int hashCode() {
+    final int prime = 31;
+    int result = 1;
+    result = prime * result + ((nodeId == null) ? 0 : nodeId.hashCode());
+    result = prime * result + (int) (totalMemory ^ (totalMemory >>> 32));
+    result = prime * result + totalCpus;
+    return result;
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (this == obj) {
+      return true;
+    }
+    if (obj == null || getClass() != obj.getClass()) {
+      return false;
+    }
+    NumaNodeResource other = (NumaNodeResource) obj;
+    if (nodeId == null) {
+      if (other.nodeId != null) {
+        return false;
+      }
+    } else if (!nodeId.equals(other.nodeId)) {
+      return false;
+    }
+    if (totalMemory != other.totalMemory) {
+      return false;
+    }
+    return totalCpus == other.totalCpus;
+  }
+
+  public String getNodeId() {
+    return nodeId;
+  }
+}