You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2017/07/31 02:07:10 UTC
[11/20] storm git commit: STORM-2497: Let Supervisor enforce memory
and add in support for shared memory regions
http://git-wip-us.apache.org/repos/asf/storm/blob/b4d33955/storm-client/src/jvm/org/apache/storm/scheduler/TopologyDetails.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/TopologyDetails.java b/storm-client/src/jvm/org/apache/storm/scheduler/TopologyDetails.java
deleted file mode 100644
index 520aa4f..0000000
--- a/storm-client/src/jvm/org/apache/storm/scheduler/TopologyDetails.java
+++ /dev/null
@@ -1,514 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.storm.scheduler;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.storm.Config;
-import org.apache.storm.generated.Bolt;
-import org.apache.storm.generated.GlobalStreamId;
-import org.apache.storm.generated.Grouping;
-import org.apache.storm.generated.SpoutSpec;
-import org.apache.storm.generated.StormTopology;
-import org.apache.storm.scheduler.resource.Component;
-import org.apache.storm.scheduler.resource.ResourceUtils;
-import org.apache.storm.utils.Utils;
-import org.apache.storm.utils.ObjectReader;
-import org.apache.storm.utils.Time;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class TopologyDetails {
- private final String topologyId;
- private final Map<String, Object> topologyConf;
- private final StormTopology topology;
- private final Map<ExecutorDetails, String> executorToComponent;
- private final int numWorkers;
- //<ExecutorDetails - Task, Map<String - Type of resource, Map<String - type of that resource, Double - amount>>>
- private Map<ExecutorDetails, Map<String, Double>> resourceList;
- //Max heap size for a worker used by topology
- private Double topologyWorkerMaxHeapSize;
- //topology priority
- private Integer topologyPriority;
- //when topology was launched
- private final int launchTime;
- private final String owner;
-
- private static final Logger LOG = LoggerFactory.getLogger(TopologyDetails.class);
-
- public TopologyDetails(String topologyId, Map<String, Object> topologyConf, StormTopology topology, int numWorkers, String owner) {
- this(topologyId, topologyConf, topology, numWorkers, null, 0, owner);
- }
-
- public TopologyDetails(String topologyId, Map<String, Object> topologyConf, StormTopology topology,
- int numWorkers, Map<ExecutorDetails, String> executorToComponents, String owner) {
- this(topologyId, topologyConf, topology, numWorkers, executorToComponents, 0, owner);
- }
-
- public TopologyDetails(String topologyId, Map<String, Object> topologyConf, StormTopology topology, int numWorkers,
- Map<ExecutorDetails, String> executorToComponents, int launchTime, String owner) {
- this.owner = owner;
- this.topologyId = topologyId;
- this.topologyConf = topologyConf;
- this.topology = topology;
- this.numWorkers = numWorkers;
- this.executorToComponent = new HashMap<>(0);
- if (executorToComponents != null) {
- this.executorToComponent.putAll(executorToComponents);
- }
- if (this.topology != null) {
- this.initResourceList();
- }
- this.initConfigs();
- this.launchTime = launchTime;
- }
-
- public String getId() {
- return this.topologyId;
- }
-
- public String getName() {
- return (String) this.topologyConf.get(Config.TOPOLOGY_NAME);
- }
-
- public Map<String, Object> getConf() {
- return this.topologyConf;
- }
-
- public int getNumWorkers() {
- return this.numWorkers;
- }
-
- public StormTopology getTopology() {
- return topology;
- }
-
- public Map<ExecutorDetails, String> getExecutorToComponent() {
- return this.executorToComponent;
- }
-
- public Map<ExecutorDetails, String> selectExecutorToComponent(Collection<ExecutorDetails> executors) {
- Map<ExecutorDetails, String> ret = new HashMap<>(executors.size());
- for (ExecutorDetails executor : executors) {
- String compId = this.executorToComponent.get(executor);
- if (compId != null) {
- ret.put(executor, compId);
- }
- }
-
- return ret;
- }
-
- public Set<ExecutorDetails> getExecutors() {
- return this.executorToComponent.keySet();
- }
-
- private void initResourceList() {
- this.resourceList = new HashMap<>();
- // Extract bolt memory info
- if (this.topology.get_bolts() != null) {
- for (Map.Entry<String, Bolt> bolt : this.topology.get_bolts().entrySet()) {
- //the json_conf is populated by TopologyBuilder (e.g. boltDeclarer.setMemoryLoad)
- Map<String, Double> topology_resources = ResourceUtils.parseResources(bolt
- .getValue().get_common().get_json_conf());
- ResourceUtils.checkIntialization(topology_resources, bolt.getValue().toString(), this.topologyConf);
- for (Map.Entry<ExecutorDetails, String> anExecutorToComponent : this.executorToComponent.entrySet()) {
- if (bolt.getKey().equals(anExecutorToComponent.getValue())) {
- this.resourceList.put(anExecutorToComponent.getKey(), topology_resources);
- }
- }
- }
- }
- // Extract spout memory info
- if (this.topology.get_spouts() != null) {
- for (Map.Entry<String, SpoutSpec> spout : this.topology.get_spouts().entrySet()) {
- Map<String, Double> topology_resources = ResourceUtils.parseResources(spout
- .getValue().get_common().get_json_conf());
- ResourceUtils.checkIntialization(topology_resources, spout.getValue().toString(), this.topologyConf);
- for (Map.Entry<ExecutorDetails, String> anExecutorToComponent : this.executorToComponent.entrySet()) {
- if (spout.getKey().equals(anExecutorToComponent.getValue())) {
- this.resourceList.put(anExecutorToComponent.getKey(), topology_resources);
- }
- }
- }
- } else {
- LOG.warn("Topology " + this.topologyId + " does not seem to have any spouts!");
- }
- //schedule tasks that are not part of components returned from topology.get_spout or topology.getbolt (AKA sys tasks most specifically __acker tasks)
- for(ExecutorDetails exec : this.getExecutors()) {
- if (!this.resourceList.containsKey(exec)) {
- LOG.debug(
- "Scheduling {} {} with memory requirement as 'on heap' - {} and 'off heap' - {} and CPU requirement as {}",
- this.getExecutorToComponent().get(exec),
- exec,
- this.topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB),
- this.topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB),
- this.topologyConf.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT));
- this.addDefaultResforExec(exec);
- }
- }
- }
-
- private List<ExecutorDetails> componentToExecs(String comp) {
- List<ExecutorDetails> execs = new ArrayList<>();
- for (Map.Entry<ExecutorDetails, String> entry : this.executorToComponent.entrySet()) {
- if (entry.getValue().equals(comp)) {
- execs.add(entry.getKey());
- }
- }
- return execs;
- }
-
- /**
- * Returns a representation of the non-system components of the topology graph
- * Each Component object in the returning map is populated with the list of its
- * parents, children and execs assigned to that component.
- * @return a map of components
- */
- public Map<String, Component> getComponents() {
- Map<String, Component> all_comp = new HashMap<>();
-
- StormTopology storm_topo = this.topology;
- // spouts
- if (storm_topo.get_spouts() != null) {
- for (Map.Entry<String, SpoutSpec> spoutEntry : storm_topo
- .get_spouts().entrySet()) {
- if (!Utils.isSystemId(spoutEntry.getKey())) {
- Component newComp;
- if (all_comp.containsKey(spoutEntry.getKey())) {
- newComp = all_comp.get(spoutEntry.getKey());
- newComp.execs = componentToExecs(newComp.id);
- } else {
- newComp = new Component(spoutEntry.getKey());
- newComp.execs = componentToExecs(newComp.id);
- all_comp.put(spoutEntry.getKey(), newComp);
- }
- newComp.type = Component.ComponentType.SPOUT;
-
- for (Map.Entry<GlobalStreamId, Grouping> spoutInput : spoutEntry
- .getValue().get_common().get_inputs()
- .entrySet()) {
- newComp.parents.add(spoutInput.getKey()
- .get_componentId());
- if (!all_comp.containsKey(spoutInput
- .getKey().get_componentId())) {
- all_comp.put(spoutInput.getKey()
- .get_componentId(),
- new Component(spoutInput.getKey()
- .get_componentId()));
- }
- all_comp.get(spoutInput.getKey()
- .get_componentId()).children.add(spoutEntry
- .getKey());
- }
- }
- }
- }
- // bolts
- if (storm_topo.get_bolts() != null) {
- for (Map.Entry<String, Bolt> boltEntry : storm_topo.get_bolts()
- .entrySet()) {
- if (!Utils.isSystemId(boltEntry.getKey())) {
- Component newComp;
- if (all_comp.containsKey(boltEntry.getKey())) {
- newComp = all_comp.get(boltEntry.getKey());
- newComp.execs = componentToExecs(newComp.id);
- } else {
- newComp = new Component(boltEntry.getKey());
- newComp.execs = componentToExecs(newComp.id);
- all_comp.put(boltEntry.getKey(), newComp);
- }
- newComp.type = Component.ComponentType.BOLT;
-
- for (Map.Entry<GlobalStreamId, Grouping> boltInput : boltEntry
- .getValue().get_common().get_inputs()
- .entrySet()) {
- newComp.parents.add(boltInput.getKey()
- .get_componentId());
- if (!all_comp.containsKey(boltInput
- .getKey().get_componentId())) {
- all_comp.put(boltInput.getKey()
- .get_componentId(),
- new Component(boltInput.getKey()
- .get_componentId()));
- }
- all_comp.get(boltInput.getKey()
- .get_componentId()).children.add(boltEntry
- .getKey());
- }
- }
- }
- }
- return all_comp;
- }
-
- /**
- * Gets the on heap memory requirement for a
- * certain task within a topology
- * @param exec the executor the inquiry is concerning.
- * @return Double the amount of on heap memory
- * requirement for this exec in topology topoId.
- */
- public Double getOnHeapMemoryRequirement(ExecutorDetails exec) {
- Double ret = null;
- if (hasExecInTopo(exec)) {
- ret = this.resourceList
- .get(exec)
- .get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB);
- }
- return ret;
- }
-
- /**
- * Gets the off heap memory requirement for a
- * certain task within a topology
- * @param exec the executor the inquiry is concerning.
- * @return Double the amount of off heap memory
- * requirement for this exec in topology topoId.
- */
- public Double getOffHeapMemoryRequirement(ExecutorDetails exec) {
- Double ret = null;
- if (hasExecInTopo(exec)) {
- ret = this.resourceList
- .get(exec)
- .get(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB);
- }
- return ret;
- }
-
- /**
- * Gets the total memory requirement for a task
- * @param exec the executor the inquiry is concerning.
- * @return Double the total memory requirement
- * for this exec in topology topoId.
- */
- public Double getTotalMemReqTask(ExecutorDetails exec) {
- if (hasExecInTopo(exec)) {
- return getOffHeapMemoryRequirement(exec)
- + getOnHeapMemoryRequirement(exec);
- }
- return null;
- }
-
- /**
- * Gets the total memory resource list for a
- * set of tasks that is part of a topology.
- * @return Map<ExecutorDetails, Double> a map of the total memory requirement
- * for all tasks in topology topoId.
- */
- public Map<ExecutorDetails, Double> getTotalMemoryResourceList() {
- Map<ExecutorDetails, Double> ret = new HashMap<>();
- for (ExecutorDetails exec : this.resourceList.keySet()) {
- ret.put(exec, getTotalMemReqTask(exec));
- }
- return ret;
- }
-
- /**
- * Get the total CPU requirement for executor
- * @return Double the total about of cpu requirement for executor
- */
- public Double getTotalCpuReqTask(ExecutorDetails exec) {
- if (hasExecInTopo(exec)) {
- return this.resourceList
- .get(exec)
- .get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT);
- }
- return null;
- }
-
- /**
- * Note: The public API relevant to resource aware scheduling is unstable as of May 2015.
- * We reserve the right to change them.
- *
- * @return the total on-heap memory requested for this topology
- */
- public Double getTotalRequestedMemOnHeap() {
- Double total_memonheap = 0.0;
- for (ExecutorDetails exec : this.getExecutors()) {
- Double exec_mem = getOnHeapMemoryRequirement(exec);
- if (exec_mem != null) {
- total_memonheap += exec_mem;
- }
- }
- return total_memonheap;
- }
-
- /**
- * Note: The public API relevant to resource aware scheduling is unstable as of May 2015.
- * We reserve the right to change them.
- *
- * @return the total off-heap memory requested for this topology
- */
- public Double getTotalRequestedMemOffHeap() {
- Double total_memoffheap = 0.0;
- for (ExecutorDetails exec : this.getExecutors()) {
- Double exec_mem = getOffHeapMemoryRequirement(exec);
- if (exec_mem != null) {
- total_memoffheap += exec_mem;
- }
- }
- return total_memoffheap;
- }
-
- /**
- * Note: The public API relevant to resource aware scheduling is unstable as of May 2015.
- * We reserve the right to change them.
- *
- * @return the total cpu requested for this topology
- */
- public Double getTotalRequestedCpu() {
- Double total_cpu = 0.0;
- for (ExecutorDetails exec : this.getExecutors()) {
- Double exec_cpu = getTotalCpuReqTask(exec);
- if (exec_cpu != null) {
- total_cpu += exec_cpu;
- }
- }
- return total_cpu;
- }
-
- /**
- * get the resources requirements for a executor
- * @param exec
- * @return a map containing the resource requirements for this exec
- */
- public Map<String, Double> getTaskResourceReqList(ExecutorDetails exec) {
- if (hasExecInTopo(exec)) {
- return this.resourceList.get(exec);
- }
- return null;
- }
-
- /**
- * Checks if a executor is part of this topology
- * @return Boolean whether or not a certain ExecutorDetail is included in the resourceList.
- */
- public boolean hasExecInTopo(ExecutorDetails exec) {
- return this.resourceList != null && this.resourceList.containsKey(exec);
- }
-
- /**
- * add resource requirements for a executor
- */
- public void addResourcesForExec(ExecutorDetails exec, Map<String, Double> resourceList) {
- if (hasExecInTopo(exec)) {
- LOG.warn("Executor {} already exists...ResourceList: {}", exec, getTaskResourceReqList(exec));
- return;
- }
- this.resourceList.put(exec, resourceList);
- }
-
- /**
- * Add default resource requirements for a executor
- */
- private void addDefaultResforExec(ExecutorDetails exec) {
- Double topologyComponentCpuPcorePercent = ObjectReader.getDouble(this.topologyConf.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT), null);
- Double topologyComponentResourcesOffheapMemoryMb = ObjectReader.getDouble(this.topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB), null);
- Double topologyComponentResourcesOnheapMemoryMb = ObjectReader.getDouble(this.topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB), null);
-
- assert topologyComponentCpuPcorePercent != null;
- assert topologyComponentResourcesOffheapMemoryMb != null;
- assert topologyComponentResourcesOnheapMemoryMb != null;
-
- Map<String, Double> defaultResourceList = new HashMap<>();
- defaultResourceList.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, topologyComponentCpuPcorePercent);
- defaultResourceList.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, topologyComponentResourcesOffheapMemoryMb);
- defaultResourceList.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, topologyComponentResourcesOnheapMemoryMb);
- LOG.debug("Scheduling Executor: {} with memory requirement as onHeap: {} - offHeap: {} " +
- "and CPU requirement: {}",
- exec, this.topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB),
- this.topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB),
- this.topologyConf.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT));
- addResourcesForExec(exec, defaultResourceList);
- }
-
- /**
- * initializes member variables
- */
- private void initConfigs() {
- this.topologyWorkerMaxHeapSize = ObjectReader.getDouble(this.topologyConf.get(Config.TOPOLOGY_WORKER_MAX_HEAP_SIZE_MB), null);
- this.topologyPriority = ObjectReader.getInt(this.topologyConf.get(Config.TOPOLOGY_PRIORITY), null);
-
- assert this.topologyWorkerMaxHeapSize != null;
- assert this.topologyPriority != null;
- }
-
- /**
- * Get the max heap size for a worker used by this topology
- * @return the worker max heap size
- */
- public Double getTopologyWorkerMaxHeapSize() {
- return this.topologyWorkerMaxHeapSize;
- }
-
- /**
- * Get the user that submitted this topology
- */
- public String getTopologySubmitter() {
- return owner;
- }
-
- /**
- * get teh priority of this topology
- */
- public int getTopologyPriority() {
- return this.topologyPriority;
- }
-
- /**
- * Get the timestamp of when this topology was launched
- */
- public int getLaunchTime() {
- return this.launchTime;
- }
-
- /**
- * Get how long this topology has been executing
- */
- public int getUpTime() {
- return Time.currentTimeSecs() - this.launchTime;
- }
-
- @Override
- public String toString() {
- return "Name: " + this.getName() + " id: " + this.getId() + " Priority: " + this.getTopologyPriority()
- + " Uptime: " + this.getUpTime() + " CPU: " + this.getTotalRequestedCpu()
- + " Memory: " + (this.getTotalRequestedMemOffHeap() + this.getTotalRequestedMemOnHeap());
- }
-
- @Override
- public int hashCode() {
- return this.topologyId.hashCode();
- }
-
- @Override
- public boolean equals(Object o) {
- if (!(o instanceof TopologyDetails)) {
- return false;
- }
- return (this.topologyId.equals(((TopologyDetails) o).getId()));
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/b4d33955/storm-client/src/jvm/org/apache/storm/scheduler/WorkerSlot.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/WorkerSlot.java b/storm-client/src/jvm/org/apache/storm/scheduler/WorkerSlot.java
index 308af85..bf8c75d 100644
--- a/storm-client/src/jvm/org/apache/storm/scheduler/WorkerSlot.java
+++ b/storm-client/src/jvm/org/apache/storm/scheduler/WorkerSlot.java
@@ -15,28 +15,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.scheduler;
public class WorkerSlot {
- private String nodeId;
- private int port;
- // amount of on-heap memory allocated to it
- private double memOnHeap = 0.0;
- // amount of off-heap memory allocated to it
- private double memOffHeap = 0.0;
- // amount of cpu allocated to it
- private double cpu = 0.0;
-
- public WorkerSlot(String nodeId, Number port) {
- this(nodeId, port, 0.0, 0.0, 0.0);
- }
+ private final String nodeId;
+ private final int port;
- public WorkerSlot(String nodeId, Number port, double memOnHeap, double memOffHeap, double cpu) {
+ public WorkerSlot(String nodeId, Number port) {
+ if (port == null) {
+ throw new NullPointerException("port cannot be null");
+ }
+ if (nodeId == null) {
+ throw new NullPointerException("node id cannot be null");
+ }
this.nodeId = nodeId;
this.port = port.intValue();
- this.memOnHeap = memOnHeap;
- this.memOffHeap = memOffHeap;
- this.cpu = cpu;
}
public String getNodeId() {
@@ -51,18 +45,6 @@ public class WorkerSlot {
return getNodeId() + ":" + getPort();
}
- public double getAllocatedMemOnHeap() {
- return memOnHeap;
- }
-
- public double getAllocatedMemOffHeap() {
- return memOffHeap;
- }
-
- public double getAllocatedCpu() {
- return cpu;
- }
-
@Override
public int hashCode() {
return nodeId.hashCode() + 13 * ((Integer) port).hashCode();
@@ -70,12 +52,15 @@ public class WorkerSlot {
@Override
public boolean equals(Object o) {
+ if (!(o instanceof WorkerSlot)){
+ return false;
+ }
WorkerSlot other = (WorkerSlot) o;
return this.port == other.port && this.nodeId.equals(other.nodeId);
}
@Override
public String toString() {
- return this.nodeId + ":" + this.port;
+ return getId();
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/b4d33955/storm-client/src/jvm/org/apache/storm/scheduler/resource/Component.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/resource/Component.java b/storm-client/src/jvm/org/apache/storm/scheduler/resource/Component.java
deleted file mode 100644
index f52ee74..0000000
--- a/storm-client/src/jvm/org/apache/storm/scheduler/resource/Component.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.storm.scheduler.resource;
-
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.storm.scheduler.ExecutorDetails;
-
-public class Component {
- public enum ComponentType {
- SPOUT(1), BOLT(2);
- private int value;
-
- private ComponentType(int value) {
- this.value = value;
- }
- }
-
- public String id;
- public List<String> parents = null;
- public List<String> children = null;
- public List<ExecutorDetails> execs = null;
- public ComponentType type = null;
-
- public Component(String id) {
- this.parents = new ArrayList<String>();
- this.children = new ArrayList<String>();
- this.execs = new ArrayList<ExecutorDetails>();
- this.id = id;
- }
-
- @Override
- public String toString() {
- String retVal = "{id: " + this.id + " Parents: " + this.parents.toString() + " Children: " + this.children.toString() + " Execs: " + this.execs + "}";
- return retVal;
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/b4d33955/storm-client/src/jvm/org/apache/storm/scheduler/resource/RAS_Node.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/resource/RAS_Node.java b/storm-client/src/jvm/org/apache/storm/scheduler/resource/RAS_Node.java
deleted file mode 100644
index be3dbc1..0000000
--- a/storm-client/src/jvm/org/apache/storm/scheduler/resource/RAS_Node.java
+++ /dev/null
@@ -1,529 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.storm.scheduler.resource;
-
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.Map;
-import java.util.Set;
-import java.util.Map.Entry;
-import java.util.ArrayList;
-
-import org.apache.storm.scheduler.Topologies;
-import org.apache.storm.scheduler.TopologyDetails;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.storm.scheduler.Cluster;
-import org.apache.storm.scheduler.ExecutorDetails;
-import org.apache.storm.scheduler.SchedulerAssignment;
-import org.apache.storm.scheduler.SupervisorDetails;
-import org.apache.storm.scheduler.WorkerSlot;
-
-/**
- * Represents a single node in the cluster.
- */
-public class RAS_Node {
- private static final Logger LOG = LoggerFactory.getLogger(RAS_Node.class);
-
- //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 = 0.0;
- private Double _availCPU = 0.0;
- private final Cluster _cluster;
- private final 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;
- 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();
- //intialize resource usages on node
- intializeResources();
- }
- }
-
- /**
- * initializes resource usages on node
- */
- private void intializeResources() {
- for (Entry<String, Map<String, Collection<ExecutorDetails>>> entry : _topIdToUsedSlots.entrySet()) {
- String topoId = entry.getKey();
- Map<String, Collection<ExecutorDetails>> assignment = entry.getValue();
- Map<ExecutorDetails, Double> topoMemoryResourceList = _topologies.getById(topoId).getTotalMemoryResourceList();
- for (Collection<ExecutorDetails> execs : assignment.values()) {
- for (ExecutorDetails exec : execs) {
- if (!_isAlive) {
- continue;
- // We do not free the assigned slots (the orphaned slots) on the inactive supervisors
- // The inactive node will be treated as a 0-resource node and not available for other unassigned workers
- }
- if (topoMemoryResourceList.containsKey(exec)) {
- consumeResourcesforTask(exec, _topologies.getById(topoId));
- } else {
- throw new IllegalStateException("Executor " + exec + "not found!");
- }
- }
- }
- }
- }
-
- public String getId() {
- return _nodeId;
- }
-
- public String getHostname() {
- return _hostname;
- }
-
- private Collection<WorkerSlot> workerIdsToWorkers(Collection<String> workerIds) {
- Collection<WorkerSlot> ret = new LinkedList<WorkerSlot>();
- for (String workerId : workerIds) {
- ret.add(_slots.get(workerId));
- }
- return ret;
- }
-
- public Collection<String> getFreeSlotsId() {
- if (!_isAlive) {
- return new HashSet<String>();
- }
- Collection<String> usedSlotsId = getUsedSlotsId();
- Set<String> ret = new HashSet<>();
- ret.addAll(_slots.keySet());
- ret.removeAll(usedSlotsId);
- return ret;
- }
-
- public Collection<WorkerSlot> getFreeSlots() {
- return workerIdsToWorkers(getFreeSlotsId());
- }
-
- public Collection<String> getUsedSlotsId() {
- Collection<String> ret = new LinkedList<String>();
- for (Map<String, Collection<ExecutorDetails>> entry : _topIdToUsedSlots.values()) {
- ret.addAll(entry.keySet());
- }
- return ret;
- }
-
- public Collection<WorkerSlot> getUsedSlots() {
- return workerIdsToWorkers(getUsedSlotsId());
- }
-
- public Collection<WorkerSlot> getUsedSlots(String topId) {
- return workerIdsToWorkers(_topIdToUsedSlots.get(topId).keySet());
- }
-
- public boolean isAlive() {
- return _isAlive;
- }
-
- /**
- * @return a collection of the topology ids currently running on this node
- */
- public Collection<String> getRunningTopologies() {
- return _topIdToUsedSlots.keySet();
- }
-
- public boolean isTotallyFree() {
- return getUsedSlots().isEmpty();
- }
-
- public int totalSlotsFree() {
- return getFreeSlots().size();
- }
-
- public int totalSlotsUsed() {
- return getUsedSlots().size();
- }
-
- public int totalSlots() {
- return _slots.size();
- }
-
- public int totalSlotsUsed(String topId) {
- return getUsedSlots(topId).size();
- }
-
- /**
- * Free all slots on this node. This will update the Cluster too.
- */
- public void freeAllSlots() {
- if (!_isAlive) {
- LOG.warn("Freeing all slots on a dead node {} ", _nodeId);
- }
- _cluster.freeSlots(_slots.values());
- _availCPU = getTotalCpuResources();
- _availMemory = getAvailableMemoryResources();
- //clearing assignments
- _topIdToUsedSlots.clear();
- }
-
- /**
- * Frees a single slot in this node
- * @param ws the slot to free
- */
- public void free(WorkerSlot ws) {
- LOG.info("freeing WorkerSlot {} on node {}", ws, _hostname);
- if (!_slots.containsKey(ws.getId())) {
- throw new IllegalArgumentException("Tried to free a slot " + ws + " that was not" +
- " part of this node " + _nodeId);
- }
-
- TopologyDetails topo = findTopologyUsingWorker(ws);
- if (topo == null) {
- throw new IllegalArgumentException("Tried to free a slot " + ws + " that was already free!");
- }
-
- double memUsed = getMemoryUsedByWorker(ws);
- double cpuUsed = getCpuUsedByWorker(ws);
- freeMemory(memUsed);
- freeCPU(cpuUsed);
-
- //free slot
- _cluster.freeSlot(ws);
- //cleanup internal assignments
- _topIdToUsedSlots.get(topo.getId()).remove(ws.getId());
- }
-
- private void freeMemory(double amount) {
- LOG.debug("freeing {} memory on node {}...avail mem: {}", amount, getHostname(), _availMemory);
- if((_availMemory + amount) > getTotalMemoryResources()) {
- LOG.warn("Freeing more memory than there exists! Memory trying to free: {} Total memory on Node: {}", (_availMemory + amount), getTotalMemoryResources());
- return;
- }
- _availMemory += amount;
- }
-
- private void freeCPU(double amount) {
- LOG.debug("freeing {} CPU on node...avail CPU: {}", amount, getHostname(), _availCPU);
- if ((_availCPU + amount) > getTotalCpuResources()) {
- LOG.warn("Freeing more CPU than there exists! CPU trying to free: {} Total CPU on Node: {}", (_availCPU + amount), getTotalCpuResources());
- return;
- }
- _availCPU += amount;
- }
-
- /**
- * get the amount of memory used by a worker
- */
- public double getMemoryUsedByWorker(WorkerSlot ws) {
- TopologyDetails topo = findTopologyUsingWorker(ws);
- if (topo == null) {
- return 0.0;
- }
- Collection<ExecutorDetails> execs = getExecutors(ws, _cluster);
- double totalMemoryUsed = 0.0;
- for (ExecutorDetails exec : execs) {
- totalMemoryUsed += topo.getTotalMemReqTask(exec);
- }
- return totalMemoryUsed;
- }
-
- /**
- * get the amount of cpu used by a worker
- */
- public double getCpuUsedByWorker(WorkerSlot ws) {
- TopologyDetails topo = findTopologyUsingWorker(ws);
- if (topo == null) {
- return 0.0;
- }
- Collection<ExecutorDetails> execs = getExecutors(ws, _cluster);
- double totalCpuUsed = 0.0;
- for (ExecutorDetails exec : execs) {
- totalCpuUsed += topo.getTotalCpuReqTask(exec);
- }
- return totalCpuUsed;
- }
-
- /**
- * Find a which topology is running on a worker slot
- * @return the topology using the worker slot. If worker slot is free then return null
- */
- public TopologyDetails findTopologyUsingWorker(WorkerSlot ws) {
- for (Entry<String, Map<String, Collection<ExecutorDetails>>> entry : _topIdToUsedSlots.entrySet()) {
- String topoId = entry.getKey();
- Set<String> workerIds = entry.getValue().keySet();
- for (String workerId : workerIds) {
- if(ws.getId().equals(workerId)) {
- return _topologies.getById(topoId);
- }
- }
- }
- return null;
- }
-
- /**
- * Assigns a worker to a node
- * @param target the worker slot to assign the executors
- * @param td the topology the executors are from
- * @param executors executors to assign to the specified worker slot
- */
- public void assign(WorkerSlot target, TopologyDetails td, Collection<ExecutorDetails> executors) {
- if (!_isAlive) {
- throw new IllegalStateException("Trying to adding to a dead node " + _nodeId);
- }
- Collection<WorkerSlot> freeSlots = getFreeSlots();
- if (freeSlots.isEmpty()) {
- throw new IllegalStateException("Trying to assign to a full node " + _nodeId);
- }
- if (executors.size() == 0) {
- LOG.warn("Trying to assign nothing from " + td.getId() + " to " + _nodeId + " (Ignored)");
- }
- if (target == null) {
- target = getFreeSlots().iterator().next();
- }
- if (!freeSlots.contains(target)) {
- throw new IllegalStateException("Trying to assign already used slot" + target.getPort() + "on node " + _nodeId);
- }
- LOG.info("target slot: {}", target);
-
- _cluster.assign(target, td.getId(), executors);
-
- //assigning internally
- if (!_topIdToUsedSlots.containsKey(td.getId())) {
- _topIdToUsedSlots.put(td.getId(), new HashMap<String, Collection<ExecutorDetails>>());
- }
-
- if (!_topIdToUsedSlots.get(td.getId()).containsKey(target.getId())) {
- _topIdToUsedSlots.get(td.getId()).put(target.getId(), new LinkedList<ExecutorDetails>());
- }
- _topIdToUsedSlots.get(td.getId()).get(target.getId()).addAll(executors);
- }
-
- @Override
- public boolean equals(Object other) {
- if (other instanceof RAS_Node) {
- return _nodeId.equals(((RAS_Node) other)._nodeId);
- }
- return false;
- }
-
- @Override
- public int hashCode() {
- return _nodeId.hashCode();
- }
-
- @Override
- public String toString() {
- return "{Node: " + ((_sup == null) ? "null (possibly down)" : _sup.getHost())
- + ", Avail [ Mem: " + ((_availMemory == null) ? "N/A" : _availMemory.toString())
- + ", CPU: " + ((_availCPU == null) ? "N/A" : _availCPU.toString()) + ", Slots: " + this.getFreeSlots()
- + "] Total [ Mem: " + ((_sup == null) ? "N/A" : this.getTotalMemoryResources())
- + ", CPU: " + ((_sup == null) ? "N/A" : this.getTotalCpuResources()) + ", Slots: "
- + this._slots.values() + " ]}";
- }
-
- public static int countSlotsUsed(String topId, Collection<RAS_Node> nodes) {
- int total = 0;
- for (RAS_Node n : nodes) {
- total += n.totalSlotsUsed(topId);
- }
- return total;
- }
-
- public static int countSlotsUsed(Collection<RAS_Node> nodes) {
- int total = 0;
- for (RAS_Node n : nodes) {
- total += n.totalSlotsUsed();
- }
- return total;
- }
-
- public static int countFreeSlotsAlive(Collection<RAS_Node> nodes) {
- int total = 0;
- for (RAS_Node n : nodes) {
- if (n.isAlive()) {
- total += n.totalSlotsFree();
- }
- }
- return total;
- }
-
- public static int countTotalSlotsAlive(Collection<RAS_Node> nodes) {
- int total = 0;
- for (RAS_Node n : nodes) {
- if (n.isAlive()) {
- total += n.totalSlots();
- }
- }
- return total;
- }
-
- public static Collection<ExecutorDetails> getExecutors(WorkerSlot ws, Cluster cluster) {
- Collection<ExecutorDetails> retList = new ArrayList<ExecutorDetails>();
- for (Entry<String, SchedulerAssignment> entry : cluster.getAssignments()
- .entrySet()) {
- Map<ExecutorDetails, WorkerSlot> executorToSlot = entry.getValue()
- .getExecutorToSlot();
- for (Map.Entry<ExecutorDetails, WorkerSlot> execToSlot : executorToSlot
- .entrySet()) {
- WorkerSlot slot = execToSlot.getValue();
- if (ws.getPort() == slot.getPort()
- && ws.getNodeId().equals(slot.getNodeId())) {
- ExecutorDetails exec = execToSlot.getKey();
- retList.add(exec);
- }
- }
- }
- return retList;
- }
-
- /**
- * Sets the Available Memory for a node
- * @param amount the amount to set as available memory
- */
- public void setAvailableMemory(Double amount) {
- _availMemory = amount;
- }
-
- /**
- * Gets the available memory resources for this node
- * @return the available memory for this node
- */
- public Double getAvailableMemoryResources() {
- if (_availMemory == null) {
- return 0.0;
- }
- return _availMemory;
- }
-
- /**
- * Gets the total memory resources for this node
- * @return the total memory for this node
- */
- public Double getTotalMemoryResources() {
- if (_sup != null && _sup.getTotalMemory() != null) {
- return _sup.getTotalMemory();
- } else {
- return 0.0;
- }
- }
-
- /**
- * Consumes a certain amount of memory for this node
- * @param amount is the amount memory to consume from this node
- * @return the current available memory for this node after consumption
- */
- public Double consumeMemory(Double amount) {
- if (amount > _availMemory) {
- LOG.error("Attempting to consume more memory than available! Needed: {}, we only have: {}", amount, _availMemory);
- throw new IllegalStateException("Attempting to consume more memory than available");
- }
- _availMemory = _availMemory - amount;
- return _availMemory;
- }
-
- /**
- * Gets the available cpu resources for this node
- * @return the available cpu for this node
- */
- public Double getAvailableCpuResources() {
- if (_availCPU == null) {
- return 0.0;
- }
- return _availCPU;
- }
-
- /**
- * Gets the total cpu resources for this node
- * @return the total cpu for this node
- */
- public Double getTotalCpuResources() {
- if (_sup != null && _sup.getTotalCPU() != null) {
- return _sup.getTotalCPU();
- } else {
- return 0.0;
- }
- }
-
- /**
- * Consumes a certain amount of cpu for this node
- * @param amount is the amount cpu to consume from this node
- * @return the current available cpu for this node after consumption
- */
- public Double consumeCPU(Double amount) {
- if (amount > _availCPU) {
- LOG.error("Attempting to consume more CPU than available! Needed: {}, we only have: {}", amount, _availCPU);
- throw new IllegalStateException("Attempting to consume more CPU than available");
- }
- _availCPU = _availCPU - amount;
- return _availCPU;
- }
-
- /**
- * Consumes a certain amount of resources for a executor in a topology.
- * @param exec is the executor that is consuming resources on this node
- * @param topo the topology the executor is a part
- */
- public void consumeResourcesforTask(ExecutorDetails exec, TopologyDetails topo) {
- Double taskMemReq = topo.getTotalMemReqTask(exec);
- Double taskCpuReq = topo.getTotalCpuReqTask(exec);
- consumeCPU(taskCpuReq);
- consumeMemory(taskMemReq);
- }
-
- /**
- * frees the amount of resources for a executor in a topology.
- * @param exec is the executor for which the resources are freed for
- * @param topo the topology the executor is a part
- */
- public void freeResourcesForTask(ExecutorDetails exec, TopologyDetails topo) {
- Double taskMemReq = topo.getTotalMemReqTask(exec);
- Double taskCpuReq = topo.getTotalCpuReqTask(exec);
- freeCPU(taskCpuReq);
- freeMemory(taskMemReq);
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/b4d33955/storm-client/src/jvm/org/apache/storm/scheduler/resource/RAS_Nodes.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/resource/RAS_Nodes.java b/storm-client/src/jvm/org/apache/storm/scheduler/resource/RAS_Nodes.java
deleted file mode 100644
index 00795a3..0000000
--- a/storm-client/src/jvm/org/apache/storm/scheduler/resource/RAS_Nodes.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.storm.scheduler.resource;
-
-import org.apache.storm.scheduler.Cluster;
-import org.apache.storm.scheduler.ExecutorDetails;
-import org.apache.storm.scheduler.SchedulerAssignment;
-import org.apache.storm.scheduler.SupervisorDetails;
-import org.apache.storm.scheduler.Topologies;
-import org.apache.storm.scheduler.WorkerSlot;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.Map;
-
-public class RAS_Nodes {
-
- private Map<String, RAS_Node> nodeMap;
-
- private static final Logger LOG = LoggerFactory.getLogger(RAS_Nodes.class);
-
- public RAS_Nodes(Cluster cluster, Topologies topologies) {
- this.nodeMap = getAllNodesFrom(cluster, topologies);
- }
-
- public static Map<String, RAS_Node> getAllNodesFrom(Cluster cluster, Topologies topologies) {
-
- //A map of node ids to node objects
- Map<String, RAS_Node> nodeIdToNode = new HashMap<String, RAS_Node>();
- //A map of assignments organized by node with the following format:
- //{nodeId -> {topologyId -> {workerId -> {execs}}}}
- Map<String, Map<String, Map<String, Collection<ExecutorDetails>>>> assignmentRelationshipMap
- = new HashMap<String, Map<String, Map<String, Collection<ExecutorDetails>>>>();
-
- Map<String, Map<String, WorkerSlot>> workerIdToWorker = new HashMap<String, Map<String, WorkerSlot>>();
- for (SchedulerAssignment assignment : cluster.getAssignments().values()) {
- String topId = assignment.getTopologyId();
-
- for (Map.Entry<WorkerSlot, Collection<ExecutorDetails>> entry : assignment.getSlotToExecutors().entrySet()) {
- WorkerSlot slot = entry.getKey();
- String nodeId = slot.getNodeId();
- Collection<ExecutorDetails> execs = entry.getValue();
- if (!assignmentRelationshipMap.containsKey(nodeId)) {
- assignmentRelationshipMap.put(nodeId, new HashMap<String, Map<String, Collection<ExecutorDetails>>>());
- workerIdToWorker.put(nodeId, new HashMap<String, WorkerSlot>());
- }
- workerIdToWorker.get(nodeId).put(slot.getId(), slot);
- if (!assignmentRelationshipMap.get(nodeId).containsKey(topId)) {
- assignmentRelationshipMap.get(nodeId).put(topId, new HashMap<String, Collection<ExecutorDetails>>());
- }
- if (!assignmentRelationshipMap.get(nodeId).get(topId).containsKey(slot.getId())) {
- assignmentRelationshipMap.get(nodeId).get(topId).put(slot.getId(), new LinkedList<ExecutorDetails>());
- }
- assignmentRelationshipMap.get(nodeId).get(topId).get(slot.getId()).addAll(execs);
- }
- }
-
- for (SupervisorDetails sup : cluster.getSupervisors().values()) {
- //Initialize a worker slot for every port even if there is no assignment to it
- for (int port : sup.getAllPorts()) {
- WorkerSlot worker = new WorkerSlot(sup.getId(), port);
- if (!workerIdToWorker.containsKey(sup.getId())) {
- workerIdToWorker.put(sup.getId(), new HashMap<String, WorkerSlot>());
- }
- if (!workerIdToWorker.get(sup.getId()).containsKey(worker.getId())) {
- workerIdToWorker.get(sup.getId()).put(worker.getId(), worker);
- }
- }
- nodeIdToNode.put(sup.getId(), new RAS_Node(sup.getId(), sup, cluster, topologies, workerIdToWorker.get(sup.getId()), assignmentRelationshipMap.get(sup.getId())));
- }
-
- //Add in supervisors that might have crashed but workers are still alive
- for(Map.Entry<String, Map<String, Map<String, Collection<ExecutorDetails>>>> entry : assignmentRelationshipMap.entrySet()) {
- String nodeId = entry.getKey();
- Map<String, Map<String, Collection<ExecutorDetails>>> assignments = entry.getValue();
- if (!nodeIdToNode.containsKey(nodeId)) {
- LOG.info("Found an assigned slot(s) on a dead supervisor {} with assignments {}",
- nodeId, assignments);
- nodeIdToNode.put(nodeId, new RAS_Node(nodeId, null, cluster, topologies, workerIdToWorker.get(nodeId), assignments));
- }
- }
- return nodeIdToNode;
- }
-
- /**
- * get node object from nodeId
- */
- public RAS_Node getNodeById(String nodeId) {
- return this.nodeMap.get(nodeId);
- }
-
- /**
- *
- * @param workerSlots
- */
- public void freeSlots(Collection<WorkerSlot> workerSlots) {
- for (RAS_Node node : nodeMap.values()) {
- for (WorkerSlot ws : node.getUsedSlots()) {
- if (workerSlots.contains(ws)) {
- LOG.debug("freeing ws {} on node {}", ws, node);
- node.free(ws);
- }
- }
- }
- }
-
- public Collection<RAS_Node> getNodes() {
- return this.nodeMap.values();
- }
-
- @Override
- public String toString() {
- String ret = "";
- for (RAS_Node node : this.nodeMap.values()) {
- ret += node.toString() + "\n";
- }
- return ret;
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/b4d33955/storm-client/src/jvm/org/apache/storm/scheduler/resource/ResourceUtils.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/resource/ResourceUtils.java b/storm-client/src/jvm/org/apache/storm/scheduler/resource/ResourceUtils.java
deleted file mode 100644
index 85c04f2..0000000
--- a/storm-client/src/jvm/org/apache/storm/scheduler/resource/ResourceUtils.java
+++ /dev/null
@@ -1,207 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.storm.scheduler.resource;
-
-import org.apache.storm.Config;
-import org.apache.storm.generated.Bolt;
-import org.apache.storm.generated.SpoutSpec;
-import org.apache.storm.generated.StormTopology;
-import org.apache.storm.scheduler.Cluster;
-import org.apache.storm.scheduler.ExecutorDetails;
-import org.apache.storm.scheduler.Topologies;
-import org.apache.storm.scheduler.TopologyDetails;
-import org.apache.storm.scheduler.WorkerSlot;
-import org.apache.storm.utils.ObjectReader;
-import org.json.simple.JSONObject;
-import org.json.simple.parser.JSONParser;
-import org.json.simple.parser.ParseException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.Map;
-
-public class ResourceUtils {
- private static final Logger LOG = LoggerFactory.getLogger(ResourceUtils.class);
-
- public static Map<String, Map<String, Double>> getBoltsResources(StormTopology topology, Map<String, Object> topologyConf) {
- Map<String, Map<String, Double>> boltResources = new HashMap<String, Map<String, Double>>();
- if (topology.get_bolts() != null) {
- for (Map.Entry<String, Bolt> bolt : topology.get_bolts().entrySet()) {
- Map<String, Double> topology_resources = parseResources(bolt.getValue().get_common().get_json_conf());
- checkIntialization(topology_resources, bolt.getValue().toString(), topologyConf);
- boltResources.put(bolt.getKey(), topology_resources);
- }
- }
- return boltResources;
- }
-
- public static Map<String, Map<String, Double>> getSpoutsResources(StormTopology topology, Map<String, Object> topologyConf) {
- Map<String, Map<String, Double>> spoutResources = new HashMap<String, Map<String, Double>>();
- if (topology.get_spouts() != null) {
- for (Map.Entry<String, SpoutSpec> spout : topology.get_spouts().entrySet()) {
- Map<String, Double> topology_resources = parseResources(spout.getValue().get_common().get_json_conf());
- checkIntialization(topology_resources, spout.getValue().toString(), topologyConf);
- spoutResources.put(spout.getKey(), topology_resources);
- }
- }
- return spoutResources;
- }
-
- public static void checkIntialization(Map<String, Double> topology_resources, String Com, Map<String, Object> topologyConf) {
- checkInitMem(topology_resources, Com, topologyConf);
- checkInitCPU(topology_resources, Com, topologyConf);
- }
-
- public static void checkInitMem(Map<String, Double> topology_resources, String Com, Map<String, Object> topologyConf) {
- if (!topology_resources.containsKey(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB)) {
- topology_resources.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB,
- ObjectReader.getDouble(topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB), null));
- debugMessage("ONHEAP", Com, topologyConf);
- }
- if (!topology_resources.containsKey(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB)) {
- topology_resources.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB,
- ObjectReader.getDouble(topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB), null));
- debugMessage("OFFHEAP", Com, topologyConf);
- }
- }
-
- public static void checkInitCPU(Map<String, Double> topology_resources, String Com, Map<String, Object> topologyConf) {
- if (!topology_resources.containsKey(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT)) {
- topology_resources.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT,
- ObjectReader.getDouble(topologyConf.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT), null));
- debugMessage("CPU", Com, topologyConf);
- }
- }
-
- public static Map<String, Double> parseResources(String input) {
- Map<String, Double> topology_resources = new HashMap<String, Double>();
- JSONParser parser = new JSONParser();
- LOG.debug("Input to parseResources {}", input);
- try {
- if (input != null) {
- Object obj = parser.parse(input);
- JSONObject jsonObject = (JSONObject) obj;
- if (jsonObject.containsKey(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB)) {
- Double topoMemOnHeap = ObjectReader
- .getDouble(jsonObject.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB), null);
- topology_resources.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, topoMemOnHeap);
- }
- if (jsonObject.containsKey(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB)) {
- Double topoMemOffHeap = ObjectReader
- .getDouble(jsonObject.get(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB), null);
- topology_resources.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, topoMemOffHeap);
- }
- if (jsonObject.containsKey(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT)) {
- Double topoCPU = ObjectReader.getDouble(jsonObject.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT), null);
- topology_resources.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, topoCPU);
- }
- LOG.debug("Topology Resources {}", topology_resources);
- }
- } catch (ParseException e) {
- LOG.error("Failed to parse component resources is:" + e.toString(), e);
- return null;
- }
- return topology_resources;
- }
-
- private static void debugMessage(String memoryType, String Com, Map<String, Object> topologyConf) {
- if (memoryType.equals("ONHEAP")) {
- LOG.debug(
- "Unable to extract resource requirement for Component {} \n Resource : Memory Type : On Heap set to default {}",
- Com, topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB));
- } else if (memoryType.equals("OFFHEAP")) {
- LOG.debug(
- "Unable to extract resource requirement for Component {} \n Resource : Memory Type : Off Heap set to default {}",
- Com, topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB));
- } else {
- LOG.debug(
- "Unable to extract resource requirement for Component {} \n Resource : CPU Pcore Percent set to default {}",
- Com, topologyConf.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT));
- }
- }
-
- /**
- * print scheduling for debug purposes
- * @param cluster
- * @param topologies
- */
- public static String printScheduling(Cluster cluster, Topologies topologies) {
- StringBuilder str = new StringBuilder();
- Map<String, Map<String, Map<WorkerSlot, Collection<ExecutorDetails>>>> schedulingMap = new HashMap<String, Map<String, Map<WorkerSlot, Collection<ExecutorDetails>>>>();
- for (TopologyDetails topo : topologies.getTopologies()) {
- if (cluster.getAssignmentById(topo.getId()) != null) {
- for (Map.Entry<ExecutorDetails, WorkerSlot> entry : cluster.getAssignmentById(topo.getId()).getExecutorToSlot().entrySet()) {
- WorkerSlot slot = entry.getValue();
- String nodeId = slot.getNodeId();
- ExecutorDetails exec = entry.getKey();
- if (!schedulingMap.containsKey(nodeId)) {
- schedulingMap.put(nodeId, new HashMap<String, Map<WorkerSlot, Collection<ExecutorDetails>>>());
- }
- if (schedulingMap.get(nodeId).containsKey(topo.getId()) == false) {
- schedulingMap.get(nodeId).put(topo.getId(), new HashMap<WorkerSlot, Collection<ExecutorDetails>>());
- }
- if (schedulingMap.get(nodeId).get(topo.getId()).containsKey(slot) == false) {
- schedulingMap.get(nodeId).get(topo.getId()).put(slot, new LinkedList<ExecutorDetails>());
- }
- schedulingMap.get(nodeId).get(topo.getId()).get(slot).add(exec);
- }
- }
- }
-
- for (Map.Entry<String, Map<String, Map<WorkerSlot, Collection<ExecutorDetails>>>> entry : schedulingMap.entrySet()) {
- if (cluster.getSupervisorById(entry.getKey()) != null) {
- str.append("/** Node: " + cluster.getSupervisorById(entry.getKey()).getHost() + "-" + entry.getKey() + " **/\n");
- } else {
- str.append("/** Node: Unknown may be dead -" + entry.getKey() + " **/\n");
- }
- for (Map.Entry<String, Map<WorkerSlot, Collection<ExecutorDetails>>> topo_sched : schedulingMap.get(entry.getKey()).entrySet()) {
- str.append("\t-->Topology: " + topo_sched.getKey() + "\n");
- for (Map.Entry<WorkerSlot, Collection<ExecutorDetails>> ws : topo_sched.getValue().entrySet()) {
- str.append("\t\t->Slot [" + ws.getKey().getPort() + "] -> " + ws.getValue() + "\n");
- }
- }
- }
- return str.toString();
- }
-
- /**
- * Calculate the sum of a collection of doubles
- * @param list collection of doubles
- * @return the sum of of collection of doubles
- */
- public static double sum(Collection<Double> list) {
- double sum = 0.0;
- for (Double elem : list) {
- sum += elem;
- }
- return sum;
- }
-
- /**
- * Caculate the average of a collection of doubles
- * @param list a collection of doubles
- * @return the average of collection of doubles
- */
- public static double avg(Collection<Double> list) {
- return sum(list) / list.size();
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/b4d33955/storm-client/src/jvm/org/apache/storm/scheduler/resource/SchedulingResult.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/resource/SchedulingResult.java b/storm-client/src/jvm/org/apache/storm/scheduler/resource/SchedulingResult.java
deleted file mode 100644
index 6a482d8..0000000
--- a/storm-client/src/jvm/org/apache/storm/scheduler/resource/SchedulingResult.java
+++ /dev/null
@@ -1,116 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.storm.scheduler.resource;
-
-import org.apache.storm.scheduler.ExecutorDetails;
-import org.apache.storm.scheduler.WorkerSlot;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Collection;
-import java.util.Map;
-
-/**
- * This class serves as a mechanism to return results and messages from a scheduling strategy to the Resource Aware Scheduler
- */
-public class SchedulingResult {
-
- //contains the result for the attempted scheduling
- private Map<WorkerSlot, Collection<ExecutorDetails>> schedulingResultMap = null;
-
- //status of scheduling the topology e.g. success or fail?
- private SchedulingStatus status = null;
-
- //arbitrary message to be returned when scheduling is done
- private String message = null;
-
- //error message returned is something went wrong
- private String errorMessage = null;
-
- private static final Logger LOG = LoggerFactory.getLogger(SchedulingResult.class);
-
- private SchedulingResult(SchedulingStatus status, Map<WorkerSlot, Collection<ExecutorDetails>> schedulingResultMap, String message, String errorMessage) {
- this.status = status;
- this.schedulingResultMap = schedulingResultMap;
- this.message = message;
- this.errorMessage = errorMessage;
- }
-
- public static SchedulingResult failure(SchedulingStatus status, String errorMessage) {
- return new SchedulingResult(status, null, null, errorMessage);
- }
-
- public static SchedulingResult success(Map<WorkerSlot, Collection<ExecutorDetails>> schedulingResultMap) {
- return SchedulingResult.successWithMsg(schedulingResultMap, null);
- }
-
- public static SchedulingResult successWithMsg(Map<WorkerSlot, Collection<ExecutorDetails>> schedulingResultMap, String message) {
- if (schedulingResultMap == null) {
- throw new IllegalStateException("Cannot declare scheduling success without providing a non null scheduling map!");
- }
- return new SchedulingResult(SchedulingStatus.SUCCESS, schedulingResultMap, message, null);
- }
-
- public SchedulingStatus getStatus() {
- return this.status;
- }
-
- public String getMessage() {
- return this.message;
- }
-
- public String getErrorMessage() {
- return this.errorMessage;
- }
-
- public Map<WorkerSlot, Collection<ExecutorDetails>> getSchedulingResultMap() {
- return this.schedulingResultMap;
- }
-
- public boolean isSuccess() {
- return SchedulingStatus.isStatusSuccess(this.status);
- }
-
- public boolean isFailure() {
- return SchedulingStatus.isStatusFailure(this.status);
- }
-
- public boolean isValid() {
- if (this.isSuccess() && this.getSchedulingResultMap() == null) {
- LOG.warn("SchedulingResult not Valid! Status is success but SchedulingResultMap is null");
- return false;
- }
- if (this.isFailure() && this.getSchedulingResultMap() != null) {
- LOG.warn("SchedulingResult not Valid! Status is Failure but SchedulingResultMap is NOT null");
- return false;
- }
- return true;
- }
-
- @Override
- public String toString() {
- String ret = null;
- if(this.isSuccess()) {
- ret = "Status: " + this.getStatus() + " message: " + this.getMessage() + " scheduling: " + this.getSchedulingResultMap();
- } else {
- ret = "Status: " + this.getStatus() + " error message: " + this.getErrorMessage();
- }
- return ret;
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/b4d33955/storm-client/src/jvm/org/apache/storm/scheduler/resource/SchedulingState.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/resource/SchedulingState.java b/storm-client/src/jvm/org/apache/storm/scheduler/resource/SchedulingState.java
deleted file mode 100644
index 80f5e9b..0000000
--- a/storm-client/src/jvm/org/apache/storm/scheduler/resource/SchedulingState.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.storm.scheduler.resource;
-
-import org.apache.storm.Config;
-import org.apache.storm.scheduler.Cluster;
-import org.apache.storm.scheduler.Topologies;
-
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * Class that holds the completely scheduling state of Resource Aware Scheduler
- */
-public class SchedulingState {
- public final Map<String, User> userMap = new HashMap<String, User>();
- public final Cluster cluster;
- public final Topologies topologies;
- public final RAS_Nodes nodes;
- public final Map<String, Object> conf = new Config();
-
- public SchedulingState(Map<String, User> userMap, Cluster cluster, Topologies topologies, Map<String, Object> conf) {
- for (Map.Entry<String, User> userMapEntry : userMap.entrySet()) {
- String userId = userMapEntry.getKey();
- User user = userMapEntry.getValue();
- this.userMap.put(userId, new User(user));
- }
- this.cluster = new Cluster(cluster);
- this.topologies = new Topologies(topologies);
- this.nodes = new RAS_Nodes(this.cluster, this.topologies);
- this.conf.putAll(conf);
- }
-
- /**
- * copy constructor
- */
- public SchedulingState(SchedulingState src) {
- this(src.userMap, src.cluster, src.topologies, src.conf);
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/b4d33955/storm-client/src/jvm/org/apache/storm/scheduler/resource/SchedulingStatus.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/resource/SchedulingStatus.java b/storm-client/src/jvm/org/apache/storm/scheduler/resource/SchedulingStatus.java
deleted file mode 100644
index 27a0e0b..0000000
--- a/storm-client/src/jvm/org/apache/storm/scheduler/resource/SchedulingStatus.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.storm.scheduler.resource;
-
-import java.util.EnumSet;
-
-public enum SchedulingStatus {
- SUCCESS,
- FAIL_NOT_ENOUGH_RESOURCES,
- FAIL_INVALID_TOPOLOGY,
- FAIL_OTHER;
-
- public static EnumSet<SchedulingStatus> success = EnumSet.of(SUCCESS);
- public static EnumSet<SchedulingStatus> failure = EnumSet.of(FAIL_INVALID_TOPOLOGY, FAIL_NOT_ENOUGH_RESOURCES, FAIL_OTHER);
-
- public static boolean isStatusSuccess(SchedulingStatus status) {
- return success.contains(status);
- }
-
- public static boolean isStatusFailure(SchedulingStatus status) {
- return failure.contains(status);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/b4d33955/storm-client/src/jvm/org/apache/storm/scheduler/resource/User.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/resource/User.java b/storm-client/src/jvm/org/apache/storm/scheduler/resource/User.java
deleted file mode 100644
index 0f5a563..0000000
--- a/storm-client/src/jvm/org/apache/storm/scheduler/resource/User.java
+++ /dev/null
@@ -1,350 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.storm.scheduler.resource;
-
-import org.apache.storm.scheduler.Cluster;
-import org.apache.storm.scheduler.TopologyDetails;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Collection;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeSet;
-
-public class User {
- private String userId;
- //Topologies yet to be scheduled sorted by priority for each user
- private TreeSet<TopologyDetails> pendingQueue = new TreeSet<TopologyDetails>(new PQsortByPriorityAndSubmittionTime());
-
- //Topologies yet to be scheduled sorted by priority for each user
- private TreeSet<TopologyDetails> runningQueue = new TreeSet<TopologyDetails>(new PQsortByPriorityAndSubmittionTime());
-
- //Topologies that was attempted to be scheduled but wasn't successful
- private TreeSet<TopologyDetails> attemptedQueue = new TreeSet<TopologyDetails>(new PQsortByPriorityAndSubmittionTime());
-
- //Topologies that was deemed to be invalid
- private TreeSet<TopologyDetails> invalidQueue = new TreeSet<TopologyDetails>(new PQsortByPriorityAndSubmittionTime());
-
- private Map<String, Double> resourcePool = new HashMap<String, Double>();
-
- private static final Logger LOG = LoggerFactory.getLogger(User.class);
-
- public User(String userId) {
- this.userId = userId;
- }
-
- public User(String userId, Map<String, Double> resourcePool) {
- this(userId);
- if (resourcePool != null) {
- this.resourcePool.putAll(resourcePool);
- }
- if (this.resourcePool.get("cpu") == null) {
- this.resourcePool.put("cpu", 0.0);
- }
- if (this.resourcePool.get("memory") == null) {
- this.resourcePool.put("memory", 0.0);
- }
- }
-
- /**
- * Copy Constructor
- */
- public User(User src) {
- this(src.userId, src.resourcePool);
- for (TopologyDetails topo : src.pendingQueue) {
- addTopologyToPendingQueue(topo);
- }
- for (TopologyDetails topo : src.runningQueue) {
- addTopologyToRunningQueue(topo);
- }
- for (TopologyDetails topo : src.attemptedQueue) {
- addTopologyToAttemptedQueue(topo);
- }
- for (TopologyDetails topo : src.invalidQueue) {
- addTopologyToInvalidQueue(topo);
- }
- }
-
- public String getId() {
- return this.userId;
- }
-
- public void addTopologyToPendingQueue(TopologyDetails topo, Cluster cluster) {
- this.pendingQueue.add(topo);
- if (cluster != null) {
- cluster.setStatus(topo.getId(), "Scheduling Pending");
- }
- }
-
- public void addTopologyToPendingQueue(TopologyDetails topo) {
- this.addTopologyToPendingQueue(topo, null);
- }
-
- public void addTopologyToRunningQueue(TopologyDetails topo, Cluster cluster) {
- this.runningQueue.add(topo);
- if (cluster != null) {
- cluster.setStatus(topo.getId(), "Fully Scheduled");
- }
- }
-
- public void addTopologyToRunningQueue(TopologyDetails topo) {
- this.addTopologyToRunningQueue(topo, null);
- }
-
- public Set<TopologyDetails> getTopologiesPending() {
- TreeSet<TopologyDetails> ret = new TreeSet<TopologyDetails>(new PQsortByPriorityAndSubmittionTime());
- ret.addAll(this.pendingQueue);
- return ret;
- }
-
- public void addTopologyToAttemptedQueue(TopologyDetails topo) {
- this.attemptedQueue.add(topo);
- }
-
- public void addTopologyToInvalidQueue(TopologyDetails topo) {
- this.invalidQueue.add(topo);
- }
-
- public Set<TopologyDetails> getTopologiesRunning() {
- TreeSet<TopologyDetails> ret = new TreeSet<TopologyDetails>(new PQsortByPriorityAndSubmittionTime());
- ret.addAll(this.runningQueue);
- return ret;
- }
-
- public Set<TopologyDetails> getTopologiesAttempted() {
- TreeSet<TopologyDetails> ret = new TreeSet<TopologyDetails>(new PQsortByPriorityAndSubmittionTime());
- ret.addAll(this.attemptedQueue);
- return ret;
- }
-
- public Set<TopologyDetails> getTopologiesInvalid() {
- TreeSet<TopologyDetails> ret = new TreeSet<TopologyDetails>(new PQsortByPriorityAndSubmittionTime());
- ret.addAll(this.invalidQueue);
- return ret;
- }
-
- public Map<String, Number> getResourcePool() {
- if (this.resourcePool != null) {
- return new HashMap<String, Number>(this.resourcePool);
- }
- return null;
- }
-
- public void moveTopoFromPendingToRunning(TopologyDetails topo, Cluster cluster) {
- moveTopology(topo, this.pendingQueue, "pending", this.runningQueue, "running");
- if (cluster != null) {
- cluster.setStatus(topo.getId(), "Fully Scheduled");
- }
- }
-
- public void moveTopoFromPendingToRunning(TopologyDetails topo) {
- this.moveTopoFromPendingToRunning(topo, null);
- }
-
- public void moveTopoFromPendingToAttempted(TopologyDetails topo, Cluster cluster) {
- moveTopology(topo, this.pendingQueue, "pending", this.attemptedQueue, "attempted");
- if (cluster != null) {
- cluster.setStatus(topo.getId(), "Scheduling Attempted but Failed");
- }
- }
-
- public void moveTopoFromPendingToAttempted(TopologyDetails topo) {
- this.moveTopoFromPendingToAttempted(topo, null);
- }
-
- public void moveTopoFromPendingToInvalid(TopologyDetails topo, Cluster cluster) {
- moveTopology(topo, this.pendingQueue, "pending", this.invalidQueue, "invalid");
- if (cluster != null) {
- cluster.setStatus(topo.getId(), "Scheduling Attempted but topology is invalid");
- }
- }
-
- public void moveTopoFromPendingToInvalid(TopologyDetails topo) {
- this.moveTopoFromPendingToInvalid(topo, null);
- }
-
- public void moveTopoFromRunningToPending(TopologyDetails topo, Cluster cluster) {
- moveTopology(topo, this.runningQueue, "running", this.pendingQueue, "pending");
- if (cluster != null) {
- cluster.setStatus(topo.getId(), "Scheduling Pending");
- }
- }
-
- public void moveTopoFromRunningToPending(TopologyDetails topo) {
- this.moveTopoFromRunningToPending(topo, null);
- }
-
- private void moveTopology(TopologyDetails topo, Set<TopologyDetails> src, String srcName, Set<TopologyDetails> dest, String destName) {
- if (topo == null) {
- return;
- }
-
- LOG.debug("For User {} Moving topo {} from {} to {}", this.userId, topo.getName(), srcName, destName);
-
- if (!src.contains(topo)) {
- LOG.warn("Topo {} not in User: {} {} queue!", topo.getName(), this.userId, srcName);
- return;
- }
- if (dest.contains(topo)) {
- LOG.warn("Topo {} already in in User: {} {} queue!", topo.getName(), this.userId, destName);
- return;
- }
- src.remove(topo);
- dest.add(topo);
- }
-
- public double getResourcePoolAverageUtilization() {
- Double cpuResourcePoolUtilization = this.getCPUResourcePoolUtilization();
- Double memoryResourcePoolUtilization = this.getMemoryResourcePoolUtilization();
-
- if (cpuResourcePoolUtilization != null && memoryResourcePoolUtilization != null) {
- //cannot be (cpuResourcePoolUtilization + memoryResourcePoolUtilization)/2
- //since memoryResourcePoolUtilization or cpuResourcePoolUtilization can be Double.MAX_VALUE
- //Should not return infinity in that case
- return ((cpuResourcePoolUtilization) / 2.0) + ((memoryResourcePoolUtilization) / 2.0);
- }
- return Double.MAX_VALUE;
- }
-
- public double getCPUResourcePoolUtilization() {
- Double cpuGuarantee = this.resourcePool.get("cpu");
- if (cpuGuarantee == null || cpuGuarantee == 0.0) {
- return Double.MAX_VALUE;
- }
- return this.getCPUResourceUsedByUser() / cpuGuarantee;
- }
-
- public double getMemoryResourcePoolUtilization() {
- Double memoryGuarantee = this.resourcePool.get("memory");
- if (memoryGuarantee == null || memoryGuarantee == 0.0) {
- return Double.MAX_VALUE;
- }
- return this.getMemoryResourceUsedByUser() / memoryGuarantee;
- }
-
- public double getCPUResourceUsedByUser() {
- double sum = 0.0;
- for (TopologyDetails topo : this.runningQueue) {
- sum += topo.getTotalRequestedCpu();
- }
- return sum;
- }
-
- public double getMemoryResourceUsedByUser() {
- double sum = 0.0;
- for (TopologyDetails topo : this.runningQueue) {
- sum += topo.getTotalRequestedMemOnHeap() + topo.getTotalRequestedMemOffHeap();
- }
- return sum;
- }
-
- public Double getMemoryResourceGuaranteed() {
- return this.resourcePool.get("memory");
- }
-
- public Double getCPUResourceGuaranteed() {
- return this.resourcePool.get("cpu");
- }
-
- public TopologyDetails getNextTopologyToSchedule() {
- for (TopologyDetails topo : this.pendingQueue) {
- if (!this.attemptedQueue.contains(topo)) {
- return topo;
- }
- }
- return null;
- }
-
- public boolean hasTopologyNeedSchedule() {
- return (!this.pendingQueue.isEmpty());
- }
-
- public TopologyDetails getRunningTopologyWithLowestPriority() {
- if (this.runningQueue.isEmpty()) {
- return null;
- }
- return this.runningQueue.last();
- }
-
- @Override
- public int hashCode() {
- return this.userId.hashCode();
- }
-
- @Override
- public boolean equals(Object o) {
- if (!(o instanceof User)) {
- return false;
- }
- return this.getId().equals(((User) o).getId());
- }
-
- @Override
- public String toString() {
- return this.userId;
- }
-
- public String getDetailedInfo() {
- String ret = "\nUser: " + this.userId;
- ret += "\n - " + " Resource Pool: " + this.resourcePool;
- ret += "\n - " + " Running Queue: " + this.runningQueue + " size: " + this.runningQueue.size();
- ret += "\n - " + " Pending Queue: " + this.pendingQueue + " size: " + this.pendingQueue.size();
- ret += "\n - " + " Attempted Queue: " + this.attemptedQueue + " size: " + this.attemptedQueue.size();
- ret += "\n - " + " Invalid Queue: " + this.invalidQueue + " size: " + this.invalidQueue.size();
- ret += "\n - " + " CPU Used: " + this.getCPUResourceUsedByUser() + " CPU guaranteed: " + this.getCPUResourceGuaranteed();
- ret += "\n - " + " Memory Used: " + this.getMemoryResourceUsedByUser() + " Memory guaranteed: " + this.getMemoryResourceGuaranteed();
- ret += "\n - " + " % Resource Guarantee Used: \n -- CPU: " + this.getCPUResourcePoolUtilization()
- + " Memory: " + this.getMemoryResourcePoolUtilization() + " Average: " + this.getResourcePoolAverageUtilization();
- return ret;
- }
-
- public static String getResourcePoolAverageUtilizationForUsers(Collection<User> users) {
- String ret = "";
- for (User user : users) {
- ret += user.getId() + " - " + user.getResourcePoolAverageUtilization() + " ";
- }
- return ret;
- }
-
- /**
- * Comparator that sorts topologies by priority and then by submission time
- * First sort by Topology Priority, if there is a tie for topology priority, topology uptime is used to sort
- */
- static class PQsortByPriorityAndSubmittionTime implements Comparator<TopologyDetails> {
-
- public int compare(TopologyDetails topo1, TopologyDetails topo2) {
- if (topo1.getTopologyPriority() > topo2.getTopologyPriority()) {
- return 1;
- } else if (topo1.getTopologyPriority() < topo2.getTopologyPriority()) {
- return -1;
- } else {
- if (topo1.getUpTime() > topo2.getUpTime()) {
- return -1;
- } else if (topo1.getUpTime() < topo2.getUpTime()) {
- return 1;
- } else {
- return topo1.getId().compareTo(topo2.getId());
- }
- }
- }
- }
-}