Skip to content

YARN-11196. NUMA support in DCE #4742

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
PrabhuJoseph merged 5 commits into from
Aug 30, 2022
Merged

YARN-11196. NUMA support in DCE #4742

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
c80e340
Numa support in DCE
Aug 14, 2022
84af21f
Revert "Numa support in DCE"
Samrat002 Aug 14, 2022
6172d5f
Numa support in DCE
Samrat002 Aug 14, 2022
0d5151e
address to comments
Samrat002 Aug 25, 2022
125c50e
Fix test case for reaquireContainer
Samrat002 Aug 25, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
148 changes: 138 additions & 10 deletions ...ger/src/main/java/org/apache/hadoop/yarn/server/nodemanager/DefaultContainerExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@

import static org.apache.hadoop.fs.CreateFlag.CREATE;
import static org.apache.hadoop.fs.CreateFlag.OVERWRITE;
import static org.apache.hadoop.yarn.conf.YarnConfiguration.numaAwarenessEnabled;

import org.apache.hadoop.classification.VisibleForTesting;
import java.io.DataOutputStream;
Expand All @@ -28,12 +29,13 @@
import java.io.IOException;
import java.io.PrintStream;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.ArrayList;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;

import org.apache.commons.lang3.RandomUtils;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.fs.FileContext;
Expand All @@ -51,14 +53,19 @@
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.ConfigurationException;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerDiagnosticsUpdateEvent;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainerLaunch;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.ResourceHandlerException;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.numa.NumaResourceAllocation;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.numa.NumaResourceAllocator;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.ContainerLocalizer;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.runtime.ContainerExecutionException;
import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerExecContext;
import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerLivenessContext;
import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerReacquisitionContext;
import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerReapContext;
import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerSignalContext;
import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerStartContext;
Expand Down Expand Up @@ -86,6 +93,8 @@ public class DefaultContainerExecutor extends ContainerExecutor {

private String logDirPermissions = null;

private NumaResourceAllocator numaResourceAllocator;

/**
* Default constructor for use in testing.
*/
Expand Down Expand Up @@ -137,7 +146,17 @@ protected void setScriptExecutable(Path script, String owner)

@Override
public void init(Context nmContext) throws IOException {
// nothing to do or verify here
if(numaAwarenessEnabled(getConf())) {
numaResourceAllocator = new NumaResourceAllocator(nmContext);
try {
numaResourceAllocator.init(this.getConf());
LOG.info("NUMA resources allocation is enabled in DefaultContainer Executor," +
" Successfully initialized NUMA resources allocator.");
} catch (YarnException e) {
LOG.warn("Improper NUMA configuration provided.", e);
throw new IOException("Failed to initialize configured numa subsystem!");
}
}
}

@Override
Expand Down Expand Up @@ -300,11 +319,28 @@ public int launchContainer(ContainerStartContext ctx)
setScriptExecutable(launchDst, user);
setScriptExecutable(sb.getWrapperScriptPath(), user);

// adding numa commands based on configuration
String[] numaCommands = new String[]{};

if (numaResourceAllocator != null) {
try {
NumaResourceAllocation numaResourceAllocation =
numaResourceAllocator.allocateNumaNodes(container);
if (numaResourceAllocation != null) {
numaCommands = getNumaCommands(numaResourceAllocation);
}
} catch (ResourceHandlerException e) {
LOG.error("NumaResource Allocation failed!", e);
throw new IOException("NumaResource Allocation Error!", e);
}
}

shExec = buildCommandExecutor(sb.getWrapperScriptPath().toString(),
containerIdStr, user, pidFile, container.getResource(),
new File(containerWorkDir.toUri().getPath()),
container.getLaunchContext().getEnvironment());

containerIdStr, user, pidFile, container.getResource(),
new File(containerWorkDir.toUri().getPath()),
container.getLaunchContext().getEnvironment(),
numaCommands);

