Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2623#discussion_r179570190
--- Diff:
storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/BaseResourceAwareStrategy.java
---
@@ -477,45 +414,136 @@ protected String nodeToRack(RAS_Node node) {
List<ExecutorDetails> execsScheduled = new LinkedList<>();
Map<String, Queue<ExecutorDetails>> compToExecsToSchedule = new
HashMap<>();
- for (Component component : componentMap.values()) {
- compToExecsToSchedule.put(component.getId(), new
LinkedList<ExecutorDetails>());
+ for (Map.Entry<String, Component> componentEntry:
componentMap.entrySet()) {
+ Component component = componentEntry.getValue();
+ compToExecsToSchedule.put(component.getId(), new
LinkedList<>());
for (ExecutorDetails exec : component.getExecs()) {
if (unassignedExecutors.contains(exec)) {
compToExecsToSchedule.get(component.getId()).add(exec);
+ LOG.info("{} has unscheduled executor {}",
component.getId(), exec);
}
}
}
- Set<Component> sortedComponents = sortComponents(componentMap);
- sortedComponents.addAll(componentMap.values());
+ List<Component> sortedComponents =
topologicalSortComponents(componentMap);
- for (Component currComp : sortedComponents) {
- Map<String, Component> neighbors = new HashMap<String,
Component>();
- for (String compId : Sets.union(currComp.getChildren(),
currComp.getParents())) {
- neighbors.put(compId, componentMap.get(compId));
+ for (Component currComp: sortedComponents) {
+ int numExecs =
compToExecsToSchedule.get(currComp.getId()).size();
+ for (int i = 0; i < numExecs; i++) {
+ execsScheduled.addAll(takeExecutors(currComp, numExecs -
i, componentMap, compToExecsToSchedule));
}
- Set<Component> sortedNeighbors = sortNeighbors(currComp,
neighbors);
- Queue<ExecutorDetails> currCompExesToSched =
compToExecsToSchedule.get(currComp.getId());
-
- boolean flag = false;
- do {
- flag = false;
- if (!currCompExesToSched.isEmpty()) {
- execsScheduled.add(currCompExesToSched.poll());
- flag = true;
- }
+ }
+
+ LOG.info("The ordering result is {}", execsScheduled);
+
+ return execsScheduled;
+ }
- for (Component neighborComp : sortedNeighbors) {
- Queue<ExecutorDetails> neighborCompExesToSched =
- compToExecsToSchedule.get(neighborComp.getId());
- if (!neighborCompExesToSched.isEmpty()) {
- execsScheduled.add(neighborCompExesToSched.poll());
- flag = true;
+ private List<ExecutorDetails> takeExecutors(Component currComp, int
numExecs,
+ final Map<String,
Component> componentMap,
+ final Map<String,
Queue<ExecutorDetails>> compToExecsToSchedule) {
+ List<ExecutorDetails> execsScheduled = new ArrayList<>();
+ Queue<ExecutorDetails> currQueue =
compToExecsToSchedule.get((currComp.getId()));
--- End diff --
nit: there is an extra unneeded pair of '(' and ')'
---
