You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2017/10/16 15:42:39 UTC
[1/2] storm git commit: [STORM-2686] Add locality awareness to
LoadAwareShuffleGrouping
Repository: storm
Updated Branches:
refs/heads/master 37403d17d -> c8947c2fe
[STORM-2686] Add locality awareness to LoadAwareShuffleGrouping
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/01bd4f82
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/01bd4f82
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/01bd4f82
Branch: refs/heads/master
Commit: 01bd4f821c940e979c360a4667c27c0477fde9a7
Parents: 352cd46
Author: Ethan Li <et...@gmail.com>
Authored: Mon Oct 9 18:41:39 2017 -0500
Committer: Ethan Li <et...@gmail.com>
Committed: Tue Oct 10 17:06:31 2017 -0500
----------------------------------------------------------------------
conf/defaults.yaml | 2 +
.../src/jvm/org/apache/storm/Config.java | 18 ++
.../apache/storm/daemon/worker/WorkerState.java | 2 +-
.../grouping/LoadAwareShuffleGrouping.java | 178 ++++++++++++++++++-
.../storm/task/GeneralTopologyContext.java | 4 +
.../storm/task/WorkerTopologyContext.java | 40 ++++-
.../grouping/LoadAwareShuffleGroupingTest.java | 37 +++-
7 files changed, 267 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/01bd4f82/conf/defaults.yaml
----------------------------------------------------------------------
diff --git a/conf/defaults.yaml b/conf/defaults.yaml
index 103b04f..ad34054 100644
--- a/conf/defaults.yaml
+++ b/conf/defaults.yaml
@@ -261,6 +261,8 @@ topology.disruptor.batch.size: 100
topology.disruptor.batch.timeout.millis: 1
topology.disable.loadaware.messaging: false
topology.state.checkpoint.interval.ms: 1000
+topology.localityaware.higher.bound.percent: 0.8
+topology.localityaware.lower.bound.percent: 0.2
# Configs for Resource Aware Scheduler
# topology priority describing the importance of the topology in decreasing importance starting from 0 (i.e. 0 is the highest priority and the priority importance decreases as the priority number increases).
http://git-wip-us.apache.org/repos/asf/storm/blob/01bd4f82/storm-client/src/jvm/org/apache/storm/Config.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/Config.java b/storm-client/src/jvm/org/apache/storm/Config.java
index e296e8f..6be0c21 100644
--- a/storm-client/src/jvm/org/apache/storm/Config.java
+++ b/storm-client/src/jvm/org/apache/storm/Config.java
@@ -65,6 +65,24 @@ public class Config extends HashMap<String, Object> {
public static final String TOPOLOGY_DISABLE_LOADAWARE_MESSAGING = "topology.disable.loadaware.messaging";
/**
+ * This signifies the load congestion among target tasks in scope. Currently it's only used in LoadAwareShuffleGrouping.
+ * When the average load is higher than the higher bound, the executor should choose target tasks in a higher scope,
+ * The scopes and their orders are: EVERYTHING > RACK_LOCAL > HOST_LOCAL > WORKER_LOCAL
+ */
+ @isPositiveNumber
+ @NotNull
+ public static final String TOPOLOGY_LOCALITYAWARE_HIGHER_BOUND_PERCENT = "topology.localityaware.higher.bound.percent";
+
+ /**
+ * This signifies the load congestion among target tasks in scope. Currently it's only used in LoadAwareShuffleGrouping.
+ * When the average load is lower than the lower bound, the executor should choose target tasks in a lower scope.
+ * The scopes and their orders are: EVERYTHING > RACK_LOCAL > HOST_LOCAL > WORKER_LOCAL
+ */
+ @isPositiveNumber
+ @NotNull
+ public static final String TOPOLOGY_LOCALITYAWARE_LOWER_BOUND_PERCENT = "topology.localityaware.lower.bound.percent";
+
+ /**
* Try to serialize all tuples, even for local transfers. This should only be used
* for testing, as a sanity check that all of your tuples are setup properly.
*/
http://git-wip-us.apache.org/repos/asf/storm/blob/01bd4f82/storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java b/storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java
index 825de4b..ec2ff59 100644
--- a/storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java
+++ b/storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java
@@ -585,7 +585,7 @@ public class WorkerState {
return new WorkerTopologyContext(systemTopology, topologyConf, taskToComponent, componentToSortedTasks,
componentToStreamToFields, topologyId, codeDir, pidDir, port, taskIds,
defaultSharedResources,
- userSharedResources);
+ userSharedResources, cachedTaskToNodePort, assignmentId);
} catch (IOException e) {
throw Utils.wrapInRuntime(e);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/01bd4f82/storm-client/src/jvm/org/apache/storm/grouping/LoadAwareShuffleGrouping.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/grouping/LoadAwareShuffleGrouping.java b/storm-client/src/jvm/org/apache/storm/grouping/LoadAwareShuffleGrouping.java
index f5b63ec..3fd75e5 100644
--- a/storm-client/src/jvm/org/apache/storm/grouping/LoadAwareShuffleGrouping.java
+++ b/storm-client/src/jvm/org/apache/storm/grouping/LoadAwareShuffleGrouping.java
@@ -19,15 +19,29 @@
package org.apache.storm.grouping;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Sets;
+
import java.io.Serializable;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.storm.Config;
import org.apache.storm.generated.GlobalStreamId;
+import org.apache.storm.generated.NodeInfo;
+import org.apache.storm.networktopography.DNSToSwitchMapping;
import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.utils.ObjectReader;
+import org.apache.storm.utils.ReflectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class LoadAwareShuffleGrouping implements LoadAwareCustomStreamGrouping, Serializable {
static final int CAPACITY = 1000;
@@ -40,8 +54,13 @@ public class LoadAwareShuffleGrouping implements LoadAwareCustomStreamGrouping,
this.index = index;
weight = MAX_WEIGHT;
}
+
+ void resetWeight() {
+ weight = MAX_WEIGHT;
+ }
}
+ private static final Logger LOG = LoggerFactory.getLogger(LoadAwareShuffleGrouping.class);
private final Map<Integer, IndexAndWeights> orig = new HashMap<>();
private Random random;
@VisibleForTesting
@@ -50,10 +69,28 @@ public class LoadAwareShuffleGrouping implements LoadAwareCustomStreamGrouping,
volatile int[] choices;
private volatile int[] prepareChoices;
private AtomicInteger current;
+ private Scope currentScope;
+ private NodeInfo sourceNodeInfo;
+ private List<Integer> targetTasks;
+ private AtomicReference<Map<Integer, NodeInfo>> taskToNodePort;
+ private Map<String, Object> conf;
+ private DNSToSwitchMapping dnsToSwitchMapping;
+ private Map<Scope, List<Integer>> localityGroup;
+ private double higherBound;
+ private double lowerBound;
@Override
public void prepare(WorkerTopologyContext context, GlobalStreamId stream, List<Integer> targetTasks) {
random = new Random();
+ sourceNodeInfo = new NodeInfo(context.getThisWorkerHost(), Sets.newHashSet((long) context.getThisWorkerPort()));
+ taskToNodePort = context.getTaskToNodePort();
+ this.targetTasks = targetTasks;
+ conf = context.getConf();
+ dnsToSwitchMapping = ReflectionUtils.newInstance((String) conf.get(Config.STORM_NETWORK_TOPOGRAPHY_PLUGIN));
+ localityGroup = new HashMap<>();
+ currentScope = Scope.WORKER_LOCAL;
+ higherBound = ObjectReader.getDouble(conf.get(Config.TOPOLOGY_LOCALITYAWARE_HIGHER_BOUND_PERCENT));
+ lowerBound = ObjectReader.getDouble(conf.get(Config.TOPOLOGY_LOCALITYAWARE_LOWER_BOUND_PERCENT));
rets = (List<Integer>[]) new List<?>[targetTasks.size()];
int i = 0;
@@ -93,12 +130,81 @@ public class LoadAwareShuffleGrouping implements LoadAwareCustomStreamGrouping,
updateRing(loadMapping);
}
+ private void refreshLocalityGroup() {
+ Map<Integer, NodeInfo> cachedTaskToNodePort = taskToNodePort.get();
+ Map<String, String> hostToRack = getHostToRackMapping(cachedTaskToNodePort);
+
+ localityGroup.values().stream().forEach(v -> v.clear());
+
+ for (int target: targetTasks) {
+ Scope scope = calculateScope(cachedTaskToNodePort, hostToRack, target);
+ if (!localityGroup.containsKey(scope)) {
+ localityGroup.put(scope, new ArrayList<>());
+ }
+ localityGroup.get(scope).add(target);
+ }
+ }
+
+ private List<Integer> getTargetsInScope(Scope scope) {
+ List<Integer> rets = new ArrayList<>();
+ List<Integer> targetInScope = localityGroup.get(scope);
+ if (null != targetInScope) {
+ rets.addAll(targetInScope);
+ }
+ Scope downgradeScope = Scope.downgrade(scope);
+ if (downgradeScope != scope) {
+ rets.addAll(getTargetsInScope(downgradeScope));
+ }
+ return rets;
+ }
+
+ private Scope transition(LoadMapping load) {
+ List<Integer> targetInScope = getTargetsInScope(currentScope);
+ if (targetInScope.isEmpty()) {
+ Scope upScope = Scope.upgrade(currentScope);
+ if (upScope == currentScope) {
+ throw new RuntimeException("This executor has no target tasks.");
+ }
+ currentScope = upScope;
+ return transition(load);
+ }
+
+ if (null == load) {
+ return currentScope;
+ }
+
+ double avg = targetInScope.stream().mapToDouble((key) -> load.get(key)).average().getAsDouble();
+ Scope nextScope;
+ if (avg < lowerBound) {
+ nextScope = Scope.downgrade(currentScope);
+ if (getTargetsInScope(nextScope).isEmpty()) {
+ nextScope = currentScope;
+ }
+ } else if (avg > higherBound) {
+ nextScope = Scope.upgrade(currentScope);
+ } else {
+ nextScope = currentScope;
+ }
+
+ return nextScope;
+ }
+
private synchronized void updateRing(LoadMapping load) {
+ refreshLocalityGroup();
+ Scope prevScope = currentScope;
+ currentScope = transition(load);
+ if (currentScope != prevScope) {
+ //reset all the weights
+ orig.values().stream().forEach(o -> o.resetWeight());
+ }
+
+ List<Integer> targetsInScope = getTargetsInScope(currentScope);
+
//We will adjust weights based off of the minimum load
- double min = load == null ? 0 : orig.keySet().stream().mapToDouble((key) -> load.get(key)).min().getAsDouble();
- for (Map.Entry<Integer, IndexAndWeights> target: orig.entrySet()) {
- IndexAndWeights val = target.getValue();
- double l = load == null ? 0.0 : load.get(target.getKey());
+ double min = load == null ? 0 : targetsInScope.stream().mapToDouble((key) -> load.get(key)).min().getAsDouble();
+ for (int target: targetsInScope) {
+ IndexAndWeights val = orig.get(target);
+ double l = load == null ? 0.0 : load.get(target);
if (l <= min + (0.05)) {
//We assume that within 5% of the minimum congestion is still fine.
//Not congested we grow (but slowly)
@@ -109,12 +215,13 @@ public class LoadAwareShuffleGrouping implements LoadAwareCustomStreamGrouping,
}
}
//Now we need to build the array
- long weightSum = orig.values().stream().mapToLong((w) -> w.weight).sum();
+ long weightSum = targetsInScope.stream().mapToLong((target) -> orig.get(target).weight).sum();
//Now we can calculate a percentage
int currentIdx = 0;
if (weightSum > 0) {
- for (IndexAndWeights indexAndWeights : orig.values()) {
+ for (int target: targetsInScope) {
+ IndexAndWeights indexAndWeights = orig.get(target);
int count = (int) ((indexAndWeights.weight / (double) weightSum) * CAPACITY);
for (int i = 0; i < count && currentIdx < CAPACITY; i++) {
prepareChoices[currentIdx] = indexAndWeights.index;
@@ -156,4 +263,63 @@ public class LoadAwareShuffleGrouping implements LoadAwareCustomStreamGrouping,
arr[i] = arr[j];
arr[j] = tmp;
}
+
+
+ private Scope calculateScope(Map<Integer, NodeInfo> taskToNodePort, Map<String, String> hostToRack, int target) {
+ NodeInfo targetNodeInfo = taskToNodePort.get(target);
+
+ if (targetNodeInfo == null) {
+ return Scope.EVERYTHING;
+ }
+
+ String sourceRack = hostToRack.get(sourceNodeInfo.get_node());
+ String targetRack = hostToRack.get(targetNodeInfo.get_node());
+
+ if(sourceRack != null && targetRack != null && sourceRack.equals(targetRack)) {
+ if(sourceNodeInfo.get_node().equals(targetNodeInfo.get_node())) {
+ if(sourceNodeInfo.get_port().equals(targetNodeInfo.get_port())) {
+ return Scope.WORKER_LOCAL;
+ }
+ return Scope.HOST_LOCAL;
+ }
+ return Scope.RACK_LOCAL;
+ } else {
+ return Scope.EVERYTHING;
+ }
+ }
+
+ private Map<String, String> getHostToRackMapping(Map<Integer, NodeInfo> taskToNodePort) {
+ Set<String> hosts = new HashSet();
+ for (int task: targetTasks) {
+ hosts.add(taskToNodePort.get(task).get_node());
+ }
+ hosts.add(sourceNodeInfo.get_node());
+ return dnsToSwitchMapping.resolve(new ArrayList<>(hosts));
+ }
+
+ enum Scope {
+ WORKER_LOCAL, HOST_LOCAL, RACK_LOCAL, EVERYTHING;
+
+ public static Scope downgrade(Scope current) {
+ switch (current) {
+ case EVERYTHING: return RACK_LOCAL;
+ case RACK_LOCAL: return HOST_LOCAL;
+ case HOST_LOCAL:
+ case WORKER_LOCAL:
+ default:
+ return WORKER_LOCAL;
+ }
+ }
+
+ public static Scope upgrade(Scope current) {
+ switch (current) {
+ case WORKER_LOCAL: return HOST_LOCAL;
+ case HOST_LOCAL: return RACK_LOCAL;
+ case RACK_LOCAL:
+ case EVERYTHING:
+ default:
+ return EVERYTHING;
+ }
+ }
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/01bd4f82/storm-client/src/jvm/org/apache/storm/task/GeneralTopologyContext.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/task/GeneralTopologyContext.java b/storm-client/src/jvm/org/apache/storm/task/GeneralTopologyContext.java
index 6614f94..deae7cf 100644
--- a/storm-client/src/jvm/org/apache/storm/task/GeneralTopologyContext.java
+++ b/storm-client/src/jvm/org/apache/storm/task/GeneralTopologyContext.java
@@ -199,4 +199,8 @@ public class GeneralTopologyContext implements JSONAware {
}
return max;
}
+
+ public Map<String, Object> getConf() {
+ return _topoConf;
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/01bd4f82/storm-client/src/jvm/org/apache/storm/task/WorkerTopologyContext.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/task/WorkerTopologyContext.java b/storm-client/src/jvm/org/apache/storm/task/WorkerTopologyContext.java
index 2817b65..8c63eb6 100644
--- a/storm-client/src/jvm/org/apache/storm/task/WorkerTopologyContext.java
+++ b/storm-client/src/jvm/org/apache/storm/task/WorkerTopologyContext.java
@@ -17,6 +17,7 @@
*/
package org.apache.storm.task;
+import org.apache.storm.generated.NodeInfo;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.tuple.Fields;
import java.io.File;
@@ -24,6 +25,7 @@ import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicReference;
public class WorkerTopologyContext extends GeneralTopologyContext {
public static final String SHARED_EXECUTOR = "executor";
@@ -34,6 +36,8 @@ public class WorkerTopologyContext extends GeneralTopologyContext {
private String _pidDir;
Map<String, Object> _userResources;
Map<String, Object> _defaultResources;
+ private AtomicReference<Map<Integer, NodeInfo>> taskToNodePort;
+ private String assignmentId;
public WorkerTopologyContext(
StormTopology topology,
@@ -47,7 +51,9 @@ public class WorkerTopologyContext extends GeneralTopologyContext {
Integer workerPort,
List<Integer> workerTasks,
Map<String, Object> defaultResources,
- Map<String, Object> userResources
+ Map<String, Object> userResources,
+ AtomicReference<Map<Integer, NodeInfo>> taskToNodePort,
+ String assignmentId
) {
super(topology, topoConf, taskToComponent, componentToSortedTasks, componentToStreamToFields, stormId);
_codeDir = codeDir;
@@ -64,6 +70,26 @@ public class WorkerTopologyContext extends GeneralTopologyContext {
}
_workerPort = workerPort;
_workerTasks = workerTasks;
+ this.taskToNodePort = taskToNodePort;
+ this.assignmentId = assignmentId;
+
+ }
+
+ public WorkerTopologyContext(
+ StormTopology topology,
+ Map<String, Object> topoConf,
+ Map<Integer, String> taskToComponent,
+ Map<String, List<Integer>> componentToSortedTasks,
+ Map<String, Map<String, Fields>> componentToStreamToFields,
+ String stormId,
+ String codeDir,
+ String pidDir,
+ Integer workerPort,
+ List<Integer> workerTasks,
+ Map<String, Object> defaultResources,
+ Map<String, Object> userResources) {
+ this(topology, topoConf, taskToComponent, componentToSortedTasks, componentToStreamToFields, stormId,
+ codeDir, pidDir, workerPort, workerTasks, defaultResources, userResources, null, null);
}
/**
@@ -78,6 +104,18 @@ public class WorkerTopologyContext extends GeneralTopologyContext {
return _workerPort;
}
+ public String getThisWorkerHost() {
+ return assignmentId;
+ }
+
+ /**
+ * Get a map from task Id to NodePort
+ * @return a map from task To NodePort
+ */
+ public AtomicReference<Map<Integer, NodeInfo>> getTaskToNodePort() {
+ return taskToNodePort;
+ }
+
/**
* Gets the location of the external resources for this worker on the
* local filesystem. These external resources typically include bolts implemented
http://git-wip-us.apache.org/repos/asf/storm/blob/01bd4f82/storm-client/test/jvm/org/apache/storm/grouping/LoadAwareShuffleGroupingTest.java
----------------------------------------------------------------------
diff --git a/storm-client/test/jvm/org/apache/storm/grouping/LoadAwareShuffleGroupingTest.java b/storm-client/test/jvm/org/apache/storm/grouping/LoadAwareShuffleGroupingTest.java
index a5f9304..d704900 100644
--- a/storm-client/test/jvm/org/apache/storm/grouping/LoadAwareShuffleGroupingTest.java
+++ b/storm-client/test/jvm/org/apache/storm/grouping/LoadAwareShuffleGroupingTest.java
@@ -18,13 +18,19 @@
package org.apache.storm.grouping;
import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
import com.google.common.util.concurrent.MoreExecutors;
import java.util.Arrays;
+
+import org.apache.commons.collections.map.HashedMap;
+import org.apache.storm.Config;
import org.apache.storm.daemon.GrouperFactory;
import org.apache.storm.generated.GlobalStreamId;
import org.apache.storm.generated.Grouping;
+import org.apache.storm.generated.NodeInfo;
import org.apache.storm.generated.NullStruct;
import org.apache.storm.task.WorkerTopologyContext;
+import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
@@ -44,20 +50,39 @@ import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
public class LoadAwareShuffleGroupingTest {
public static final double ACCEPTABLE_MARGIN = 0.015;
private static final Logger LOG = LoggerFactory.getLogger(LoadAwareShuffleGroupingTest.class);
+ private WorkerTopologyContext mockContext(List<Integer> availableTaskIds) {
+ Map<String, Object> conf = new HashMap<>();
+ conf.put(Config.STORM_NETWORK_TOPOGRAPHY_PLUGIN, "org.apache.storm.networktopography.DefaultRackDNSToSwitchMapping");
+ conf.put(Config.TOPOLOGY_LOCALITYAWARE_HIGHER_BOUND_PERCENT, 0.8);
+ conf.put(Config.TOPOLOGY_LOCALITYAWARE_LOWER_BOUND_PERCENT, 0.2);
+
+ WorkerTopologyContext context = mock(WorkerTopologyContext.class);
+ when(context.getConf()).thenReturn(conf);
+ Map<Integer, NodeInfo> taskNodeToPort = new HashMap<>();
+ NodeInfo nodeInfo = new NodeInfo("node-id", Sets.newHashSet(6700L));
+ availableTaskIds.forEach(e -> taskNodeToPort.put(e, nodeInfo));
+ when(context.getTaskToNodePort()).thenReturn(new AtomicReference<>(taskNodeToPort));
+ when(context.getThisWorkerHost()).thenReturn("node-id");
+ when(context.getThisWorkerPort()).thenReturn(6700);
+ return context;
+ }
+
@Test
public void testUnevenLoadOverTime() throws Exception {
LoadAwareShuffleGrouping grouping = new LoadAwareShuffleGrouping();
- WorkerTopologyContext context = mock(WorkerTopologyContext.class);
+ WorkerTopologyContext context = mockContext(Arrays.asList(1, 2));
grouping.prepare(context, new GlobalStreamId("a", "default"), Arrays.asList(1, 2));
double expectedOneWeight = 100.0;
double expectedTwoWeight = 100.0;
@@ -122,7 +147,7 @@ public class LoadAwareShuffleGroupingTest {
final List<Integer> availableTaskIds = getAvailableTaskIds(numTasks);
final LoadMapping loadMapping = buildLocalTasksEvenLoadMapping(availableTaskIds);
- WorkerTopologyContext context = mock(WorkerTopologyContext.class);
+ final WorkerTopologyContext context = mockContext(availableTaskIds);
grouper.prepare(context, null, availableTaskIds);
// Keep track of how many times we see each taskId
@@ -152,7 +177,7 @@ public class LoadAwareShuffleGroupingTest {
final List<Integer> availableTaskIds = getAvailableTaskIds(numTasks);
final LoadMapping loadMapping = buildLocalTasksEvenLoadMapping(availableTaskIds);
- final WorkerTopologyContext context = mock(WorkerTopologyContext.class);
+ final WorkerTopologyContext context = mockContext(availableTaskIds);
grouper.prepare(context, null, availableTaskIds);
// force triggers building ring
@@ -222,7 +247,7 @@ public class LoadAwareShuffleGroupingTest {
public void testShuffleLoadEven() {
// port test-shuffle-load-even
LoadAwareCustomStreamGrouping shuffler = GrouperFactory
- .mkGrouper(null, "comp", "stream", null, Grouping.shuffle(new NullStruct()),
+ .mkGrouper(mockContext(Lists.newArrayList(1, 2)), "comp", "stream", null, Grouping.shuffle(new NullStruct()),
Lists.newArrayList(1, 2), Collections.emptyMap());
int numMessages = 100000;
int minPrCount = (int) (numMessages * (0.5 - ACCEPTABLE_MARGIN));
@@ -367,7 +392,7 @@ public class LoadAwareShuffleGroupingTest {
// Task Id not used, so just pick a static value
final int inputTaskId = 100;
- WorkerTopologyContext context = mock(WorkerTopologyContext.class);
+ WorkerTopologyContext context = mockContext(availableTaskIds);
grouper.prepare(context, null, availableTaskIds);
// periodically calls refreshLoad in 1 sec to simulate worker load update timer
@@ -405,7 +430,7 @@ public class LoadAwareShuffleGroupingTest {
// Task Id not used, so just pick a static value
final int inputTaskId = 100;
- final WorkerTopologyContext context = mock(WorkerTopologyContext.class);
+ final WorkerTopologyContext context = mockContext(availableTaskIds);
// Call prepare with our available taskIds
grouper.prepare(context, null, availableTaskIds);
[2/2] storm git commit: Merge branch 'STORM-2686-redesign' of
https://github.com/Ethanlm/storm into STORM-2686
Posted by bo...@apache.org.
Merge branch 'STORM-2686-redesign' of https://github.com/Ethanlm/storm into STORM-2686
STORM-2686: Add locality awareness to LoadAwareShuffleGrouping
This closes #2366
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/c8947c2f
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/c8947c2f
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/c8947c2f
Branch: refs/heads/master
Commit: c8947c2fede62036c20472c9e0335ef90a06b536
Parents: 37403d1 01bd4f8
Author: Robert Evans <ev...@yahoo-inc.com>
Authored: Mon Oct 16 10:24:39 2017 -0500
Committer: Robert Evans <ev...@yahoo-inc.com>
Committed: Mon Oct 16 10:24:39 2017 -0500
----------------------------------------------------------------------
conf/defaults.yaml | 2 +
.../src/jvm/org/apache/storm/Config.java | 18 ++
.../apache/storm/daemon/worker/WorkerState.java | 2 +-
.../grouping/LoadAwareShuffleGrouping.java | 178 ++++++++++++++++++-
.../storm/task/GeneralTopologyContext.java | 4 +
.../storm/task/WorkerTopologyContext.java | 40 ++++-
.../grouping/LoadAwareShuffleGroupingTest.java | 37 +++-
7 files changed, 267 insertions(+), 14 deletions(-)
----------------------------------------------------------------------