if (isContainerActive(containerId)) {
shExec.execute();
} else {
Expand Down Expand Up @@ -350,6 +386,7 @@ public int launchContainer(ContainerStartContext ctx)
return exitCode;
} finally {
if (shExec != null) shExec.close();
postComplete(containerId);
}
return 0;
}
Expand All @@ -372,16 +409,19 @@ public int relaunchContainer(ContainerStartContext ctx)
* as the current working directory for the command. If null,
* the current working directory is not modified.
* @param environment the container environment
* @param numaCommands list of prefix numa commands
* @return the new {@link ShellCommandExecutor}
* @see ShellCommandExecutor
*/
protected CommandExecutor buildCommandExecutor(String wrapperScriptPath,
String containerIdStr, String user, Path pidFile, Resource resource,
File workDir, Map<String, String> environment) {
protected CommandExecutor buildCommandExecutor(String wrapperScriptPath,
String containerIdStr, String user, Path pidFile, Resource resource,
File workDir, Map<String, String> environment, String[] numaCommands) {

String[] command = getRunCommand(wrapperScriptPath,
containerIdStr, user, pidFile, this.getConf(), resource);

command = concatStringCommands(command, numaCommands);
Copy link
Contributor

@PrabhuJoseph PrabhuJoseph Aug 25, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shall we skip calling this when numaCommands is not passed.


LOG.info("launchContainer: {}", Arrays.toString(command));
return new ShellCommandExecutor(
command,
Expand Down Expand Up @@ -1040,4 +1080,92 @@ public void updateYarnSysFS(Context ctx, String user,
String appId, String spec) throws IOException {
throw new ServiceStateException("Implementation unavailable");
}

@Override
public int reacquireContainer(ContainerReacquisitionContext ctx)
throws IOException, InterruptedException {
try {
if (numaResourceAllocator != null) {
numaResourceAllocator.recoverNumaResource(ctx.getContainerId());
}
return super.reacquireContainer(ctx);
} finally {
postComplete(ctx.getContainerId());
}
}

/**
* clean up and release of resources.
*
* @param containerId containerId of running container
*/
public void postComplete(final ContainerId containerId) {
if (numaResourceAllocator != null) {
try {
numaResourceAllocator.releaseNumaResource(containerId);
} catch (ResourceHandlerException e) {
LOG.warn("NumaResource release failed for " +
"containerId: {}. Exception: ", containerId, e);
}
}
}

/**
* @param resourceAllocation NonNull NumaResourceAllocation object reference
* @return Array of numa specific commands
*/
String[] getNumaCommands(NumaResourceAllocation resourceAllocation) {
String[] numaCommand = new String[3];
numaCommand[0] = this.getConf().get(YarnConfiguration.NM_NUMA_AWARENESS_NUMACTL_CMD,
Copy link
Contributor

@PrabhuJoseph PrabhuJoseph Aug 25, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Better to read this config and initialize in the init.

YarnConfiguration.DEFAULT_NM_NUMA_AWARENESS_NUMACTL_CMD);
numaCommand[1] = "--interleave=" + String.join(",", resourceAllocation.getMemNodes());
numaCommand[2] = "--cpunodebind=" + String.join(",", resourceAllocation.getCpuNodes());
return numaCommand;

}

/**
* @param firstStringArray Array of String
* @param secondStringArray Array of String
* @return combined array of string where first elements are from firstStringArray
* and later are the elements from secondStringArray
*/
String[] concatStringCommands(String[] firstStringArray, String[] secondStringArray) {
Copy link
Contributor

@PrabhuJoseph PrabhuJoseph Aug 25, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The Numa Commands has to be in prefix.


if(firstStringArray == null && secondStringArray == null) {
return null;
}

int len = 0;

if(firstStringArray != null){
len = len + firstStringArray.length;
}

if(secondStringArray != null){
len = len + secondStringArray.length;
}

if (len == 0) {
return new String[]{};
}

String[] ret = new String[len];
int idx = 0;
for (int i=0; firstStringArray !=null && i < firstStringArray.length; i++) {
ret[idx] = firstStringArray[i];
idx++;
}
for (int i=0; secondStringArray !=null && i < secondStringArray.length; i++) {
ret[idx] = secondStringArray[i];
idx++;
}
return ret;
}

@VisibleForTesting
public void setNumaResourceAllocator(NumaResourceAllocator numaResourceAllocator) {
this.numaResourceAllocator = numaResourceAllocator;
}

}
4 changes: 2 additions & 2 deletions ...c/main/java/org/apache/hadoop/yarn/server/nodemanager/WindowsSecureContainerExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -718,10 +718,10 @@ public void startLocalizer(LocalizerStartContext ctx) throws IOException,
@Override
protected CommandExecutor buildCommandExecutor(String wrapperScriptPath,
String containerIdStr, String userName, Path pidFile, Resource resource,
File wordDir, Map<String, String> environment) {
File wordDir, Map<String, String> environment, String[] numaCommands) {
return new WintuilsProcessStubExecutor(
wordDir.toString(),
containerIdStr, userName, pidFile.toString(),
containerIdStr, userName, pidFile.toString(),
"cmd /c " + wrapperScriptPath);
}

Expand Down
2 changes: 1 addition & 1 deletion .../yarn/server/nodemanager/containermanager/linux/resources/numa/NumaResourceAllocator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,7 +143,7 @@ public void init(Configuration conf) throws YarnException {
}

@VisibleForTesting
String executeNGetCmdOutput(Configuration conf) throws YarnException {
public String executeNGetCmdOutput(Configuration conf) throws YarnException {
String numaCtlCmd = conf.get(
YarnConfiguration.NM_NUMA_AWARENESS_NUMACTL_CMD,
YarnConfiguration.DEFAULT_NM_NUMA_AWARENESS_NUMACTL_CMD);
Expand Down
Loading