Repository: storm Updated Branches: refs/heads/master 4d15d4c38 -> 762ca287a
[STORM-1232] port backtype.storm.scheduler.DefaultScheduler to java [STORM-1231] port backtype.storm.scheduler.EvenScheduler to java Project: http://git-wip-us.apache.org/repos/asf/storm/repo Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/4c41ac18 Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/4c41ac18 Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/4c41ac18 Branch: refs/heads/master Commit: 4c41ac18be1129d350bc0f6d6d72eea84570c776 Parents: 12ceb09 Author: vesense <[email protected]> Authored: Mon Feb 15 17:00:05 2016 +0800 Committer: vesense <[email protected]> Committed: Mon Feb 15 18:17:13 2016 +0800 ---------------------------------------------------------------------- .../src/clj/org/apache/storm/daemon/nimbus.clj | 2 +- .../apache/storm/scheduler/DefaultScheduler.clj | 80 --------- .../apache/storm/scheduler/EvenScheduler.clj | 98 ---------- .../storm/scheduler/IsolationScheduler.clj | 4 +- .../storm/scheduler/DefaultScheduler.java | 110 ++++++++++++ .../apache/storm/scheduler/EvenScheduler.java | 177 +++++++++++++++++++ .../src/jvm/org/apache/storm/utils/Utils.java | 19 ++ .../clj/org/apache/storm/scheduler_test.clj | 6 +- 8 files changed, 312 insertions(+), 184 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/storm/blob/4c41ac18/storm-core/src/clj/org/apache/storm/daemon/nimbus.clj ---------------------------------------------------------------------- diff --git a/storm-core/src/clj/org/apache/storm/daemon/nimbus.clj b/storm-core/src/clj/org/apache/storm/daemon/nimbus.clj index 710cd83..250c861 100644 --- a/storm-core/src/clj/org/apache/storm/daemon/nimbus.clj +++ b/storm-core/src/clj/org/apache/storm/daemon/nimbus.clj @@ -35,7 +35,7 @@ (:import [java.nio.channels Channels WritableByteChannel]) (:import [org.apache.storm.security.auth ThriftServer ThriftConnectionType ReqContext AuthUtils] [org.apache.storm.logging ThriftAccessLogger]) - (:use [org.apache.storm.scheduler.DefaultScheduler]) + (:import [org.apache.storm.scheduler DefaultScheduler]) (:import [org.apache.storm.scheduler INimbus SupervisorDetails WorkerSlot TopologyDetails Cluster Topologies SchedulerAssignment SchedulerAssignmentImpl DefaultScheduler ExecutorDetails]) (:import [org.apache.storm.nimbus NimbusInfo]) http://git-wip-us.apache.org/repos/asf/storm/blob/4c41ac18/storm-core/src/clj/org/apache/storm/scheduler/DefaultScheduler.clj ---------------------------------------------------------------------- diff --git a/storm-core/src/clj/org/apache/storm/scheduler/DefaultScheduler.clj b/storm-core/src/clj/org/apache/storm/scheduler/DefaultScheduler.clj deleted file mode 100644 index 71b507e..0000000 --- a/storm-core/src/clj/org/apache/storm/scheduler/DefaultScheduler.clj +++ /dev/null @@ -1,80 +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. -(ns org.apache.storm.scheduler.DefaultScheduler - (:use [org.apache.storm util config]) - (:require [org.apache.storm.scheduler.EvenScheduler :as EvenScheduler]) - (:import [org.apache.storm.scheduler IScheduler Topologies - Cluster TopologyDetails WorkerSlot SchedulerAssignment - EvenScheduler ExecutorDetails] - [org.apache.storm.utils Utils]) - (:gen-class - :implements [org.apache.storm.scheduler.IScheduler])) - -(defn- bad-slots [existing-slots num-executors num-workers] - (if (= 0 num-workers) - '() - (let [distribution (->> (Utils/integerDivided num-executors num-workers) - clojurify-structure - atom) - keepers (atom {})] - (doseq [[node+port executor-list] existing-slots :let [executor-count (count executor-list)]] - (when (pos? (get @distribution executor-count 0)) - (swap! keepers assoc node+port executor-list) - (swap! distribution update-in [executor-count] dec) - )) - (->> @keepers - keys - (apply dissoc existing-slots) - keys - (map (fn [[node port]] - (WorkerSlot. node port))))))) - -(defn slots-can-reassign [^Cluster cluster slots] - (->> slots - (filter - (fn [[node port]] - (if-not (.isBlackListed cluster node) - (if-let [supervisor (.getSupervisorById cluster node)] - (.contains (.getAllPorts supervisor) (int port)) - )))))) - -(defn -prepare [this conf] - ) - -(defn default-schedule [^Topologies topologies ^Cluster cluster] - (let [needs-scheduling-topologies (.needsSchedulingTopologies cluster topologies)] - (doseq [^TopologyDetails topology needs-scheduling-topologies - :let [topology-id (.getId topology) - available-slots (->> (.getAvailableSlots cluster) - (map #(vector (.getNodeId %) (.getPort %)))) - all-executors (->> topology - .getExecutors - (map #(vector (.getStartTask %) (.getEndTask %))) - set) - alive-assigned (EvenScheduler/get-alive-assigned-node+port->executors cluster topology-id) - alive-executors (->> alive-assigned vals (apply concat) set) - can-reassign-slots (slots-can-reassign cluster (keys alive-assigned)) - total-slots-to-use (min (.getNumWorkers topology) - (+ (count can-reassign-slots) (count available-slots))) - bad-slots (if (or (> total-slots-to-use (count alive-assigned)) - (not= alive-executors all-executors)) - (bad-slots alive-assigned (count all-executors) total-slots-to-use) - [])]] - (.freeSlots cluster bad-slots) - (EvenScheduler/schedule-topologies-evenly (Topologies. {topology-id topology}) cluster)))) - -(defn -schedule [this ^Topologies topologies ^Cluster cluster] - (default-schedule topologies cluster)) http://git-wip-us.apache.org/repos/asf/storm/blob/4c41ac18/storm-core/src/clj/org/apache/storm/scheduler/EvenScheduler.clj ---------------------------------------------------------------------- diff --git a/storm-core/src/clj/org/apache/storm/scheduler/EvenScheduler.clj b/storm-core/src/clj/org/apache/storm/scheduler/EvenScheduler.clj deleted file mode 100644 index fce535f..0000000 --- a/storm-core/src/clj/org/apache/storm/scheduler/EvenScheduler.clj +++ /dev/null @@ -1,98 +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. -(ns org.apache.storm.scheduler.EvenScheduler - (:use [org.apache.storm util log config]) - (:require [clojure.set :as set]) - (:import [org.apache.storm.scheduler IScheduler Topologies - Cluster TopologyDetails WorkerSlot ExecutorDetails] - [org.apache.storm.utils Utils]) - (:gen-class - :implements [org.apache.storm.scheduler.IScheduler])) - -; this can be rewritten to be tail recursive -(defn- interleave-all - [& colls] - (if (empty? colls) - [] - (let [colls (filter (complement empty?) colls) - my-elems (map first colls) - rest-elems (apply interleave-all (map rest colls))] - (concat my-elems rest-elems)))) - -(defn sort-slots [all-slots] - (let [split-up (sort-by count > (vals (group-by first all-slots)))] - (apply interleave-all split-up) - )) - -(defn get-alive-assigned-node+port->executors [cluster topology-id] - (let [existing-assignment (.getAssignmentById cluster topology-id) - executor->slot (if existing-assignment - (.getExecutorToSlot existing-assignment) - {}) - executor->node+port (into {} (for [[^ExecutorDetails executor ^WorkerSlot slot] executor->slot - :let [executor [(.getStartTask executor) (.getEndTask executor)] - node+port [(.getNodeId slot) (.getPort slot)]]] - {executor node+port})) - alive-assigned (clojurify-structure (Utils/reverseMap executor->node+port))] - alive-assigned)) - -(defn- repeat-seq - ([aseq] - (apply concat (repeat aseq))) - ([amt aseq] - (apply concat (repeat amt aseq)))) - -(defn- schedule-topology [^TopologyDetails topology ^Cluster cluster] - (let [topology-id (.getId topology) - available-slots (->> (.getAvailableSlots cluster) - (map #(vector (.getNodeId %) (.getPort %)))) - all-executors (->> topology - .getExecutors - (map #(vector (.getStartTask %) (.getEndTask %))) - set) - alive-assigned (get-alive-assigned-node+port->executors cluster topology-id) - total-slots-to-use (min (.getNumWorkers topology) - (+ (count available-slots) (count alive-assigned))) - reassign-slots (take (- total-slots-to-use (count alive-assigned)) - (sort-slots available-slots)) - reassign-executors (sort (set/difference all-executors (set (apply concat (vals alive-assigned))))) - reassignment (into {} - (map vector - reassign-executors - ;; for some reason it goes into infinite loop without limiting the repeat-seq - (repeat-seq (count reassign-executors) reassign-slots)))] - (when-not (empty? reassignment) - (log-message "Available slots: " (pr-str available-slots)) - ) - reassignment)) - -(defn schedule-topologies-evenly [^Topologies topologies ^Cluster cluster] - (let [needs-scheduling-topologies (.needsSchedulingTopologies cluster topologies)] - (doseq [^TopologyDetails topology needs-scheduling-topologies - :let [topology-id (.getId topology) - new-assignment (schedule-topology topology cluster) - node+port->executors (clojurify-structure (Utils/reverseMap new-assignment))]] - (doseq [[node+port executors] node+port->executors - :let [^WorkerSlot slot (WorkerSlot. (first node+port) (last node+port)) - executors (for [[start-task end-task] executors] - (ExecutorDetails. start-task end-task))]] - (.assign cluster slot topology-id executors))))) - -(defn -prepare [this conf] - ) - -(defn -schedule [this ^Topologies topologies ^Cluster cluster] - (schedule-topologies-evenly topologies cluster)) http://git-wip-us.apache.org/repos/asf/storm/blob/4c41ac18/storm-core/src/clj/org/apache/storm/scheduler/IsolationScheduler.clj ---------------------------------------------------------------------- diff --git a/storm-core/src/clj/org/apache/storm/scheduler/IsolationScheduler.clj b/storm-core/src/clj/org/apache/storm/scheduler/IsolationScheduler.clj index 151fcbb..0446f22 100644 --- a/storm-core/src/clj/org/apache/storm/scheduler/IsolationScheduler.clj +++ b/storm-core/src/clj/org/apache/storm/scheduler/IsolationScheduler.clj @@ -15,7 +15,7 @@ ;; limitations under the License. (ns org.apache.storm.scheduler.IsolationScheduler (:use [org.apache.storm util config log]) - (:require [org.apache.storm.scheduler.DefaultScheduler :as DefaultScheduler]) + (:import [org.apache.storm.scheduler DefaultScheduler]) (:import [java.util HashSet Set List LinkedList ArrayList Map HashMap] [org.apache.storm.utils]) (:import [org.apache.storm.utils Utils Container]) @@ -219,7 +219,7 @@ (-<> topology-worker-specs allocated-topologies (leftover-topologies topologies <>) - (DefaultScheduler/default-schedule <> cluster)) + (DefaultScheduler/defaultSchedule <> cluster)) (do (log-warn "Unable to isolate topologies " (pr-str failed-iso-topologies) ". No machine had enough worker slots to run the remaining workers for these topologies. Clearing all other resources and will wait for enough resources for isolated topologies before allocating any other resources.") ;; clear workers off all hosts that are not blacklisted http://git-wip-us.apache.org/repos/asf/storm/blob/4c41ac18/storm-core/src/jvm/org/apache/storm/scheduler/DefaultScheduler.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/org/apache/storm/scheduler/DefaultScheduler.java b/storm-core/src/jvm/org/apache/storm/scheduler/DefaultScheduler.java new file mode 100644 index 0000000..e9cd180 --- /dev/null +++ b/storm-core/src/jvm/org/apache/storm/scheduler/DefaultScheduler.java @@ -0,0 +1,110 @@ +/** + * 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.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Set; + +import org.apache.storm.utils.Utils; + +public class DefaultScheduler implements IScheduler { + + private static Set<WorkerSlot> badSlots(Map<WorkerSlot, List<ExecutorDetails>> existingSlots, int numExecutors, int numWorkers) { + if (numWorkers != 0) { + Map<Integer, Integer> distribution = Utils.integerDivided(numExecutors, numWorkers); + Set<WorkerSlot> _slots = new HashSet<WorkerSlot>(); + + for (Entry<WorkerSlot, List<ExecutorDetails>> entry : existingSlots.entrySet()) { + Integer executorCount = distribution.get(entry.getValue().size()); + if (executorCount != null && executorCount > 0) { + _slots.add(entry.getKey()); + executorCount--; + distribution.put(entry.getValue().size(), executorCount); + } + } + + for (WorkerSlot slot : _slots) { + existingSlots.remove(slot); + } + + return existingSlots.keySet(); + } + + return null; + } + + public static Set<WorkerSlot> slotsCanReassign(Cluster cluster, Set<WorkerSlot> slots) { + Set<WorkerSlot> result = new HashSet<WorkerSlot>(); + for (WorkerSlot slot : slots) { + if (!cluster.isBlackListed(slot.getNodeId())) { + SupervisorDetails supervisor = cluster.getSupervisorById(slot.getNodeId()); + if (supervisor != null) { + Set<Integer> ports = supervisor.getAllPorts(); + if (ports != null && ports.contains(slot.getPort())) { + result.add(slot); + } + } + } + } + return result; + } + + public static void defaultSchedule(Topologies topologies, Cluster cluster) { + List<TopologyDetails> needsSchedulingTopologies = cluster.needsSchedulingTopologies(topologies); + for (TopologyDetails topology : needsSchedulingTopologies) { + List<WorkerSlot> availableSlots = cluster.getAvailableSlots(); + Set<ExecutorDetails> allExecutors = (Set<ExecutorDetails>) topology.getExecutors(); + + Map<WorkerSlot, List<ExecutorDetails>> aliveAssigned = EvenScheduler.getAliveAssignedWorkerSlotExecutors(cluster, topology.getId()); + Set<ExecutorDetails> aliveExecutors = new HashSet<ExecutorDetails>(); + for (List<ExecutorDetails> list : aliveAssigned.values()) { + aliveExecutors.addAll(list); + } + + Set<WorkerSlot> canReassignSlots = slotsCanReassign(cluster, aliveAssigned.keySet()); + int totalSlotsToUse = Math.min(topology.getNumWorkers(), canReassignSlots.size() + availableSlots.size()); + + Set<WorkerSlot> badSlot = null; + if (totalSlotsToUse > aliveAssigned.size() || !allExecutors.equals(aliveExecutors)) { + badSlot = badSlots(aliveAssigned, allExecutors.size(), totalSlotsToUse); + } + if (badSlot != null) { + cluster.freeSlots(badSlot); + } + + Map<String, TopologyDetails> _topologies = new HashMap<String, TopologyDetails>(); + _topologies.put(topology.getId(), topology); + EvenScheduler.scheduleTopologiesEvenly(new Topologies(_topologies), cluster); + } + } + + @Override + public void prepare(Map conf) { + //noop + } + + @Override + public void schedule(Topologies topologies, Cluster cluster) { + defaultSchedule(topologies, cluster); + } + +} http://git-wip-us.apache.org/repos/asf/storm/blob/4c41ac18/storm-core/src/jvm/org/apache/storm/scheduler/EvenScheduler.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/org/apache/storm/scheduler/EvenScheduler.java b/storm-core/src/jvm/org/apache/storm/scheduler/EvenScheduler.java new file mode 100644 index 0000000..a29d45f --- /dev/null +++ b/storm-core/src/jvm/org/apache/storm/scheduler/EvenScheduler.java @@ -0,0 +1,177 @@ +/** + * 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.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Set; +import java.util.TreeMap; + +import org.apache.storm.utils.Utils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.collect.Sets; + +public class EvenScheduler implements IScheduler { + private static final Logger LOG = LoggerFactory.getLogger(EvenScheduler.class); + + public static List<WorkerSlot> sortSlots(List<WorkerSlot> availableSlots, Cluster cluster) { + if (availableSlots != null && availableSlots.size() > 0) { + // group by node + Map<String, List<WorkerSlot>> slotGroups = new TreeMap<String, List<WorkerSlot>>(); + for (WorkerSlot slot : availableSlots) { + String host = cluster.getHost(slot.getNodeId()); + List<WorkerSlot> slots = slotGroups.get(host); + if (slots == null) { + slots = new ArrayList<WorkerSlot>(); + slotGroups.put(host, slots); + } + slots.add(slot); + } + + // sort by port + for (List<WorkerSlot> slots : slotGroups.values()) { + Collections.sort(slots, new Comparator<WorkerSlot>() { + @Override + public int compare(WorkerSlot o1, WorkerSlot o2) { + return o1.getPort() - o2.getPort(); + } + }); + } + + // sort by count + List<List<WorkerSlot>> list = new ArrayList<List<WorkerSlot>>(slotGroups.values()); + Collections.sort(list, new Comparator<List<WorkerSlot>>() { + @Override + public int compare(List<WorkerSlot> o1, List<WorkerSlot> o2) { + return o2.size() - o1.size(); + } + }); + + return Utils.interleaveAll(list); + } + + return null; + } + + public static Map<WorkerSlot, List<ExecutorDetails>> getAliveAssignedWorkerSlotExecutors(Cluster cluster, String topologyId) { + SchedulerAssignment existingAssignment = cluster.getAssignmentById(topologyId); + Map<ExecutorDetails, WorkerSlot> executorToSlot = null; + if (existingAssignment != null) { + executorToSlot = existingAssignment.getExecutorToSlot(); + } + + Map<WorkerSlot, List<ExecutorDetails>> result = new HashMap<WorkerSlot, List<ExecutorDetails>>(); + if (executorToSlot != null) { + for (Entry<ExecutorDetails, WorkerSlot> entry : executorToSlot.entrySet()) { + List<ExecutorDetails> list = result.get(entry.getValue()); + if (list == null) { + list = new ArrayList<ExecutorDetails>(); + result.put(entry.getValue(), list); + } + list.add(entry.getKey()); + } + } + return result; + } + + private static Map<ExecutorDetails, WorkerSlot> scheduleTopology(TopologyDetails topology, Cluster cluster) { + List<WorkerSlot> availableSlots = cluster.getAvailableSlots(); + Set<ExecutorDetails> allExecutors = (Set<ExecutorDetails>) topology.getExecutors(); + Map<WorkerSlot, List<ExecutorDetails>> aliveAssigned = getAliveAssignedWorkerSlotExecutors(cluster, topology.getId()); + int totalSlotsToUse = Math.min(topology.getNumWorkers(), availableSlots.size() + aliveAssigned.size()); + + List<WorkerSlot> sortedList = sortSlots(availableSlots, cluster); + if (sortedList == null) { + LOG.error("Available slots are not enough for topology: {}", topology.getName()); + return new HashMap<ExecutorDetails, WorkerSlot>(); + } + + List<WorkerSlot> reassignSlots = sortedList.subList(0, totalSlotsToUse - aliveAssigned.size()); + Set<ExecutorDetails> aliveExecutors = new HashSet<ExecutorDetails>(); + for (List<ExecutorDetails> list : aliveAssigned.values()) { + aliveExecutors.addAll(list); + } + Set<ExecutorDetails> reassignExecutors = Sets.difference(allExecutors, aliveExecutors); + + Map<ExecutorDetails, WorkerSlot> reassignment = new HashMap<ExecutorDetails, WorkerSlot>(); + if (reassignSlots.size() == 0) { + return reassignment; + } + + List<ExecutorDetails> _executors = new ArrayList<ExecutorDetails>(reassignExecutors); + Collections.sort(_executors, new Comparator<ExecutorDetails>() { + @Override + public int compare(ExecutorDetails o1, ExecutorDetails o2) { + return o1.getStartTask() - o2.getStartTask(); + } + }); + + int numExecutors = _executors.size(); + List<WorkerSlot> _slots = new ArrayList<WorkerSlot>(numExecutors); + int numSlots = reassignSlots.size(); + for (int i = 0; i < numExecutors; i++) { + _slots.add(reassignSlots.get(i % numSlots)); + } + + Iterator<WorkerSlot> slotIterator = _slots.iterator(); + Iterator<ExecutorDetails> executorIterator = _executors.iterator(); + while (slotIterator.hasNext() && executorIterator.hasNext()) { + reassignment.put(executorIterator.next(), slotIterator.next()); + } + + if (reassignment.size() != 0) { + LOG.info("Available slots: {}", availableSlots.toString()); + } + return reassignment; + } + + public static void scheduleTopologiesEvenly(Topologies topologies, Cluster cluster) { + List<TopologyDetails> needsSchedulingTopologies = cluster.needsSchedulingTopologies(topologies); + for (TopologyDetails topology : needsSchedulingTopologies) { + String topologyId = topology.getId(); + Map<ExecutorDetails, WorkerSlot> newAssignment = scheduleTopology(topology, cluster); + Map<WorkerSlot, List<ExecutorDetails>> nodePortToExecutors = Utils.reverseMap(newAssignment); + + for (Map.Entry<WorkerSlot, List<ExecutorDetails>> entry : nodePortToExecutors.entrySet()) { + WorkerSlot nodePort = entry.getKey(); + List<ExecutorDetails> executors = entry.getValue(); + cluster.assign(nodePort, topologyId, executors); + } + } + } + + @Override + public void prepare(Map conf) { + //noop + } + + @Override + public void schedule(Topologies topologies, Cluster cluster) { + scheduleTopologiesEvenly(topologies, cluster); + } + +} http://git-wip-us.apache.org/repos/asf/storm/blob/4c41ac18/storm-core/src/jvm/org/apache/storm/utils/Utils.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/org/apache/storm/utils/Utils.java b/storm-core/src/jvm/org/apache/storm/utils/Utils.java index a0c0b1a..b1f59f2 100644 --- a/storm-core/src/jvm/org/apache/storm/utils/Utils.java +++ b/storm-core/src/jvm/org/apache/storm/utils/Utils.java @@ -2248,4 +2248,23 @@ public class Utils { } return process; } + + public static <T> List<T> interleaveAll(List<List<T>> nodeList) { + if (nodeList != null && nodeList.size() > 0) { + List<T> first = new ArrayList<T>(); + List<List<T>> rest = new ArrayList<List<T>>(); + for (List<T> node : nodeList) { + if (null != node && node.size() > 0) { + first.add(node.get(0)); + rest.add(node.subList(1, node.size())); + } + } + List<T> interleaveRest = interleaveAll(rest); + if (interleaveRest != null) { + first.addAll(interleaveRest); + } + return first; + } + return null; + } } http://git-wip-us.apache.org/repos/asf/storm/blob/4c41ac18/storm-core/test/clj/org/apache/storm/scheduler_test.clj ---------------------------------------------------------------------- diff --git a/storm-core/test/clj/org/apache/storm/scheduler_test.clj b/storm-core/test/clj/org/apache/storm/scheduler_test.clj index fc6e8e3..b933372 100644 --- a/storm-core/test/clj/org/apache/storm/scheduler_test.clj +++ b/storm-core/test/clj/org/apache/storm/scheduler_test.clj @@ -16,7 +16,7 @@ (ns org.apache.storm.scheduler-test (:use [clojure test]) (:use [org.apache.storm config testing]) - (:use [org.apache.storm.scheduler EvenScheduler]) + (:import [org.apache.storm.scheduler EvenScheduler]) (:require [org.apache.storm.daemon [nimbus :as nimbus]]) (:import [org.apache.storm.generated StormTopology]) (:import [org.apache.storm.scheduler Cluster SupervisorDetails WorkerSlot ExecutorDetails @@ -265,7 +265,7 @@ (is (= '(["supervisor2" 6700] ["supervisor1" 6700] ["supervisor2" 6701] ["supervisor1" 6701] ["supervisor2" 6702]) - (sort-slots [["supervisor1" 6700] ["supervisor1" 6701] + (EvenScheduler/sortSlots [["supervisor1" 6700] ["supervisor1" 6701] ["supervisor2" 6700] ["supervisor2" 6701] ["supervisor2" 6702] ]))) ;; test supervisor3 has more free slots @@ -273,7 +273,7 @@ ["supervisor3" 6703] ["supervisor2" 6701] ["supervisor1" 6701] ["supervisor3" 6702] ["supervisor2" 6702] ["supervisor3" 6701]) - (sort-slots [["supervisor1" 6700] ["supervisor1" 6701] + (EvenScheduler/sortSlots [["supervisor1" 6700] ["supervisor1" 6701] ["supervisor2" 6700] ["supervisor2" 6701] ["supervisor2" 6702] ["supervisor3" 6700] ["supervisor3" 6703] ["supervisor3" 6702] ["supervisor3" 6701] ])))
