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/11/25 00:40:56 UTC
[2/3] storm git commit: STORM-2813: Use a class for normalized
resources not a map.
STORM-2813: Use a class for normalized resources not a map.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/6a4aeb91
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/6a4aeb91
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/6a4aeb91
Branch: refs/heads/master
Commit: 6a4aeb9101bb1e6b196fa348ece03e55807391d1
Parents: 9d52435
Author: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Authored: Tue Nov 14 12:16:00 2017 -0600
Committer: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Committed: Mon Nov 20 13:47:47 2017 -0600
----------------------------------------------------------------------
storm-client/pom.xml | 2 +-
.../src/jvm/org/apache/storm/Constants.java | 12 -
.../apache/storm/scheduler/ExecutorDetails.java | 54 ----
.../storm/scheduler/SchedulerAssignment.java | 72 -----
.../scheduler/SchedulerAssignmentImpl.java | 239 ---------------
.../storm/scheduler/SupervisorDetails.java | 147 ----------
storm-server/pom.xml | 2 +-
.../org/apache/storm/daemon/nimbus/Nimbus.java | 49 ++--
.../supervisor/timer/SupervisorHeartbeat.java | 3 +-
.../org/apache/storm/scheduler/Cluster.java | 114 ++------
.../apache/storm/scheduler/ExecutorDetails.java | 57 ++++
.../storm/scheduler/ISchedulingState.java | 8 +-
.../storm/scheduler/SchedulerAssignment.java | 85 ++++++
.../scheduler/SchedulerAssignmentImpl.java | 251 ++++++++++++++++
.../storm/scheduler/SupervisorDetails.java | 153 ++++++++++
.../apache/storm/scheduler/TopologyDetails.java | 77 +----
.../resource/NormalizedResourceOffer.java | 89 ++++++
.../resource/NormalizedResourceRequest.java | 194 +++++++++++++
.../scheduler/resource/NormalizedResources.java | 290 +++++++++++++++++++
.../storm/scheduler/resource/RAS_Node.java | 55 +---
.../storm/scheduler/resource/ResourceUtils.java | 222 ++------------
.../scheduling/BaseResourceAwareStrategy.java | 81 +++---
.../DefaultResourceAwareStrategy.java | 39 +--
.../GenericResourceAwareStrategy.java | 36 +--
.../org/apache/storm/utils/ServerUtils.java | 18 +-
.../resource/TestResourceAwareScheduler.java | 27 +-
.../TestUtilsForResourceAwareScheduler.java | 2 +-
.../TestDefaultResourceAwareStrategy.java | 9 +-
28 files changed, 1318 insertions(+), 1069 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/6a4aeb91/storm-client/pom.xml
----------------------------------------------------------------------
diff --git a/storm-client/pom.xml b/storm-client/pom.xml
index 6097b35..22c8bcc 100644
--- a/storm-client/pom.xml
+++ b/storm-client/pom.xml
@@ -261,7 +261,7 @@
<!--Note - the version would be inherited-->
<configuration>
<excludes>**/generated/**</excludes>
- <maxAllowedViolations>10298</maxAllowedViolations>
+ <maxAllowedViolations>10263</maxAllowedViolations>
</configuration>
</plugin>
<plugin>
http://git-wip-us.apache.org/repos/asf/storm/blob/6a4aeb91/storm-client/src/jvm/org/apache/storm/Constants.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/Constants.java b/storm-client/src/jvm/org/apache/storm/Constants.java
index c8d3b09..a8e2c01 100644
--- a/storm-client/src/jvm/org/apache/storm/Constants.java
+++ b/storm-client/src/jvm/org/apache/storm/Constants.java
@@ -64,17 +64,5 @@ public class Constants {
public static final String COMMON_ONHEAP_MEMORY_RESOURCE_NAME = "onheap.memory.mb";
public static final String COMMON_OFFHEAP_MEMORY_RESOURCE_NAME = "offheap.memory.mb";
public static final String COMMON_TOTAL_MEMORY_RESOURCE_NAME = "memory.mb";
-
- public static final Map<String, String> resourceNameMapping;
-
- static {
- Map<String, String> tmp = new HashMap<>();
- tmp.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, COMMON_CPU_RESOURCE_NAME);
- tmp.put(Config.SUPERVISOR_CPU_CAPACITY, COMMON_CPU_RESOURCE_NAME);
- tmp.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, COMMON_ONHEAP_MEMORY_RESOURCE_NAME);
- tmp.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, COMMON_OFFHEAP_MEMORY_RESOURCE_NAME);
- tmp.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, COMMON_TOTAL_MEMORY_RESOURCE_NAME);
- resourceNameMapping = Collections.unmodifiableMap(tmp);
- }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/6a4aeb91/storm-client/src/jvm/org/apache/storm/scheduler/ExecutorDetails.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/ExecutorDetails.java b/storm-client/src/jvm/org/apache/storm/scheduler/ExecutorDetails.java
deleted file mode 100644
index d1e776d..0000000
--- a/storm-client/src/jvm/org/apache/storm/scheduler/ExecutorDetails.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
- *
- * 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;
-
-public class ExecutorDetails {
- int startTask;
- int endTask;
-
- public ExecutorDetails(int startTask, int endTask){
- this.startTask = startTask;
- this.endTask = endTask;
- }
-
- public int getStartTask() {
- return startTask;
- }
-
- public int getEndTask() {
- return endTask;
- }
-
- public boolean equals(Object other) {
- if (other == null || !(other instanceof ExecutorDetails)) {
- return false;
- }
-
- ExecutorDetails executor = (ExecutorDetails)other;
- return (this.startTask == executor.startTask) && (this.endTask == executor.endTask);
- }
-
- public int hashCode() {
- return this.startTask + 13 * this.endTask;
- }
-
- @Override
- public String toString() {
- return "[" + this.startTask + ", " + this.endTask + "]";
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/6a4aeb91/storm-client/src/jvm/org/apache/storm/scheduler/SchedulerAssignment.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/SchedulerAssignment.java b/storm-client/src/jvm/org/apache/storm/scheduler/SchedulerAssignment.java
deleted file mode 100644
index 58ee98a..0000000
--- a/storm-client/src/jvm/org/apache/storm/scheduler/SchedulerAssignment.java
+++ /dev/null
@@ -1,72 +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.Collection;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.storm.generated.Assignment;
-import org.apache.storm.generated.WorkerResources;
-
-public interface SchedulerAssignment {
- /**
- * Does this slot occupied by this assignment?
- * @param slot
- * @return true if the slot is occupied else false
- */
- public boolean isSlotOccupied(WorkerSlot slot);
-
- /**
- * Is the executor assigned?
- *
- * @param executor
- * @return true if it is assigned else false
- */
- public boolean isExecutorAssigned(ExecutorDetails executor);
-
- /**
- * Return the ID of the topology.
- * @return the topology-id this assignment is for.
- */
- public String getTopologyId();
-
- /**
- * @return the executor -> slot map.
- */
- public Map<ExecutorDetails, WorkerSlot> getExecutorToSlot();
-
- /**
- * @return the executors covered by this assignments
- */
- public Set<ExecutorDetails> getExecutors();
-
- public Set<WorkerSlot> getSlots();
-
- public Map<WorkerSlot, Collection<ExecutorDetails>> getSlotToExecutors();
-
- /**
- * @return The slot to resource mapping
- */
- public Map<WorkerSlot, WorkerResources> getScheduledResources();
-
- /**
- * @return host to total shared off heap memory mapping.
- */
- public Map<String, Double> getNodeIdToTotalSharedOffHeapMemory();
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/6a4aeb91/storm-client/src/jvm/org/apache/storm/scheduler/SchedulerAssignmentImpl.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/SchedulerAssignmentImpl.java b/storm-client/src/jvm/org/apache/storm/scheduler/SchedulerAssignmentImpl.java
deleted file mode 100644
index bff68fe..0000000
--- a/storm-client/src/jvm/org/apache/storm/scheduler/SchedulerAssignmentImpl.java
+++ /dev/null
@@ -1,239 +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.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.storm.generated.WorkerResources;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class SchedulerAssignmentImpl implements SchedulerAssignment {
- private static final Logger LOG = LoggerFactory.getLogger(SchedulerAssignmentImpl.class);
-
- /**
- * topology-id this assignment is for.
- */
- private final String topologyId;
-
- /**
- * assignment detail, a mapping from executor to <code>WorkerSlot</code>
- */
- private final Map<ExecutorDetails, WorkerSlot> executorToSlot = new HashMap<>();
- private final Map<WorkerSlot, WorkerResources> resources = new HashMap<>();
- private final Map<String, Double> nodeIdToTotalSharedOffHeap = new HashMap<>();
- //Used to cache the slotToExecutors mapping.
- private Map<WorkerSlot, Collection<ExecutorDetails>> slotToExecutorsCache = null;
-
- public SchedulerAssignmentImpl(String topologyId, Map<ExecutorDetails, WorkerSlot> executorToSlot,
- Map<WorkerSlot, WorkerResources> resources, Map<String, Double> nodeIdToTotalSharedOffHeap) {
- this.topologyId = topologyId;
- if (executorToSlot != null) {
- if (executorToSlot.entrySet().stream().anyMatch((entry) -> entry.getKey() == null || entry.getValue() == null)) {
- throw new RuntimeException("Cannot create a scheduling with a null in it " + executorToSlot);
- }
- this.executorToSlot.putAll(executorToSlot);
- }
- if (resources != null) {
- if (resources.entrySet().stream().anyMatch((entry) -> entry.getKey() == null || entry.getValue() == null)) {
- throw new RuntimeException("Cannot create resources with a null in it " + resources);
- }
- this.resources.putAll(resources);
- }
- if (nodeIdToTotalSharedOffHeap != null) {
- if (nodeIdToTotalSharedOffHeap.entrySet().stream().anyMatch((entry) -> entry.getKey() == null || entry.getValue() == null)) {
- throw new RuntimeException("Cannot create off heap with a null in it " + nodeIdToTotalSharedOffHeap);
- }
- this.nodeIdToTotalSharedOffHeap.putAll(nodeIdToTotalSharedOffHeap);
- }
- }
-
- public SchedulerAssignmentImpl(String topologyId) {
- this(topologyId, null, null, null);
- }
-
- public SchedulerAssignmentImpl(SchedulerAssignment assignment) {
- this(assignment.getTopologyId(), assignment.getExecutorToSlot(),
- assignment.getScheduledResources(), assignment.getNodeIdToTotalSharedOffHeapMemory());
- }
-
- @Override
- public String toString() {
- return this.getClass().getSimpleName() + " topo: " + topologyId + " execToSlots: " + executorToSlot;
- }
-
- public boolean equalsIgnoreResources(Object other) {
- if (other == this) {
- return true;
- }
- if (!(other instanceof SchedulerAssignmentImpl)) {
- return false;
- }
- SchedulerAssignmentImpl o = (SchedulerAssignmentImpl) other;
-
- return this.topologyId.equals(o.topologyId) &&
- this.executorToSlot.equals(o.executorToSlot);
- }
-
- @Override
- public int hashCode() {
- final int prime = 31;
- int result = 1;
- result = prime * result + ((topologyId == null) ? 0 : topologyId.hashCode());
- result = prime * result + ((executorToSlot == null) ? 0 : executorToSlot.hashCode());
- return result;
- }
-
- @Override
- public boolean equals(Object other) {
- if (!equalsIgnoreResources(other)) {
- return false;
- }
- SchedulerAssignmentImpl o = (SchedulerAssignmentImpl) other;
-
- return this.resources.equals(o.resources) &&
- this.nodeIdToTotalSharedOffHeap.equals(o.nodeIdToTotalSharedOffHeap);
- }
-
- @Override
- public Set<WorkerSlot> getSlots() {
- return new HashSet<>(executorToSlot.values());
- }
-
- @Deprecated
- public void assign(WorkerSlot slot, Collection<ExecutorDetails> executors) {
- assign(slot, executors, null);
- }
-
- /**
- * Assign the slot to executors.
- */
- public void assign(WorkerSlot slot, Collection<ExecutorDetails> executors, WorkerResources slotResources) {
- assert(slot != null);
- for (ExecutorDetails executor : executors) {
- this.executorToSlot.put(executor, slot);
- }
- if (slotResources != null) {
- resources.put(slot, slotResources);
- } else {
- resources.remove(slot);
- }
- //Clear the cache scheduling changed
- slotToExecutorsCache = null;
- }
-
- /**
- * Release the slot occupied by this assignment.
- */
- public void unassignBySlot(WorkerSlot slot) {
- //Clear the cache scheduling is going to change
- slotToExecutorsCache = null;
- List<ExecutorDetails> executors = new ArrayList<>();
- for (ExecutorDetails executor : executorToSlot.keySet()) {
- WorkerSlot ws = executorToSlot.get(executor);
- if (ws.equals(slot)) {
- executors.add(executor);
- }
- }
-
- // remove
- for (ExecutorDetails executor : executors) {
- executorToSlot.remove(executor);
- }
-
- resources.remove(slot);
-
- String node = slot.getNodeId();
- boolean isFound = false;
- for (WorkerSlot ws: executorToSlot.values()) {
- if (node.equals(ws.getNodeId())) {
- isFound = true;
- break;
- }
- }
- if (!isFound) {
- nodeIdToTotalSharedOffHeap.remove(node);
- }
- }
-
- /**
- * @param slot
- * @return true if slot is occupied by this assignment
- */
- public boolean isSlotOccupied(WorkerSlot slot) {
- return this.executorToSlot.containsValue(slot);
- }
-
- public boolean isExecutorAssigned(ExecutorDetails executor) {
- return this.executorToSlot.containsKey(executor);
- }
-
- public String getTopologyId() {
- return this.topologyId;
- }
-
- public Map<ExecutorDetails, WorkerSlot> getExecutorToSlot() {
- return this.executorToSlot;
- }
-
- /**
- * @return the executors covered by this assignments
- */
- public Set<ExecutorDetails> getExecutors() {
- return this.executorToSlot.keySet();
- }
-
- public Map<WorkerSlot, Collection<ExecutorDetails>> getSlotToExecutors() {
- Map<WorkerSlot, Collection<ExecutorDetails>> ret = slotToExecutorsCache;
- if (ret != null) {
- return ret;
- }
- ret = new HashMap<>();
- for (Map.Entry<ExecutorDetails, WorkerSlot> entry : executorToSlot.entrySet()) {
- ExecutorDetails exec = entry.getKey();
- WorkerSlot ws = entry.getValue();
- if (!ret.containsKey(ws)) {
- ret.put(ws, new LinkedList<ExecutorDetails>());
- }
- ret.get(ws).add(exec);
- }
- slotToExecutorsCache = ret;
- return ret;
- }
-
- @Override
- public Map<WorkerSlot, WorkerResources> getScheduledResources() {
- return resources;
- }
-
- public void setTotalSharedOffHeapMemory(String node, double value) {
- nodeIdToTotalSharedOffHeap.put(node, value);
- }
-
- @Override
- public Map<String, Double> getNodeIdToTotalSharedOffHeapMemory() {
- return nodeIdToTotalSharedOffHeap;
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/6a4aeb91/storm-client/src/jvm/org/apache/storm/scheduler/SupervisorDetails.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/SupervisorDetails.java b/storm-client/src/jvm/org/apache/storm/scheduler/SupervisorDetails.java
deleted file mode 100644
index 0df8bdd..0000000
--- a/storm-client/src/jvm/org/apache/storm/scheduler/SupervisorDetails.java
+++ /dev/null
@@ -1,147 +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.Collection;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-
-
-import org.apache.storm.Constants;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class SupervisorDetails {
- private static final Logger LOG = LoggerFactory.getLogger(SupervisorDetails.class);
-
- String id;
- /**
- * hostname of this supervisor.
- */
- String host;
- Object meta;
- /**
- * meta data configured for this supervisor.
- */
- Object schedulerMeta;
- /**
- * all the ports of the supervisor.
- */
- Set<Integer> allPorts;
- /**
- * Map containing a manifest of resources for the node the supervisor resides.
- */
- private Map<String, Double> _total_resources;
-
- public SupervisorDetails(String id, String host, Object meta, Object schedulerMeta,
- Collection<? extends Number> allPorts, Map<String, Double> total_resources) {
-
- this.id = id;
- this.host = host;
- this.meta = meta;
- this.schedulerMeta = schedulerMeta;
- if (allPorts != null) {
- setAllPorts(allPorts);
- } else {
- this.allPorts = new HashSet<>();
- }
- this._total_resources = total_resources;
- LOG.debug("Creating a new supervisor ({}-{}) with resources: {}", this.host, this.id, total_resources);
- }
-
- public SupervisorDetails(String id, Object meta){
- this(id, null, meta, null, null, null);
- }
-
- public SupervisorDetails(String id, Object meta, Map<String, Double> total_resources) {
- this(id, null, meta, null, null, total_resources);
- }
-
- public SupervisorDetails(String id, Object meta, Collection<? extends Number> allPorts) {
- this(id, null, meta, null, allPorts, null);
- }
-
- public SupervisorDetails(String id, String host, Object schedulerMeta, Collection<? extends Number> allPorts) {
- this(id, host, null, schedulerMeta, allPorts, null);
- }
-
- public SupervisorDetails(String id, String host, Object schedulerMeta,
- Collection<? extends Number> allPorts, Map<String, Double> total_resources) {
- this(id, host, null, schedulerMeta, allPorts, total_resources);
- }
-
- @Override
- public String toString() {
- return getClass().getSimpleName() + " ID: " + id + " HOST: " + host + " META: " + meta
- + " SCHED_META: " + schedulerMeta + " PORTS: " + allPorts;
- }
-
- private void setAllPorts(Collection<? extends Number> allPorts) {
- this.allPorts = new HashSet<>();
- if (allPorts != null) {
- for (Number n: allPorts) {
- this.allPorts.add(n.intValue());
- }
- }
- }
-
- public String getId() {
- return id;
- }
-
- public String getHost() {
- return host;
- }
-
- public Object getMeta() {
- return meta;
- }
-
- public Set<Integer> getAllPorts() {
- return allPorts;
- }
-
- public Object getSchedulerMeta() {
- return this.schedulerMeta;
- }
-
- private Double getTotalResource(String type) {
- return this._total_resources.get(type);
- }
-
- public double getTotalMemory() {
- Double totalMemory = getTotalResource(Constants.COMMON_TOTAL_MEMORY_RESOURCE_NAME);
- assert totalMemory != null;
- return totalMemory;
- }
-
- public double getTotalCPU() {
- Double totalCPU = getTotalResource(Constants.COMMON_CPU_RESOURCE_NAME);
- assert totalCPU != null;
- return totalCPU;
- }
-
- /**
- * get all resources for this Supervisor.
- */
- public Map<String, Double> getTotalResources() {
- return _total_resources;
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/6a4aeb91/storm-server/pom.xml
----------------------------------------------------------------------
diff --git a/storm-server/pom.xml b/storm-server/pom.xml
index 18fec7f..d6554de 100644
--- a/storm-server/pom.xml
+++ b/storm-server/pom.xml
@@ -130,7 +130,7 @@
<artifactId>maven-checkstyle-plugin</artifactId>
<!--Note - the version would be inherited-->
<configuration>
- <maxAllowedViolations>2655</maxAllowedViolations>
+ <maxAllowedViolations>2630</maxAllowedViolations>
</configuration>
</plugin>
<plugin>
http://git-wip-us.apache.org/repos/asf/storm/blob/6a4aeb91/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
index fc427a8..ea3c9c4 100644
--- a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
+++ b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
@@ -52,6 +52,7 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.function.UnaryOperator;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import java.util.stream.Collectors;
import javax.security.auth.Subject;
import org.apache.storm.Config;
import org.apache.storm.Constants;
@@ -148,6 +149,7 @@ import org.apache.storm.scheduler.TopologyDetails;
import org.apache.storm.scheduler.WorkerSlot;
import org.apache.storm.scheduler.blacklist.BlacklistScheduler;
import org.apache.storm.scheduler.multitenant.MultitenantScheduler;
+import org.apache.storm.scheduler.resource.NormalizedResourceRequest;
import org.apache.storm.scheduler.resource.ResourceAwareScheduler;
import org.apache.storm.scheduler.resource.ResourceUtils;
import org.apache.storm.security.INimbusCredentialPlugin;
@@ -967,13 +969,12 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
return ret;
}
- private static Map<String, Double> setResourcesDefaultIfNotSet(Map<String, Map<String, Double>> compResourcesMap, String compId, Map<String, Object> topoConf) {
- Map<String, Double> resourcesMap = compResourcesMap.get(compId);
- if (resourcesMap == null) {
- resourcesMap = new HashMap<>();
+ private static void setResourcesDefaultIfNotSet(Map<String, NormalizedResourceRequest> compResourcesMap, String compId,
+ Map<String, Object> topoConf) {
+ NormalizedResourceRequest resources = compResourcesMap.get(compId);
+ if (resources == null) {
+ compResourcesMap.put(compId, new NormalizedResourceRequest(topoConf));
}
- ResourceUtils.checkInitialization(resourcesMap, compId, topoConf);
- return resourcesMap;
}
private static void validatePortAvailable(Map<String, Object> conf) throws IOException {
@@ -2563,20 +2564,16 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
private static double getMaxExecutorMemoryUsageForTopo(
StormTopology topology, Map<String, Object> topologyConf) {
double largestMemoryOperator = 0.0;
- for (Map<String, Double> entry :
+ for (NormalizedResourceRequest entry :
ResourceUtils.getBoltsResources(topology, topologyConf).values()) {
- double memoryRequirement =
- entry.getOrDefault(Constants.COMMON_OFFHEAP_MEMORY_RESOURCE_NAME, 0.0)
- + entry.getOrDefault(Constants.COMMON_ONHEAP_MEMORY_RESOURCE_NAME, 0.0);
+ double memoryRequirement = entry.getTotalMemoryMb();
if (memoryRequirement > largestMemoryOperator) {
largestMemoryOperator = memoryRequirement;
}
}
- for (Map<String, Double> entry :
+ for (NormalizedResourceRequest entry :
ResourceUtils.getSpoutsResources(topology, topologyConf).values()) {
- double memoryRequirement =
- entry.getOrDefault(Constants.COMMON_OFFHEAP_MEMORY_RESOURCE_NAME, 0.0)
- + entry.getOrDefault(Constants.COMMON_ONHEAP_MEMORY_RESOURCE_NAME, 0.0);
+ double memoryRequirement = entry.getTotalMemoryMb();
if (memoryRequirement > largestMemoryOperator) {
largestMemoryOperator = memoryRequirement;
}
@@ -3649,16 +3646,18 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
topoPageInfo.set_storm_version(topology.get_storm_version());
}
- Map<String, Map<String, Double>> spoutResources = ResourceUtils.getSpoutsResources(topology, topoConf);
+ Map<String, NormalizedResourceRequest> spoutResources = ResourceUtils.getSpoutsResources(topology, topoConf);
for (Entry<String, ComponentAggregateStats> entry: topoPageInfo.get_id_to_spout_agg_stats().entrySet()) {
CommonAggregateStats commonStats = entry.getValue().get_common_stats();
- commonStats.set_resources_map(setResourcesDefaultIfNotSet(spoutResources, entry.getKey(), topoConf));
+ setResourcesDefaultIfNotSet(spoutResources, entry.getKey(), topoConf);
+ commonStats.set_resources_map(spoutResources.get(entry.getKey()).toNormalizedMap());
}
- Map<String, Map<String, Double>> boltResources = ResourceUtils.getBoltsResources(topology, topoConf);
+ Map<String, NormalizedResourceRequest> boltResources = ResourceUtils.getBoltsResources(topology, topoConf);
for (Entry<String, ComponentAggregateStats> entry: topoPageInfo.get_id_to_bolt_agg_stats().entrySet()) {
CommonAggregateStats commonStats = entry.getValue().get_common_stats();
- commonStats.set_resources_map(setResourcesDefaultIfNotSet(boltResources, entry.getKey(), topoConf));
+ setResourcesDefaultIfNotSet(boltResources, entry.getKey(), topoConf);
+ commonStats.set_resources_map(boltResources.get(entry.getKey()).toNormalizedMap());
}
if (workerSummaries != null) {
@@ -3816,11 +3815,17 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
ComponentPageInfo compPageInfo = StatsUtil.aggCompExecsStats(exec2HostPort, info.taskToComponent, info.beats, window,
includeSys, topoId, topology, componentId);
if (compPageInfo.get_component_type() == ComponentType.SPOUT) {
- compPageInfo.set_resources_map(setResourcesDefaultIfNotSet(
- ResourceUtils.getSpoutsResources(topology, topoConf), componentId, topoConf));
+ NormalizedResourceRequest spoutResources = ResourceUtils.getSpoutResources(topology, topoConf, componentId);
+ if (spoutResources == null) {
+ spoutResources = new NormalizedResourceRequest(topoConf);
+ }
+ compPageInfo.set_resources_map(spoutResources.toNormalizedMap());
} else { //bolt
- compPageInfo.set_resources_map(setResourcesDefaultIfNotSet(
- ResourceUtils.getBoltsResources(topology, topoConf), componentId, topoConf));
+ NormalizedResourceRequest boltResources = ResourceUtils.getBoltResources(topology, topoConf, componentId);
+ if (boltResources == null) {
+ boltResources = new NormalizedResourceRequest(topoConf);
+ }
+ compPageInfo.set_resources_map(boltResources.toNormalizedMap());
}
compPageInfo.set_topology_name(info.topoName);
compPageInfo.set_errors(stormClusterState.errors(topoId, componentId));
http://git-wip-us.apache.org/repos/asf/storm/blob/6a4aeb91/storm-server/src/main/java/org/apache/storm/daemon/supervisor/timer/SupervisorHeartbeat.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/daemon/supervisor/timer/SupervisorHeartbeat.java b/storm-server/src/main/java/org/apache/storm/daemon/supervisor/timer/SupervisorHeartbeat.java
index 26fb729..87e310e 100644
--- a/storm-server/src/main/java/org/apache/storm/daemon/supervisor/timer/SupervisorHeartbeat.java
+++ b/storm-server/src/main/java/org/apache/storm/daemon/supervisor/timer/SupervisorHeartbeat.java
@@ -26,13 +26,12 @@ import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import static org.apache.storm.scheduler.resource.ResourceUtils.normalizedResourceMap;
+import static org.apache.storm.scheduler.resource.NormalizedResources.normalizedResourceMap;
public class SupervisorHeartbeat implements Runnable {
http://git-wip-us.apache.org/repos/asf/storm/blob/6a4aeb91/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java b/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
index 93e4dfa..c311897 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
@@ -19,7 +19,6 @@
package org.apache.storm.scheduler;
import com.google.common.annotations.VisibleForTesting;
-
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -29,13 +28,14 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
-
import org.apache.storm.Config;
import org.apache.storm.Constants;
import org.apache.storm.daemon.nimbus.TopologyResources;
import org.apache.storm.generated.SharedMemory;
import org.apache.storm.generated.WorkerResources;
import org.apache.storm.networktopography.DNSToSwitchMapping;
+import org.apache.storm.scheduler.resource.NormalizedResourceOffer;
+import org.apache.storm.scheduler.resource.NormalizedResourceRequest;
import org.apache.storm.utils.ConfigUtils;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.ReflectionUtils;
@@ -43,7 +43,7 @@ import org.apache.storm.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.apache.storm.scheduler.resource.ResourceUtils.normalizedResourceMap;
+import static org.apache.storm.scheduler.resource.NormalizedResources.normalizedResourceMap;
public class Cluster implements ISchedulingState {
private static final Logger LOG = LoggerFactory.getLogger(Cluster.class);
@@ -79,7 +79,7 @@ public class Cluster implements ISchedulingState {
private Set<String> blackListedHosts = new HashSet<>();
private INimbus inimbus;
private final Topologies topologies;
- private final Map<String, Double> scheduledCPUCache = new HashMap<>();
+ private final Map<String, Double> scheduledCpuCache = new HashMap<>();
private final Map<String, Double> scheduledMemoryCache = new HashMap<>();
public Cluster(
@@ -327,10 +327,10 @@ public class Cluster implements ISchedulingState {
@Override
public Set<Integer> getAssignablePorts(SupervisorDetails supervisor) {
- if (isBlackListed(supervisor.id)) {
+ if (isBlackListed(supervisor.getId())) {
return Collections.emptySet();
}
- return supervisor.allPorts;
+ return supervisor.getAllPorts();
}
@Override
@@ -417,33 +417,18 @@ public class Cluster implements ISchedulingState {
private WorkerResources calculateWorkerResources(
TopologyDetails td, Collection<ExecutorDetails> executors) {
- Map<String, Double> totalResources = new HashMap<>();
+ NormalizedResourceRequest totalResources = new NormalizedResourceRequest();
Map<String, Double> sharedTotalResources = new HashMap<>();
for (ExecutorDetails exec : executors) {
- Map<String, Double> allResources = td.getTotalResources(exec);
+ NormalizedResourceRequest allResources = td.getTotalResources(exec);
if (allResources == null) {
continue;
}
- for (Entry<String, Double> resource : allResources.entrySet()) {
-
- if (!totalResources.containsKey(resource.getKey())) {
- totalResources.put(resource.getKey(), 0.0);
- }
- totalResources.put(
- resource.getKey(),
- totalResources.get(resource.getKey()) + resource.getValue());
- }
+ totalResources.add(allResources);
}
- totalResources = normalizedResourceMap(totalResources);
for (SharedMemory shared : td.getSharedMemoryRequests(executors)) {
- addResource(
- totalResources,
- Constants.COMMON_OFFHEAP_MEMORY_RESOURCE_NAME, shared.get_off_heap_worker()
- );
- addResource(
- totalResources,
- Constants.COMMON_ONHEAP_MEMORY_RESOURCE_NAME, shared.get_on_heap()
- );
+ totalResources.addOffHeap(shared.get_off_heap_worker());
+ totalResources.addOnHeap(shared.get_off_heap_worker());
addResource(
sharedTotalResources,
@@ -456,21 +441,12 @@ public class Cluster implements ISchedulingState {
}
sharedTotalResources = normalizedResourceMap(sharedTotalResources);
WorkerResources ret = new WorkerResources();
- ret.set_resources(totalResources);
+ ret.set_resources(totalResources.toNormalizedMap());
ret.set_shared_resources(sharedTotalResources);
- ret.set_cpu(
- totalResources.getOrDefault(
- Constants.COMMON_CPU_RESOURCE_NAME, 0.0)
- );
- ret.set_mem_off_heap(
- totalResources.getOrDefault(
- Constants.COMMON_OFFHEAP_MEMORY_RESOURCE_NAME, 0.0)
- );
- ret.set_mem_on_heap(
- totalResources.getOrDefault(
- Constants.COMMON_ONHEAP_MEMORY_RESOURCE_NAME, 0.0)
- );
+ ret.set_cpu(totalResources.getTotalCpu());
+ ret.set_mem_off_heap(totalResources.getOffHeapMemoryMb());
+ ret.set_mem_on_heap(totalResources.getOnHeapMemoryMb());
ret.set_shared_mem_off_heap(
sharedTotalResources.getOrDefault(
Constants.COMMON_OFFHEAP_MEMORY_RESOURCE_NAME, 0.0)
@@ -487,32 +463,12 @@ public class Cluster implements ISchedulingState {
WorkerSlot ws,
ExecutorDetails exec,
TopologyDetails td,
- Map<String, Double> resourcesAvailable,
+ NormalizedResourceOffer resourcesAvailable,
double maxHeap) {
- Map<String, Double> requestedResources = td.getTotalResources(exec);
-
- for (Entry resourceNeededEntry : requestedResources.entrySet()) {
- String resourceName = resourceNeededEntry.getKey().toString();
- if (resourceName.equals(Constants.COMMON_OFFHEAP_MEMORY_RESOURCE_NAME) ||
- resourceName.equals(Constants.COMMON_ONHEAP_MEMORY_RESOURCE_NAME)) {
- continue;
- }
- Double resourceNeeded = ObjectReader.getDouble(resourceNeededEntry.getValue());
- Double resourceAvailable = ObjectReader.getDouble(resourcesAvailable.get(resourceName), 0.0);
- if (resourceNeeded > resourceAvailable) {
- if (LOG.isTraceEnabled()) {
- LOG.trace("Could not schedule {}:{} on {} not enough {} {} > {}",
- td.getName(),
- exec,
- ws,
- resourceName,
- resourceNeeded,
- resourceAvailable);
- }
- //Not enough resources - stop trying
- return false;
- }
+ NormalizedResourceRequest requestedResources = td.getTotalResources(exec);
+ if (!resourcesAvailable.couldHoldIgnoringMemory(requestedResources)) {
+ return false;
}
double currentTotal = 0.0;
@@ -539,8 +495,7 @@ public class Cluster implements ISchedulingState {
}
double memoryAdded = afterTotal - currentTotal;
- double memoryAvailable = ObjectReader.getDouble(resourcesAvailable.get(
- Constants.COMMON_TOTAL_MEMORY_RESOURCE_NAME), 0.0);
+ double memoryAvailable = resourcesAvailable.getTotalMemoryMb();
if (memoryAdded > memoryAvailable) {
if (LOG.isTraceEnabled()) {
@@ -613,7 +568,7 @@ public class Cluster implements ISchedulingState {
String nodeId = slot.getNodeId();
assignment.setTotalSharedOffHeapMemory(
nodeId, calculateSharedOffHeapMemory(nodeId, assignment));
- scheduledCPUCache.remove(nodeId);
+ scheduledCpuCache.remove(nodeId);
scheduledMemoryCache.remove(nodeId);
}
@@ -684,7 +639,7 @@ public class Cluster implements ISchedulingState {
String nodeId = slot.getNodeId();
assignment.setTotalSharedOffHeapMemory(
nodeId, calculateSharedOffHeapMemory(nodeId, assignment));
- scheduledCPUCache.remove(nodeId);
+ scheduledCpuCache.remove(nodeId);
scheduledMemoryCache.remove(nodeId);
}
}
@@ -795,7 +750,7 @@ public class Cluster implements ISchedulingState {
public double getClusterTotalCpuResource() {
double sum = 0.0;
for (SupervisorDetails sup : supervisors.values()) {
- sum += sup.getTotalCPU();
+ sum += sup.getTotalCpu();
}
return sum;
}
@@ -935,7 +890,7 @@ public class Cluster implements ISchedulingState {
public Map<String, SupervisorResources> getSupervisorsResourcesMap() {
Map<String, SupervisorResources> ret = new HashMap<>();
for (SupervisorDetails sd : supervisors.values()) {
- ret.put(sd.getId(), new SupervisorResources(sd.getTotalMemory(), sd.getTotalCPU(), 0, 0));
+ ret.put(sd.getId(), new SupervisorResources(sd.getTotalMemory(), sd.getTotalCpu(), 0, 0));
}
for (SchedulerAssignmentImpl assignment : assignments.values()) {
for (Entry<WorkerSlot, WorkerResources> entry :
@@ -986,27 +941,18 @@ public class Cluster implements ISchedulingState {
}
@Override
- public Map<String, Double> getAllScheduledResourcesForNode(String nodeId) {
- Map<String, Double> totalScheduledResources = new HashMap<>();
+ public NormalizedResourceRequest getAllScheduledResourcesForNode(String nodeId) {
+ NormalizedResourceRequest totalScheduledResources = new NormalizedResourceRequest();
for (SchedulerAssignmentImpl assignment : assignments.values()) {
for (Entry<WorkerSlot, WorkerResources> entry :
assignment.getScheduledResources().entrySet()) {
if (nodeId.equals(entry.getKey().getNodeId())) {
- WorkerResources resources = entry.getValue();
- for (Map.Entry<String, Double> resourceEntry : resources.get_resources().entrySet()) {
- Double currentResourceValue = totalScheduledResources.getOrDefault(resourceEntry.getKey(), 0.0);
- totalScheduledResources.put(
- resourceEntry.getKey(),
- currentResourceValue + ObjectReader.getDouble(resourceEntry.getValue()));
- }
-
+ totalScheduledResources.add(entry.getValue());
}
}
Double sharedOffHeap = assignment.getNodeIdToTotalSharedOffHeapMemory().get(nodeId);
if (sharedOffHeap != null) {
- String resourceName = Constants.COMMON_OFFHEAP_MEMORY_RESOURCE_NAME;
- Double currentResourceValue = totalScheduledResources.getOrDefault(resourceName, 0.0);
- totalScheduledResources.put(resourceName, currentResourceValue + sharedOffHeap);
+ totalScheduledResources.addOffHeap(sharedOffHeap);
}
}
return totalScheduledResources;
@@ -1038,7 +984,7 @@ public class Cluster implements ISchedulingState {
@Override
public double getScheduledCpuForNode(String nodeId) {
- Double ret = scheduledCPUCache.get(nodeId);
+ Double ret = scheduledCpuCache.get(nodeId);
if (ret != null) {
return ret;
}
@@ -1052,7 +998,7 @@ public class Cluster implements ISchedulingState {
}
}
}
- scheduledCPUCache.put(nodeId, totalCpu);
+ scheduledCpuCache.put(nodeId, totalCpu);
return totalCpu;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/6a4aeb91/storm-server/src/main/java/org/apache/storm/scheduler/ExecutorDetails.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/ExecutorDetails.java b/storm-server/src/main/java/org/apache/storm/scheduler/ExecutorDetails.java
new file mode 100644
index 0000000..bbbbf3f
--- /dev/null
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/ExecutorDetails.java
@@ -0,0 +1,57 @@
+/*
+ * 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;
+
+public class ExecutorDetails {
+ public final int startTask;
+ public final int endTask;
+
+ public ExecutorDetails(int startTask, int endTask) {
+ this.startTask = startTask;
+ this.endTask = endTask;
+ }
+
+ public int getStartTask() {
+ return startTask;
+ }
+
+ public int getEndTask() {
+ return endTask;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (other == null || !(other instanceof ExecutorDetails)) {
+ return false;
+ }
+
+ ExecutorDetails executor = (ExecutorDetails)other;
+ return (this.startTask == executor.startTask) && (this.endTask == executor.endTask);
+ }
+
+ @Override
+ public int hashCode() {
+ return this.startTask + 13 * this.endTask;
+ }
+
+ @Override
+ public String toString() {
+ return "[" + this.startTask + ", " + this.endTask + "]";
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/6a4aeb91/storm-server/src/main/java/org/apache/storm/scheduler/ISchedulingState.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/ISchedulingState.java b/storm-server/src/main/java/org/apache/storm/scheduler/ISchedulingState.java
index b539b1f..7175b7e 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/ISchedulingState.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/ISchedulingState.java
@@ -25,6 +25,8 @@ import java.util.Set;
import org.apache.storm.daemon.nimbus.TopologyResources;
import org.apache.storm.generated.WorkerResources;
+import org.apache.storm.scheduler.resource.NormalizedResourceOffer;
+import org.apache.storm.scheduler.resource.NormalizedResourceRequest;
/** An interface that provides access to the current scheduling state. */
public interface ISchedulingState {
@@ -179,7 +181,7 @@ public interface ISchedulingState {
WorkerSlot ws,
ExecutorDetails exec,
TopologyDetails td,
- Map<String, Double> resourcesAvailable,
+ NormalizedResourceOffer resourcesAvailable,
double maxHeap);
/** get the current assignment for the topology. */
@@ -205,8 +207,8 @@ public interface ISchedulingState {
/** Get all the supervisors. */
Map<String, SupervisorDetails> getSupervisors();
- /** Get all scheduled resources for node **/
- Map<String, Double> getAllScheduledResourcesForNode(String nodeId);
+ /** Get all scheduled resources for node. **/
+ NormalizedResourceRequest getAllScheduledResourcesForNode(String nodeId);
/** Get the total amount of CPU resources in cluster. */
double getClusterTotalCpuResource();
http://git-wip-us.apache.org/repos/asf/storm/blob/6a4aeb91/storm-server/src/main/java/org/apache/storm/scheduler/SchedulerAssignment.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/SchedulerAssignment.java b/storm-server/src/main/java/org/apache/storm/scheduler/SchedulerAssignment.java
new file mode 100644
index 0000000..c16a67f
--- /dev/null
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/SchedulerAssignment.java
@@ -0,0 +1,85 @@
+/*
+ * 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.Collection;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.storm.generated.Assignment;
+import org.apache.storm.generated.WorkerResources;
+
+public interface SchedulerAssignment {
+ /**
+ * Is this slot part of this assignment or not.
+ * @param slot the slot to check.
+ * @return true if the slot is occupied by this assignment else false.
+ */
+ public boolean isSlotOccupied(WorkerSlot slot);
+
+ /**
+ * Is the executor assigned or not.
+ *
+ * @param executor the executor to check it if is assigned.
+ * @return true if it is assigned else false
+ */
+ public boolean isExecutorAssigned(ExecutorDetails executor);
+
+ /**
+ * Return the ID of the topology.
+ * @return the topology-id this assignment is for.
+ */
+ public String getTopologyId();
+
+ /**
+ * Get the map of executor to WorkerSlot.
+ * @return the executor -> slot map.
+ */
+ public Map<ExecutorDetails, WorkerSlot> getExecutorToSlot();
+
+ /**
+ * Get the set of all executors.
+ * @return the executors covered by this assignments
+ */
+ public Set<ExecutorDetails> getExecutors();
+
+ /**
+ * Get the set of all slots that are a part of this.
+ * @return the set of all slots.
+ */
+ public Set<WorkerSlot> getSlots();
+
+ /**
+ * Get the mapping of slot to executors on that slot.
+ * @return the slot to the executors assigned to that slot.
+ */
+ public Map<WorkerSlot, Collection<ExecutorDetails>> getSlotToExecutors();
+
+ /**
+ * Get the slot to resource mapping.
+ * @return The slot to resource mapping
+ */
+ public Map<WorkerSlot, WorkerResources> getScheduledResources();
+
+ /**
+ * Get the total shared off heap memory mapping.
+ * @return host to total shared off heap memory mapping.
+ */
+ public Map<String, Double> getNodeIdToTotalSharedOffHeapMemory();
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/6a4aeb91/storm-server/src/main/java/org/apache/storm/scheduler/SchedulerAssignmentImpl.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/SchedulerAssignmentImpl.java b/storm-server/src/main/java/org/apache/storm/scheduler/SchedulerAssignmentImpl.java
new file mode 100644
index 0000000..f92c3d1
--- /dev/null
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/SchedulerAssignmentImpl.java
@@ -0,0 +1,251 @@
+/*
+ * 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.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.storm.generated.WorkerResources;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SchedulerAssignmentImpl implements SchedulerAssignment {
+ private static final Logger LOG = LoggerFactory.getLogger(SchedulerAssignmentImpl.class);
+
+ /**
+ * topology-id this assignment is for.
+ */
+ private final String topologyId;
+
+ /**
+ * assignment detail, a mapping from executor to <code>WorkerSlot</code>.
+ */
+ private final Map<ExecutorDetails, WorkerSlot> executorToSlot = new HashMap<>();
+ private final Map<WorkerSlot, WorkerResources> resources = new HashMap<>();
+ private final Map<String, Double> nodeIdToTotalSharedOffHeap = new HashMap<>();
+ //Used to cache the slotToExecutors mapping.
+ private Map<WorkerSlot, Collection<ExecutorDetails>> slotToExecutorsCache = null;
+
+ /**
+ * Create a new assignment.
+ * @param topologyId the id of the topology the assignment is for.
+ * @param executorToSlot the executor to slot mapping for the assignment. Can be null and set through other methods later.
+ * @param resources the resources for the current assignments. Can be null and set through other methods later.
+ * @param nodeIdToTotalSharedOffHeap the shared memory for this assignment can be null and set through other methods later.
+ */
+ public SchedulerAssignmentImpl(String topologyId, Map<ExecutorDetails, WorkerSlot> executorToSlot,
+ Map<WorkerSlot, WorkerResources> resources, Map<String, Double> nodeIdToTotalSharedOffHeap) {
+ this.topologyId = topologyId;
+ if (executorToSlot != null) {
+ if (executorToSlot.entrySet().stream().anyMatch((entry) -> entry.getKey() == null || entry.getValue() == null)) {
+ throw new RuntimeException("Cannot create a scheduling with a null in it " + executorToSlot);
+ }
+ this.executorToSlot.putAll(executorToSlot);
+ }
+ if (resources != null) {
+ if (resources.entrySet().stream().anyMatch((entry) -> entry.getKey() == null || entry.getValue() == null)) {
+ throw new RuntimeException("Cannot create resources with a null in it " + resources);
+ }
+ this.resources.putAll(resources);
+ }
+ if (nodeIdToTotalSharedOffHeap != null) {
+ if (nodeIdToTotalSharedOffHeap.entrySet().stream().anyMatch((entry) -> entry.getKey() == null || entry.getValue() == null)) {
+ throw new RuntimeException("Cannot create off heap with a null in it " + nodeIdToTotalSharedOffHeap);
+ }
+ this.nodeIdToTotalSharedOffHeap.putAll(nodeIdToTotalSharedOffHeap);
+ }
+ }
+
+ public SchedulerAssignmentImpl(String topologyId) {
+ this(topologyId, null, null, null);
+ }
+
+ public SchedulerAssignmentImpl(SchedulerAssignment assignment) {
+ this(assignment.getTopologyId(), assignment.getExecutorToSlot(),
+ assignment.getScheduledResources(), assignment.getNodeIdToTotalSharedOffHeapMemory());
+ }
+
+ @Override
+ public String toString() {
+ return this.getClass().getSimpleName() + " topo: " + topologyId + " execToSlots: " + executorToSlot;
+ }
+
+ /**
+ * Like the equals command, but ignores the resources.
+ * @param other the object to check for equality against.
+ * @return true if they are equal, ignoring resources, else false.
+ */
+ public boolean equalsIgnoreResources(Object other) {
+ if (other == this) {
+ return true;
+ }
+ if (!(other instanceof SchedulerAssignmentImpl)) {
+ return false;
+ }
+ SchedulerAssignmentImpl o = (SchedulerAssignmentImpl) other;
+
+ return topologyId.equals(o.topologyId)
+ && executorToSlot.equals(o.executorToSlot);
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + ((topologyId == null) ? 0 : topologyId.hashCode());
+ result = prime * result + ((executorToSlot == null) ? 0 : executorToSlot.hashCode());
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (!equalsIgnoreResources(other)) {
+ return false;
+ }
+ SchedulerAssignmentImpl o = (SchedulerAssignmentImpl) other;
+
+ return resources.equals(o.resources)
+ && nodeIdToTotalSharedOffHeap.equals(o.nodeIdToTotalSharedOffHeap);
+ }
+
+ @Override
+ public Set<WorkerSlot> getSlots() {
+ return new HashSet<>(executorToSlot.values());
+ }
+
+ @Deprecated
+ public void assign(WorkerSlot slot, Collection<ExecutorDetails> executors) {
+ assign(slot, executors, null);
+ }
+
+ /**
+ * Assign the slot to executors.
+ */
+ public void assign(WorkerSlot slot, Collection<ExecutorDetails> executors, WorkerResources slotResources) {
+ assert slot != null;
+ for (ExecutorDetails executor : executors) {
+ this.executorToSlot.put(executor, slot);
+ }
+ if (slotResources != null) {
+ resources.put(slot, slotResources);
+ } else {
+ resources.remove(slot);
+ }
+ //Clear the cache scheduling changed
+ slotToExecutorsCache = null;
+ }
+
+ /**
+ * Release the slot occupied by this assignment.
+ */
+ public void unassignBySlot(WorkerSlot slot) {
+ //Clear the cache scheduling is going to change
+ slotToExecutorsCache = null;
+ List<ExecutorDetails> executors = new ArrayList<>();
+ for (ExecutorDetails executor : executorToSlot.keySet()) {
+ WorkerSlot ws = executorToSlot.get(executor);
+ if (ws.equals(slot)) {
+ executors.add(executor);
+ }
+ }
+
+ // remove
+ for (ExecutorDetails executor : executors) {
+ executorToSlot.remove(executor);
+ }
+
+ resources.remove(slot);
+
+ String node = slot.getNodeId();
+ boolean isFound = false;
+ for (WorkerSlot ws: executorToSlot.values()) {
+ if (node.equals(ws.getNodeId())) {
+ isFound = true;
+ break;
+ }
+ }
+ if (!isFound) {
+ nodeIdToTotalSharedOffHeap.remove(node);
+ }
+ }
+
+ @Override
+ public boolean isSlotOccupied(WorkerSlot slot) {
+ return this.executorToSlot.containsValue(slot);
+ }
+
+ @Override
+ public boolean isExecutorAssigned(ExecutorDetails executor) {
+ return this.executorToSlot.containsKey(executor);
+ }
+
+ @Override
+ public String getTopologyId() {
+ return this.topologyId;
+ }
+
+ @Override
+ public Map<ExecutorDetails, WorkerSlot> getExecutorToSlot() {
+ return this.executorToSlot;
+ }
+
+ @Override
+ public Set<ExecutorDetails> getExecutors() {
+ return this.executorToSlot.keySet();
+ }
+
+ @Override
+ public Map<WorkerSlot, Collection<ExecutorDetails>> getSlotToExecutors() {
+ Map<WorkerSlot, Collection<ExecutorDetails>> ret = slotToExecutorsCache;
+ if (ret != null) {
+ return ret;
+ }
+ ret = new HashMap<>();
+ for (Map.Entry<ExecutorDetails, WorkerSlot> entry : executorToSlot.entrySet()) {
+ ExecutorDetails exec = entry.getKey();
+ WorkerSlot ws = entry.getValue();
+ if (!ret.containsKey(ws)) {
+ ret.put(ws, new LinkedList<ExecutorDetails>());
+ }
+ ret.get(ws).add(exec);
+ }
+ slotToExecutorsCache = ret;
+ return ret;
+ }
+
+ @Override
+ public Map<WorkerSlot, WorkerResources> getScheduledResources() {
+ return resources;
+ }
+
+ public void setTotalSharedOffHeapMemory(String node, double value) {
+ nodeIdToTotalSharedOffHeap.put(node, value);
+ }
+
+ @Override
+ public Map<String, Double> getNodeIdToTotalSharedOffHeapMemory() {
+ return nodeIdToTotalSharedOffHeap;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/6a4aeb91/storm-server/src/main/java/org/apache/storm/scheduler/SupervisorDetails.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/SupervisorDetails.java b/storm-server/src/main/java/org/apache/storm/scheduler/SupervisorDetails.java
new file mode 100644
index 0000000..3d08715
--- /dev/null
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/SupervisorDetails.java
@@ -0,0 +1,153 @@
+/**
+ * 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.Collection;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import org.apache.storm.Constants;
+import org.apache.storm.scheduler.resource.NormalizedResourceOffer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SupervisorDetails {
+ private static final Logger LOG = LoggerFactory.getLogger(SupervisorDetails.class);
+
+ private final String id;
+ /**
+ * hostname of this supervisor.
+ */
+ private final String host;
+ private final Object meta;
+ /**
+ * meta data configured for this supervisor.
+ */
+ private final Object schedulerMeta;
+ /**
+ * all the ports of the supervisor.
+ */
+ private Set<Integer> allPorts;
+ /**
+ * Map containing a manifest of resources for the node the supervisor resides.
+ */
+ private final NormalizedResourceOffer totalResources;
+
+ /**
+ * Create the details of a new supervisor.
+ * @param id the ID as reported by the supervisor.
+ * @param host the host the supervisor is on.
+ * @param meta meta data reported by the supervisor (should be a collection of the ports on the supervisor).
+ * @param schedulerMeta Not used and can probably be removed.
+ * @param allPorts all of the ports for the supervisor (a better version of meta)
+ * @param totalResources all of the resources for this supervisor.
+ */
+ public SupervisorDetails(String id, String host, Object meta, Object schedulerMeta,
+ Collection<? extends Number> allPorts, Map<String, Double> totalResources) {
+
+ this.id = id;
+ this.host = host;
+ this.meta = meta;
+ this.schedulerMeta = schedulerMeta;
+ if (allPorts != null) {
+ setAllPorts(allPorts);
+ } else {
+ this.allPorts = new HashSet<>();
+ }
+ this.totalResources = new NormalizedResourceOffer(totalResources);
+ LOG.debug("Creating a new supervisor ({}-{}) with resources: {}", this.host, this.id, totalResources);
+ }
+
+ public SupervisorDetails(String id, Object meta) {
+ this(id, null, meta, null, null, null);
+ }
+
+ public SupervisorDetails(String id, Object meta, Map<String, Double> totalResources) {
+ this(id, null, meta, null, null, totalResources);
+ }
+
+ public SupervisorDetails(String id, Object meta, Collection<? extends Number> allPorts) {
+ this(id, null, meta, null, allPorts, null);
+ }
+
+ public SupervisorDetails(String id, String host, Object schedulerMeta, Collection<? extends Number> allPorts) {
+ this(id, host, null, schedulerMeta, allPorts, null);
+ }
+
+ public SupervisorDetails(String id, String host, Object schedulerMeta,
+ Collection<? extends Number> allPorts, Map<String, Double> totalResources) {
+ this(id, host, null, schedulerMeta, allPorts, totalResources);
+ }
+
+ @Override
+ public String toString() {
+ return getClass().getSimpleName() + " ID: " + id + " HOST: " + host + " META: " + meta
+ + " SCHED_META: " + schedulerMeta + " PORTS: " + allPorts;
+ }
+
+ private void setAllPorts(Collection<? extends Number> allPorts) {
+ this.allPorts = new HashSet<>();
+ if (allPorts != null) {
+ for (Number n: allPorts) {
+ this.allPorts.add(n.intValue());
+ }
+ }
+ }
+
+ public String getId() {
+ return id;
+ }
+
+ public String getHost() {
+ return host;
+ }
+
+ public Object getMeta() {
+ return meta;
+ }
+
+ public Set<Integer> getAllPorts() {
+ return allPorts;
+ }
+
+ public Object getSchedulerMeta() {
+ return this.schedulerMeta;
+ }
+
+ /**
+ * Get the total Memory on this supervisor in MB.
+ */
+ public double getTotalMemory() {
+ return totalResources.getTotalMemoryMb();
+ }
+
+ /**
+ * Get the total CPU on this supervisor in % CPU.
+ */
+ public double getTotalCpu() {
+ return totalResources.getTotalCpu();
+ }
+
+ /**
+ * Get all resources for this Supervisor.
+ */
+ public NormalizedResourceOffer getTotalResources() {
+ return totalResources;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/6a4aeb91/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java b/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java
index d92c80a..8e4c1d5 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java
@@ -35,15 +35,13 @@ import org.apache.storm.generated.GlobalStreamId;
import org.apache.storm.generated.SharedMemory;
import org.apache.storm.generated.SpoutSpec;
import org.apache.storm.generated.StormTopology;
-import org.apache.storm.scheduler.resource.ResourceUtils;
+import org.apache.storm.scheduler.resource.NormalizedResourceRequest;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
import org.apache.storm.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.apache.storm.scheduler.resource.ResourceUtils.normalizedResourceMap;
-
public class TopologyDetails {
private final String topologyId;
private final Map<String, Object> topologyConf;
@@ -51,7 +49,7 @@ public class TopologyDetails {
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;
+ private Map<ExecutorDetails, NormalizedResourceRequest> resourceList;
//Max heap size for a worker used by topology
private Double topologyWorkerMaxHeapSize;
//topology priority
@@ -138,12 +136,10 @@ public class TopologyDetails {
if (topology.get_bolts() != null) {
for (Map.Entry<String, Bolt> bolt : topology.get_bolts().entrySet()) {
//the json_conf is populated by TopologyBuilder (e.g. boltDeclarer.setMemoryLoad)
- Map<String, Double> topologyResources =
- ResourceUtils.parseResources(bolt.getValue().get_common().get_json_conf());
- ResourceUtils.checkInitialization(topologyResources, bolt.getKey(), this.topologyConf);
+ NormalizedResourceRequest topologyResources = new NormalizedResourceRequest(bolt.getValue().get_common(), topologyConf);
for (Map.Entry<ExecutorDetails, String> anExecutorToComponent :
executorToComponent.entrySet()) {
- if (bolt.getKey().equals(anExecutorToComponent.getValue()) && !topologyResources.isEmpty()) {
+ if (bolt.getKey().equals(anExecutorToComponent.getValue())) {
resourceList.put(anExecutorToComponent.getKey(), topologyResources);
}
}
@@ -152,12 +148,10 @@ public class TopologyDetails {
// Extract spout resource info
if (topology.get_spouts() != null) {
for (Map.Entry<String, SpoutSpec> spout : topology.get_spouts().entrySet()) {
- Map<String, Double> topologyResources =
- ResourceUtils.parseResources(spout.getValue().get_common().get_json_conf());
- ResourceUtils.checkInitialization(topologyResources, spout.getKey(), this.topologyConf);
+ NormalizedResourceRequest topologyResources = new NormalizedResourceRequest(spout.getValue().get_common(), topologyConf);
for (Map.Entry<ExecutorDetails, String> anExecutorToComponent :
executorToComponent.entrySet()) {
- if (spout.getKey().equals(anExecutorToComponent.getValue()) && !topologyResources.isEmpty()) {
+ if (spout.getKey().equals(anExecutorToComponent.getValue())) {
resourceList.put(anExecutorToComponent.getKey(), topologyResources);
}
}
@@ -261,8 +255,7 @@ public class TopologyDetails {
Double ret = null;
if (hasExecInTopo(exec)) {
ret = resourceList
- .get(exec)
- .get(Constants.COMMON_ONHEAP_MEMORY_RESOURCE_NAME);
+ .get(exec).getOnHeapMemoryMb();;
}
return ret;
}
@@ -276,8 +269,7 @@ public class TopologyDetails {
Double ret = null;
if (hasExecInTopo(exec)) {
ret = resourceList
- .get(exec)
- .get(Constants.COMMON_OFFHEAP_MEMORY_RESOURCE_NAME);
+ .get(exec).getOffHeapMemoryMb();
}
return ret;
}
@@ -335,7 +327,7 @@ public class TopologyDetails {
* @param exec the executor to get the resources for.
* @return Double the total about of cpu requirement for executor
*/
- public Map<String, Double> getTotalResources(ExecutorDetails exec) {
+ public NormalizedResourceRequest getTotalResources(ExecutorDetails exec) {
if (hasExecInTopo(exec)) {
return this.resourceList.get(exec);
}
@@ -350,8 +342,7 @@ public class TopologyDetails {
public Double getTotalCpuReqTask(ExecutorDetails exec) {
if (hasExecInTopo(exec)) {
return resourceList
- .get(exec)
- .get(Constants.COMMON_CPU_RESOURCE_NAME);
+ .get(exec).getTotalCpu();
}
return null;
}
@@ -440,7 +431,7 @@ public class TopologyDetails {
* @param exec
* @return a map containing the resource requirements for this exec
*/
- public Map<String, Double> getTaskResourceReqList(ExecutorDetails exec) {
+ public NormalizedResourceRequest getTaskResourceReqList(ExecutorDetails exec) {
if (hasExecInTopo(exec)) {
return resourceList.get(exec);
}
@@ -458,7 +449,7 @@ public class TopologyDetails {
/**
* add resource requirements for a executor.
*/
- public void addResourcesForExec(ExecutorDetails exec, Map<String, Double> resourceList) {
+ public void addResourcesForExec(ExecutorDetails exec, NormalizedResourceRequest resourceList) {
if (hasExecInTopo(exec)) {
LOG.warn("Executor {} already exists...ResourceList: {}", exec, getTaskResourceReqList(exec));
return;
@@ -470,49 +461,7 @@ public class TopologyDetails {
* Add default resource requirements for a executor.
*/
private void addDefaultResforExec(ExecutorDetails exec) {
- Double topologyComponentCpuPcorePercent =
- ObjectReader.getDouble(
- topologyConf.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT), null);
- Double topologyComponentResourcesOffheapMemoryMb =
- ObjectReader.getDouble(
- topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB), null);
- Double topologyComponentResourcesOnheapMemoryMb =
- ObjectReader.getDouble(
- 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);
-
- adjustResourcesForExec(exec, defaultResourceList);
-
- Map<String,Double> topologyComponentResourcesMap = (
- Map<String, Double>) this.topologyConf.getOrDefault(
- Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, new HashMap<>());
-
- topologyComponentResourcesMap = normalizedResourceMap(topologyComponentResourcesMap);
-
- LOG.info("Scheduling Executor: {} with resource requirement as {}",
- exec, topologyComponentResourcesMap);
- LOG.debug(
- "Scheduling Executor: {} {} with memory requirement as onHeap: {} - offHeap: {} "
- + "and CPU requirement: {}",
- getExecutorToComponent().get(exec),
- exec,
- topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB),
- topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB),
- topologyConf.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT));
- addResourcesForExec(exec, normalizedResourceMap(defaultResourceList));
+ addResourcesForExec(exec, new NormalizedResourceRequest(topologyConf));
}
/**
http://git-wip-us.apache.org/repos/asf/storm/blob/6a4aeb91/storm-server/src/main/java/org/apache/storm/scheduler/resource/NormalizedResourceOffer.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/NormalizedResourceOffer.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/NormalizedResourceOffer.java
new file mode 100644
index 0000000..2623655
--- /dev/null
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/resource/NormalizedResourceOffer.java
@@ -0,0 +1,89 @@
+/*
+ * 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.resource;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.storm.Constants;
+import org.apache.storm.generated.WorkerResources;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * An offer of resources that has been normalized.
+ */
+public class NormalizedResourceOffer extends NormalizedResources {
+ private static final Logger LOG = LoggerFactory.getLogger(NormalizedResourceOffer.class);
+ private double totalMemory;
+
+ /**
+ * Create a new normalized set of resources. Note that memory is not covered here becasue it is not consistent in requests vs offers
+ * because of how on heap vs off heap is used.
+ *
+ * @param resources the resources to be normalized.
+ */
+ public NormalizedResourceOffer(Map<String, ? extends Number> resources) {
+ super(resources, null);
+ }
+
+ public NormalizedResourceOffer() {
+ super(null, null);
+ }
+
+ public NormalizedResourceOffer(NormalizedResourceOffer other) {
+ super(other);
+ this.totalMemory = other.totalMemory;
+ }
+
+ @Override
+ protected void initializeMemory(Map<String, Double> normalizedResources) {
+ totalMemory = normalizedResources.getOrDefault(Constants.COMMON_TOTAL_MEMORY_RESOURCE_NAME, 0.0);
+ }
+
+ @Override
+ public double getTotalMemoryMb() {
+ return totalMemory;
+ }
+
+ @Override
+ public String toString() {
+ return super.toString() + " MEM: " + totalMemory;
+ }
+
+ @Override
+ public Map<String,Double> toNormalizedMap() {
+ Map<String, Double> ret = super.toNormalizedMap();
+ ret.put(Constants.COMMON_TOTAL_MEMORY_RESOURCE_NAME, totalMemory);
+ return ret;
+ }
+
+ @Override
+ public void add(NormalizedResources other) {
+ super.add(other);
+ totalMemory += other.getTotalMemoryMb();
+ }
+
+ @Override
+ public void remove(NormalizedResources other) {
+ super.remove(other);
+ totalMemory -= other.getTotalMemoryMb();
+ assert totalMemory >= 0.0;
+ }
+}