Github user d2r commented on a diff in the pull request:
https://github.com/apache/storm/pull/1016#discussion_r49786540
--- Diff:
storm-core/src/jvm/org/apache/storm/scheduler/resource/RAS_Node.java ---
@@ -43,32 +43,84 @@
*/
public class RAS_Node {
private static final Logger LOG =
LoggerFactory.getLogger(RAS_Node.class);
- private Map<String, Set<WorkerSlot>> _topIdToUsedSlots = new
HashMap<String, Set<WorkerSlot>>();
- private Set<WorkerSlot> _freeSlots = new HashSet<WorkerSlot>();
+
+ //A map consisting of all workers on the node.
+ //The key of the map is the worker id and the value is the
corresponding workerslot object
+ Map<String, WorkerSlot> _slots = new HashMap<String, WorkerSlot> ();
+
+ // A map describing which topologies are using which slots on this
node. The format of the map is the following:
+ // {TopologyId -> {WorkerId -> {Executors}}}
+ private Map<String, Map<String, Collection<ExecutorDetails>>>
_topIdToUsedSlots = new HashMap<String, Map<String,
Collection<ExecutorDetails>>>();
+
private final String _nodeId;
private String _hostname;
private boolean _isAlive;
private SupervisorDetails _sup;
- private Double _availMemory;
- private Double _availCPU;
- private Cluster _cluster;
- private Topologies _topologies;
+ private Double _availMemory = 0.0;
+ private Double _availCPU = 0.0;
+ private final Cluster _cluster;
+ private final Topologies _topologies;
- public RAS_Node(String nodeId, Set<Integer> allPorts, boolean isAlive,
- SupervisorDetails sup, Cluster cluster, Topologies
topologies) {
+ public RAS_Node(String nodeId, SupervisorDetails sup, Cluster cluster,
Topologies topologies, Map<String, WorkerSlot> workerIdToWorker, Map<String,
Map<String, Collection<ExecutorDetails>>> assignmentMap) {
+ //Node ID and supervisor ID are the same.
_nodeId = nodeId;
- _isAlive = isAlive;
- if (_isAlive && allPorts != null) {
- for (int port : allPorts) {
- _freeSlots.add(new WorkerSlot(_nodeId, port));
- }
- _sup = sup;
+ if (sup == null) {
+ _isAlive = false;
+ } else {
+ _isAlive = !cluster.isBlackListed(_nodeId);
+ }
+
+ _cluster = cluster;
+ _topologies = topologies;
+
+ // initialize slots for this node
+ if (workerIdToWorker != null) {
+ _slots = workerIdToWorker;
+ }
+
+ //initialize assignment map
+ if (assignmentMap != null) {
+ _topIdToUsedSlots = assignmentMap;
+ }
+
+ //check if node is alive
+ if (_isAlive && sup != null) {
_hostname = sup.getHost();
+ _sup = sup;
_availMemory = getTotalMemoryResources();
_availCPU = getTotalCpuResources();
+
+ LOG.debug("Found a {} Node {} {}",
+ _isAlive ? "living" : "dead", _nodeId,
sup.getAllPorts());
+ LOG.debug("resources_mem: {}, resources_CPU: {}",
sup.getTotalMemory(), sup.getTotalCPU());
+ //intialize resource usages on node
+ intializeResources();
+ }
+ }
+
+ /**
+ * intializes resource usages on node
--- End diff --
`initializes`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
