[GitHub] storm pull request #2881: STORM-3259: NUMA Support for Storm
Github user govind-menon commented on a diff in the pull request:
https://github.com/apache/storm/pull/2881#discussion_r229883970
--- Diff:
storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java ---
@@ -414,6 +418,34 @@ public void validateInteger(String name, Object o) {
}
}
+public static class NumaEntryValidator extends Validator {
+
+@Override
+public void validateField(String name, Object o) {
+if (o == null) {
+return;
+}
+Map numa = (Map) o;
+for (String key : new String[]{NUMA_CORES, NUMA_MEMORY_IN_MB,
NUMA_PORTS}) {
+if (!numa.containsKey(key)) {
+throw new IllegalArgumentException(
+"The numa configuration key [" + key + "] is
missing!"
+);
+}
+}
+
+List cores = (List) numa.get(NUMA_CORES);
+Set coreSet = new HashSet();
+coreSet.addAll(cores);
+if (coreSet.size() != cores.size()) {
--- End diff --
We don't - the only reason I'm even keeping the core list as opposed to
number of cores is if in the future users wanted even more fine grained control
of core pinning. In that case duplicates would mean they wanted the core to be
used by multiple node
---
[GitHub] storm pull request #2881: STORM-3259: NUMA Support for Storm
Github user Ethanlm commented on a diff in the pull request:
https://github.com/apache/storm/pull/2881#discussion_r229880617
--- Diff:
storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java ---
@@ -414,6 +418,34 @@ public void validateInteger(String name, Object o) {
}
}
+public static class NumaEntryValidator extends Validator {
+
+@Override
+public void validateField(String name, Object o) {
+if (o == null) {
+return;
+}
+Map numa = (Map) o;
+for (String key : new String[]{NUMA_CORES, NUMA_MEMORY_IN_MB,
NUMA_PORTS}) {
+if (!numa.containsKey(key)) {
+throw new IllegalArgumentException(
+"The numa configuration key [" + key + "] is
missing!"
+);
+}
+}
+
+List cores = (List) numa.get(NUMA_CORES);
+Set coreSet = new HashSet();
+coreSet.addAll(cores);
+if (coreSet.size() != cores.size()) {
--- End diff --
Thanks for adding this. Do we care about duplicate cores across the numa
zones? (I don't
for example.
```
numaid=0, cores=[0,1,2,3]
numaid=1, cores=[0,1,2,3]
```
---
[GitHub] storm pull request #2881: STORM-3259: NUMA Support for Storm
Github user Ethanlm commented on a diff in the pull request:
https://github.com/apache/storm/pull/2881#discussion_r229879326
--- Diff:
storm-server/src/main/java/org/apache/storm/daemon/supervisor/BasicContainer.java
---
@@ -607,6 +608,27 @@ protected String javaCmd(String cmd) {
return ret;
}
+/**
+ * Extracting out to mock it for tests.
+ * @return true if on Linux.
+ */
+protected boolean isOnLinux() {
+return SystemUtils.IS_OS_LINUX;
+}
+
+private void prefixNumaPinningIfApplicable(String numaId, List
commandList) {
+if (numaId != null) {
+if (isOnLinux()) {
+commandList.add("numactl");
+commandList.add("--i=" + numaId);
--- End diff --
sorry to keep beating you on this :D How about:
```
commandList.add(0, "numactl");
commandList.add(1, "--i=" + numaId);
```
since we want to add prefix to `commandList` no matter it's empty or not
---
[GitHub] storm pull request #2881: STORM-3259: NUMA Support for Storm
Github user Ethanlm commented on a diff in the pull request:
https://github.com/apache/storm/pull/2881#discussion_r229876743
--- Diff:
storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java ---
@@ -414,6 +418,34 @@ public void validateInteger(String name, Object o) {
}
}
+public static class NumaEntryValidator extends Validator {
+
+@Override
+public void validateField(String name, Object o) {
+if (o == null) {
+return;
+}
+Map numa = (Map) o;
+for (String key : new String[]{NUMA_CORES, NUMA_MEMORY_IN_MB,
NUMA_PORTS}) {
+if (!numa.containsKey(key)) {
+throw new IllegalArgumentException(
+"The numa configuration key [" + key + "] is
missing!"
+);
+}
+}
+
+List cores = (List) numa.get(NUMA_CORES);
+Set coreSet = new HashSet();
+coreSet.addAll(cores);
+if (coreSet.size() != cores.size()) {
+throw new IllegalArgumentException(
+"No duplicate cores in NUMA config"
--- End diff --
Better to be "duplicate cores in NUMA config"?
---
[GitHub] storm pull request #2881: STORM-3259: NUMA Support for Storm
Github user Ethanlm commented on a diff in the pull request:
https://github.com/apache/storm/pull/2881#discussion_r229878430
--- Diff:
storm-server/src/main/java/org/apache/storm/daemon/supervisor/BasicContainer.java
---
@@ -607,6 +608,27 @@ protected String javaCmd(String cmd) {
return ret;
}
+/**
+ * Extracting out to mock it for tests.
+ * @return true if on Linux.
+ */
+protected boolean isOnLinux() {
+return SystemUtils.IS_OS_LINUX;
+}
+
+private void prefixNumaPinningIfApplicable(String numaId, List
commandList) {
+if (numaId != null) {
+if (isOnLinux()) {
+commandList.add("numactl");
+commandList.add("--i=" + numaId);
+return;
+} else {
+// TODO : Add support for pinning on Windows host
+throw new RuntimeException("numactl pinning currently not
supported on non-Linux hosts");
--- End diff --
Maybe `UnsupportedOperationException` is a better option. But I am fine
with this.
---
[GitHub] storm pull request #2881: STORM-3259: NUMA Support for Storm
Github user Ethanlm commented on a diff in the pull request:
https://github.com/apache/storm/pull/2881#discussion_r229875813
--- Diff: storm-client/src/jvm/org/apache/storm/Config.java ---
@@ -1041,6 +1041,24 @@
@isPositiveNumber
@NotNull
public static final String SUPERVISOR_WORKER_TIMEOUT_SECS =
"supervisor.worker.timeout.secs";
+
+/**
+ * A map with keys mapped to each NUMA Node on the supervisor that
will be used
+ * by scheduler. CPUs, memory and ports available on each NUMA node
will be provided.
+ * Each supervisor will have different map of NUMAs.
+ * Example: "supervisor.numa.meta": {
+ * "0": { "memory.mb": 122880, "cores": [ 0, 12, 1, 13, 2, 14, 3, 15,
4, 16, 5, 17],
+ * "ports": [6700, 6701]},
+ * "1" : {"memory.mb": 122880, "cores": [ 6, 18, 7, 19, 8, 20, 9, 21,
10, 22, 11, 23],
--- End diff --
"ports" is numa.ports" now since ` "NUMA_PORTS = "numa.ports"`
---
[GitHub] storm pull request #2881: STORM-3259: NUMA Support for Storm
Github user govind-menon commented on a diff in the pull request:
https://github.com/apache/storm/pull/2881#discussion_r229869065
--- Diff:
storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java ---
@@ -622,22 +624,26 @@ public boolean areAllConnectionsReady() {
return this.autoCredentials;
}
-private List> readWorkerExecutors(IStormClusterState
stormClusterState, String topologyId, String assignmentId,
+private List> readWorkerExecutors(IStormClusterState
stormClusterState,
+ String topologyId, String
assignmentId,
int port) {
LOG.info("Reading assignments");
List> executorsAssignedToThisWorker = new ArrayList<>();
executorsAssignedToThisWorker.add(Constants.SYSTEM_EXECUTOR_ID);
-Map, NodeInfo> executorToNodePort =
getLocalAssignment(conf, stormClusterState,
topologyId).get_executor_node_port();
+Map, NodeInfo> executorToNodePort =
+getLocalAssignment(conf, stormClusterState,
topologyId).get_executor_node_port();
for (Map.Entry, NodeInfo> entry :
executorToNodePort.entrySet()) {
NodeInfo nodeInfo = entry.getValue();
-if (nodeInfo.get_node().equals(assignmentId) &&
nodeInfo.get_port().iterator().next() == port) {
+if (nodeInfo.get_node().startsWith(assignmentId) &&
nodeInfo.get_port().iterator().next() == port) {
--- End diff --
The supervisor Id is unique and unlikely to be the same
---
[GitHub] storm pull request #2881: STORM-3259: NUMA Support for Storm
Github user govind-menon commented on a diff in the pull request:
https://github.com/apache/storm/pull/2881#discussion_r229854058
--- Diff:
storm-server/src/test/java/org/apache/storm/daemon/supervisor/BasicContainerTest.java
---
@@ -319,7 +319,15 @@ public void testLaunch() throws Exception {
superConf.put(Config.STORM_WORKERS_ARTIFACTS_DIR, stormLocal);
superConf.put(DaemonConfig.STORM_LOG4J2_CONF_DIR, log4jdir);
superConf.put(Config.WORKER_CHILDOPTS, " -Dtesting=true");
-
+Map numaNode = new HashMap();
+numaNode.put(Utils.NUMA_ID, "0");
+numaNode.put(Utils.NUMA_CORES, Collections.singletonList("0"));
+numaNode.put(Utils.NUMAS_PORTS, Collections.singletonList(8081));
+numaNode.put(Utils.NUMA_MEMORY_IN_MB, 2048);
+Map numaMap = new HashMap();
+numaMap.put(Utils.NUMAS_BASE, Collections.singletonList(numaNode));
+
+superConf.put(Config.SUPERVISOR_NUMA_META, numaMap);
--- End diff --
No i explicitly want this so that we show that there is no numa pinning
when there is no config.
---
[GitHub] storm pull request #2881: STORM-3259: NUMA Support for Storm
Github user govind-menon commented on a diff in the pull request:
https://github.com/apache/storm/pull/2881#discussion_r229853904
--- Diff:
storm-server/src/main/java/org/apache/storm/scheduler/resource/normalization/NormalizedResourceOffer.java
---
@@ -81,6 +81,20 @@ public void add(NormalizedResourcesWithMemory other) {
totalMemoryMb += other.getTotalMemoryMb();
}
+/**
+ * Remove the resources in other from this.
+ * @param other the resources to be removed.
+ * @return true if one or more resources in other were larger than
available resources in this, else false.
+ */
+public boolean remove(NormalizedResourcesWithMemory other) {
--- End diff --
I considered that - but my approach seems cleaner so that in the place
where it's called you don't have to wonder about the null.
---
[GitHub] storm-site pull request #7: Add XenonStack to powered-by
Github user asfgit closed the pull request at: https://github.com/apache/storm-site/pull/7 ---
[GitHub] storm pull request #2896: [Storm-3261] Fix broken links in documentation
Github user asfgit closed the pull request at: https://github.com/apache/storm/pull/2896 ---
