Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2400#discussion_r148852799
--- Diff:
storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceAwareScheduler.java
---
@@ -62,123 +70,144 @@ public void prepare(Map<String, Object> conf) {
@Override
public void schedule(Topologies topologies, Cluster cluster) {
- //initialize data structures
- for (TopologyDetails td : cluster.getTopologies()) {
+ Map<String, User> userMap = getUsers(cluster);
+ List<TopologyDetails> orderedTopologies = new
ArrayList<>(schedulingPriorityStrategy.getOrderedTopologies(cluster, userMap));
+ LOG.info("Ordered list of topologies is: {}",
orderedTopologies.stream().map((t) -> t.getId()).collect(Collectors.toList()));
+ for (TopologyDetails td : orderedTopologies) {
if (!cluster.needsSchedulingRas(td)) {
//cluster forgets about its previous status, so if it is
scheduled just leave it.
cluster.setStatusIfAbsent(td.getId(), "Fully Scheduled");
- }
- }
- Map<String, User> userMap = getUsers(cluster);
-
- while (true) {
- TopologyDetails td;
- try {
- //Call scheduling priority strategy
- td =
schedulingPrioritystrategy.getNextTopologyToSchedule(cluster, userMap);
- } catch (Exception ex) {
- LOG.error("Exception thrown when running priority strategy
{}. No topologies will be scheduled!",
- schedulingPrioritystrategy.getClass().getName(),
ex);
- break;
- }
- if (td == null) {
- break;
- }
- User submitter = userMap.get(td.getTopologySubmitter());
- if (cluster.needsSchedulingRas(td)) {
- scheduleTopology(td, cluster, submitter, userMap);
} else {
- LOG.warn("Topology {} is already fully scheduled!",
td.getName());
- cluster.setStatusIfAbsent(td.getId(), "Fully Scheduled");
+ User submitter = userMap.get(td.getTopologySubmitter());
+ scheduleTopology(td, cluster, submitter,
orderedTopologies);
}
}
}
+ private static void markFailedTopology(User u, Cluster c,
TopologyDetails td, String message) {
+ markFailedTopology(u, c, td, message, null);
+ }
- public void scheduleTopology(TopologyDetails td, Cluster cluster,
final User topologySubmitter,
- Map<String, User> userMap) {
+ private static void markFailedTopology(User u, Cluster c,
TopologyDetails td, String message, Throwable t) {
+ c.setStatus(td, message);
+ String realMessage = td.getId() + " " + message;
+ if (t != null) {
+ LOG.error(realMessage, t);
+ } else {
+ LOG.error(realMessage);
+ }
+ u.markTopoUnsuccess(td);
+ }
+
+ private void scheduleTopology(TopologyDetails td, Cluster cluster,
final User topologySubmitter,
+ List<TopologyDetails> orderedTopologies)
{
//A copy of cluster that we can modify, but does not get committed
back to cluster unless scheduling succeeds
Cluster workingState = new Cluster(cluster);
+ RAS_Nodes nodes = new RAS_Nodes(workingState);
IStrategy rasStrategy = null;
String strategyConf = (String)
td.getConf().get(Config.TOPOLOGY_SCHEDULER_STRATEGY);
try {
- rasStrategy = (IStrategy)
ReflectionUtils.newSchedulerStrategyInstance((String)
td.getConf().get(Config.TOPOLOGY_SCHEDULER_STRATEGY), conf);
+ rasStrategy =
ReflectionUtils.newSchedulerStrategyInstance((String)
td.getConf().get(Config.TOPOLOGY_SCHEDULER_STRATEGY), conf);
rasStrategy.prepare(conf);
} catch (DisallowedStrategyException e) {
- topologySubmitter.markTopoUnsuccess(td);
- cluster.setStatus(td.getId(), "Unsuccessful in scheduling - "
+ e.getAttemptedClass()
- + " is not an allowed strategy. Please make sure your
" + Config.TOPOLOGY_SCHEDULER_STRATEGY
- + " config is one of the allowed strategies: " +
e.getAllowedStrategies().toString());
+ markFailedTopology(topologySubmitter, cluster, td,
+ "Unsuccessful in scheduling - " + e.getAttemptedClass()
+ + " is not an allowed strategy. Please make sure your "
+ + Config.TOPOLOGY_SCHEDULER_STRATEGY
+ + " config is one of the allowed strategies: "
+ + e.getAllowedStrategies(), e);
return;
} catch (RuntimeException e) {
- LOG.error("failed to create instance of IStrategy: {} Topology
{} will not be scheduled.",
- strategyConf, td.getName(), e);
- topologySubmitter.markTopoUnsuccess(td);
- cluster.setStatus(td.getId(), "Unsuccessful in scheduling -
failed to create instance of topology strategy "
- + strategyConf + ". Please check logs for details");
+ markFailedTopology(topologySubmitter, cluster, td,
+ "Unsuccessful in scheduling - failed to create instance of
topology strategy "
+ + strategyConf
+ + ". Please check logs for details", e);
return;
}
- while (true) {
- // A copy of the cluster that restricts the strategy to only
modify a single topology
+ for (int i = 0; i < maxSchedulingAttempts; i++) {
SingleTopologyCluster toSchedule = new
SingleTopologyCluster(workingState, td.getId());
- SchedulingResult result = null;
try {
- result = rasStrategy.schedule(toSchedule, td);
- } catch (Exception ex) {
- LOG.error("Exception thrown when running strategy {} to
schedule topology {}."
- + " Topology will not be scheduled!",
rasStrategy.getClass().getName(), td.getName(), ex);
- topologySubmitter.markTopoUnsuccess(td);
- cluster.setStatus(td.getId(), "Unsuccessful in scheduling
- Exception thrown when running strategy {}"
- + rasStrategy.getClass().getName() + ". Please
check logs for details");
- }
- LOG.debug("scheduling result: {}", result);
- if (result != null) {
- if (result.isSuccess()) {
- try {
+ SchedulingResult result = rasStrategy.schedule(toSchedule,
td);
+ LOG.debug("scheduling result: {}", result);
+ if (result != null) {
+ if (result.isSuccess()) {
cluster.updateFrom(toSchedule);
cluster.setStatus(td.getId(), "Running - " +
result.getMessage());
- } catch (Exception ex) {
- LOG.error("Unsuccessful attempting to assign
executors to nodes.", ex);
- topologySubmitter.markTopoUnsuccess(td);
- cluster.setStatus(td.getId(), "Unsuccessful in
scheduling - "
- + "IllegalStateException thrown when
attempting to assign executors to nodes. Please check"
- + " log for details.");
- }
- return;
- } else {
- if (result.getStatus() ==
SchedulingStatus.FAIL_NOT_ENOUGH_RESOURCES) {
- boolean madeSpace = false;
+ //DONE
+ return;
+ } else if (result.getStatus() ==
SchedulingStatus.FAIL_NOT_ENOUGH_RESOURCES) {
+ LOG.info("Not enough resources to schedule {}",
td.getName());
+ List<TopologyDetails> reversedList =
ImmutableList.copyOf(orderedTopologies).reverse();
try {
- //need to re prepare since scheduling state
might have been restored
- madeSpace =
evictionStrategy.makeSpaceForTopo(td, workingState, userMap);
+ boolean evictedSomething = false;
+ LOG.debug("attempting to make space for topo
{} from user {}", td.getName(), td.getTopologySubmitter());
+ int tdIndex = reversedList.indexOf(td);
+ double cpuNeeded = td.getTotalRequestedCpu();
+ double memoryNeeded =
td.getTotalRequestedMemOffHeap() + td.getTotalRequestedMemOnHeap();
+ SchedulerAssignment assignment =
cluster.getAssignmentById(td.getId());
+ if (assignment != null) {
+ cpuNeeded -= getCpuUsed(assignment);
+ memoryNeeded -= getMemoryUsed(assignment);
+ }
+ cluster.getTopologyResourcesMap();
+ for (int index = 0; index < tdIndex; index++) {
+ TopologyDetails topologyEvict =
reversedList.get(index);
+ SchedulerAssignment evictAssignemnt =
workingState.getAssignmentById(topologyEvict.getId());
+ if (evictAssignemnt != null &&
!evictAssignemnt.getSlots().isEmpty()) {
+ Collection<WorkerSlot> workersToEvict
= workingState.getUsedSlotsByTopologyId(topologyEvict.getId());
+
+ LOG.debug("Evicting Topology {} with
workers: {} from user {}", topologyEvict.getName(), workersToEvict,
+
topologyEvict.getTopologySubmitter());
+ cpuNeeded -=
getCpuUsed(evictAssignemnt);
+ memoryNeeded -=
getMemoryUsed(evictAssignemnt);
+ evictedSomething = true;
+ nodes.freeSlots(workersToEvict);
+ if (cpuNeeded <= 0 && memoryNeeded <=
0) {
+ //We evicted enough topologies to
have a hope of scheduling, so try it now, and don't evict more
+ // than is needed
+ break;
+ }
+ }
+ }
+
+ if (!evictedSomething) {
+ markFailedTopology(topologySubmitter,
cluster, td,
+ "Not enough resources to schedule - "
+ result.getErrorMessage());
+ return;
+ }
} catch (Exception ex) {
- LOG.error("Exception thrown when running
eviction strategy {} to schedule topology {}."
- + " No evictions will be
done!", evictionStrategy.getClass().getName(),
- td.getName(), ex);
- topologySubmitter.markTopoUnsuccess(td);
- return;
- }
- if (!madeSpace) {
- LOG.debug("Could not make space for topo {}
will move to attempted", td);
- topologySubmitter.markTopoUnsuccess(td);
- cluster.setStatus(td.getId(), "Not enough
resources to schedule - "
- + result.getErrorMessage());
- return;
+ LOG.error("Exception thrown when running
eviction to schedule topology {}."
+ + " No evictions will be done! Error:
{}",
+ td.getName(), ex.getClass().getName(), ex);
}
+ //Only place we fall though to do the loop over
again...
continue;
} else {
--- End diff --
Should comment that this is for all other failed status
---
