Storm2

Posted 2020-07-10

package storm.scheduler;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;

import backtype.storm.scheduler.Cluster;
import backtype.storm.scheduler.EvenScheduler;
import backtype.storm.scheduler.ExecutorDetails;
import backtype.storm.scheduler.IScheduler;
import backtype.storm.scheduler.SupervisorDetails;
import backtype.storm.scheduler.Topologies;
import backtype.storm.scheduler.TopologyDetails;
import backtype.storm.scheduler.WorkerSlot;

/**
 * 基于Storm Topology 热边的调度算法
 * @author wxweven
 */
public class OnlineScheduler implements IScheduler {
	private static final int DEFAULT_RESCHEDULE_TIMEOUT = 180;	// 单位为s
	private Logger logger = Logger.getLogger(OnlineScheduler.class);
	private AssignmentTracker assignmentTracker = new AssignmentTracker();
	private long lastRescheduling;

	/**
	 * 重写IScheduler中的方法，实现具体的调度算法
	 */
	@Override
	public void schedule(Topologies topologies, Cluster cluster) {
		logger.info("HotEdge Scheduler");
		if (!topologies.getTopologies().isEmpty()) {
			int rescheduleTimeout = DEFAULT_RESCHEDULE_TIMEOUT;
			for (TopologyDetails topology : topologies.getTopologies()) {
				rescheduleTimeout = Integer.parseInt(topology.getConf()
						.get(Utils.RESCHEDULE_TIMEOUT).toString());
			}
			long now = System.currentTimeMillis();
			long elapsedTime = (now - lastRescheduling) / 1000; // s
			if (lastRescheduling == 0 || elapsedTime >= rescheduleTimeout) {
				//执行具体的调度算法
				doSchedule(topologies, cluster);
			}
		}

		new EvenScheduler().schedule(topologies, cluster);
		assignmentTracker.checkAssignment(topologies, cluster);
	}

	/**
	 * 实现具体的改进调度算法
	 * 
	 * @param topologies
	 *            提交的Topology
	 * @param cluster
	 *            集群
	 */
	private void doSchedule(Topologies topologies, Cluster cluster) {
		try {
			// 从数据库中获取提交的Topology
			List<String> dbTopologies = DataManager.getInstance().getTopologies();
			List<String> topologiesToBeRemoved = new ArrayList<String>(dbTopologies);
			List<String> stormTopologyList = new ArrayList<String>();
			int trafficImprovement = 0;
			for (TopologyDetails topology : topologies.getTopologies()) {
				topologiesToBeRemoved.remove(topology.getId());
				stormTopologyList.add(topology.getId());
				for (Object key : topology.getConf().keySet()) {
					logger.debug("- " + key + ": " + topology.getConf().get(key));
				}
				trafficImprovement = Integer.parseInt(topology.getConf()
						.get(Utils.TRAFFIC_IMPROVEMENT).toString());
			}

			dbTopologies.removeAll(topologiesToBeRemoved);

			if (!topologiesToBeRemoved.isEmpty()) {
				DataManager.getInstance().removeTopologies(topologiesToBeRemoved);
				logger.info("Topologies succesfully removed from DB");
			}

			// 计算调度策略
			TrafficManager.getInstance().clear();
			computeBestScheduling(dbTopologies, topologies, cluster);
			Map<Node, List<Slot>> bestAssignment = TrafficManager.getInstance().getAssignments();
			int bestInterNodeTraffic = TrafficManager.getInstance().computeInterNodeTraffic();
			int currentInterNodeTraffic = DataManager.getInstance().getCurrentInterNodeTraffic();
			List<Node> overloadedNodeList = DataManager.getInstance().getOverloadedNodes();

			logger.info("These nodes are currently overloaded: "
					+ Utils.collectionToString(overloadedNodeList));
			logger.info("Currently, the inter-node traffic is " + currentInterNodeTraffic
					+ " tuple/s");

			if (bestAssignment != null) {
				logger.info("The best assignment can lead to an inter-node traffic of "
						+ bestInterNodeTraffic + " tuple/s");
				boolean reschedulingDueToOverloading = false;
				boolean reschedulingDueToInterNodeTraffic = false;
				if (!overloadedNodeList.isEmpty()) {
					logger.info("Check how the new assignment can offload some of the currently overloaded nodes");
					for (Node node : overloadedNodeList) {
						Node nodeAfterTheAssignment = null;
						for (Node n : bestAssignment.keySet()) {
							if (n.equals(node)) {
								nodeAfterTheAssignment = n;
								break;
							}
						}
						if (nodeAfterTheAssignment != null) {
							logger.info("Node " + node.getName() + " currenlty has a load of "
									+ node.getLoad() + " Hz/s, and the assignment can lead to "
									+ nodeAfterTheAssignment.getLoad());
							if (node.getLoad() > nodeAfterTheAssignment.getLoad()) {
								reschedulingDueToOverloading = true;
							}
						} else {
							logger.warn("Node " + node.getName() + " currenlty has a load of "
									+ node.getLoad()
									+ " Hz/s, but it doesn‘t appear in the new assignment");
						}
					}
				}
				if (reschedulingDueToOverloading) {
					logger.info("A rescheduling is required to offload currently overloaded nodes");
				}
				int trafficThreshold = (int) (currentInterNodeTraffic * (1 - (float) trafficImprovement / 100));
				logger.info("Minimum traffic threshold is " + trafficThreshold + " tuple/s");
				if (trafficThreshold >= bestInterNodeTraffic) {
					logger.info("A rescheduling is required to lower inter-node traffic");
					reschedulingDueToInterNodeTraffic = true;
				}
				if (reschedulingDueToInterNodeTraffic || reschedulingDueToOverloading) {
					logger.info("Let‘s apply the best assignment!!");
					lastRescheduling = System.currentTimeMillis();

					// free all available slots
					for (SupervisorDetails supervisor : cluster.getSupervisors().values()) {
						List<Integer> usedPorts = cluster.getUsedPorts(supervisor);
						for (int usedPort : usedPorts) {
							cluster.freeSlot(new WorkerSlot(supervisor.getId(), usedPort));
						}
					}

					for (Node node : bestAssignment.keySet()) {
						SupervisorDetails supervisor = cluster.getSupervisorsByHost(node.getName())
								.get(0);
						List<WorkerSlot> availableSlots = cluster.getAvailableSlots(supervisor);
						int slotIndex = 0;
						for (Slot slot : bestAssignment.get(node)) {
							logger.info("Assigning executors of slot " + slot);
							String topology = slot.getTopology().getTopologyID();
							List<ExecutorDetails> executorList = new ArrayList<ExecutorDetails>();

							// here a match is required to link scheduler
							// executors to storm executors
							Collection<ExecutorDetails> allTopologyExecutors = topologies.getById(
									topology).getExecutors();
							for (Executor executor : slot.getExecutors()) {
								for (ExecutorDetails executorDetails : allTopologyExecutors) {
									if (executor.match(executorDetails)) {
										executorList.add(executorDetails);
										break;
									}
								}
							}

							cluster.assign(availableSlots.get(slotIndex), topology, executorList);
							logger.info("We assigned executors:"
									+ Utils.collectionToString(executorList) + " to slot: ["
									+ availableSlots.get(slotIndex).getNodeId() + ", "
									+ availableSlots.get(slotIndex).getPort() + "]");
							slotIndex++;

						}

					}

					DataManager.getInstance().removeTopologies(dbTopologies);

				}
			} else {
				logger.info("No assignment has been simulated");
			}

		} catch (Throwable t) {
			logger.error("Error occurred during scheduling", t);
		}
	}



	private void computeBestScheduling(List<String> dbTopologies, Topologies stormTopologies,
			Cluster cluster) throws Exception {

		logger.info("-- First phase --");
		List<Topology> topologyList = new ArrayList<Topology>();
		for (String topologyID : dbTopologies) {
			logger.info("Topology ID: " + topologyID);
			TopologyDetails topologyDetails = stormTopologies.getById(topologyID);
			Topology topology = new Topology(topologyDetails);
			topology.setTotalLoad(DataManager.getInstance().getTotalLoad(topologyID));
			topologyList.add(topology);


			List<ExecutorPair> interExecutorTrafficList = TrafficManager.getInstance()
					.getInterExecutorTrafficList(topologyID);
			logger.info("Inter-executor traffic stats: "
					+ Utils.collectionToString(interExecutorTrafficList));
			if (interExecutorTrafficList.isEmpty()) {
				logger.info("Traffic stats are not complete yet, skip this topology");
			} else {
				for (ExecutorPair executorPair : interExecutorTrafficList) {
					logger.debug("Executor pair: " + executorPair);
					List<Slot> slotList = topology.getContainingSlotList(executorPair.getSource(),
							executorPair.getDestination());
					logger.debug("Slots that already contain either executors: "
							+ Utils.collectionToString(slotList));
					if (slotList.isEmpty()) {
						logger.debug("Both executors have not been assigned yet, try to add them to the least loaded slot");
						Slot leastLoadedSlot = topology.getLeastLoadedSlot(
								executorPair.getSource(), executorPair.getDestination());
						if (leastLoadedSlot != null) {
							logger.debug("Least loaded slot able to get both the executors: "
									+ leastLoadedSlot);
							leastLoadedSlot.assign(executorPair.getSource());
							leastLoadedSlot.assign(executorPair.getDestination());
							logger.info("Executors " + executorPair.getSource() + " and "
									+ executorPair.getDestination() + " assigned to slot "
									+ leastLoadedSlot);
						} else {
							logger.debug("No slot exists that can get both the executors, assign them to distinct slots");
							leastLoadedSlot = topology.getLeastLoadedSlot(executorPair.getSource());
							if (leastLoadedSlot == null) {
								throw new RuntimeException(
										"Cannot find a slot able to get executor "
												+ executorPair.getSource() + " for topology "
												+ topology);
							}
							logger.debug("Least loaded slot for source executor: "
									+ leastLoadedSlot);
							leastLoadedSlot.assign(executorPair.getSource());
							logger.info("Executor " + executorPair.getSource()
									+ " assigned to slot " + leastLoadedSlot);
							if (leastLoadedSlot.canAccept(executorPair.getDestination())) {
								logger.debug("After having added executor "
										+ executorPair.getSource() + ", the slot "
										+ leastLoadedSlot + " can also get the executor "
										+ executorPair.getDestination());
								leastLoadedSlot.assign(executorPair.getDestination());
							} else {
								logger.debug("After having added executor "
										+ executorPair.getSource() + ", the slot "
										+ leastLoadedSlot + " cannot get the executor "
										+ executorPair.getDestination());
								leastLoadedSlot = topology.getLeastLoadedSlot(executorPair
										.getDestination());
								if (leastLoadedSlot == null) {
									throw new RuntimeException(
											"Cannot find a slot able to get executor "
													+ executorPair.getDestination()
													+ " for topology " + topology);
								}
								logger.debug("Least loaded slot for destination executor: "
										+ leastLoadedSlot);
								leastLoadedSlot.assign(executorPair.getDestination());
							}
							logger.info("Executor " + executorPair.getDestination()
									+ " assigned to slot " + leastLoadedSlot);
						}
					} else {
						logger.debug("Some executor has been already assigned, compute the best assignment using the slot(s) found before and the least loaded one");
						Slot leastLoadedSlot = topology.getLeastLoadedSlot(executorPair);
						logger.debug("Least loaded slot: " + leastLoadedSlot);
						if (leastLoadedSlot != null && !slotList.contains(leastLoadedSlot)) {
							slotList.add(leastLoadedSlot);
						}
						logger.debug("Slots to use: " + Utils.collectionToString(slotList));

						logger.debug("Remove source and destination from the slots they are currently assigned to");
						for (Slot slot : slotList) {
							if (slot.contains(executorPair.getSource())) {
								slot.remove(executorPair.getSource());
							}
							if (slot.contains(executorPair.getDestination())) {
								slot.remove(executorPair.getDestination());
							}
						}
						logger.debug("Slots to use after such removals: "
								+ Utils.collectionToString(slotList));

						logger.debug("Check every possible combination");
						Slot bestSlotForSource = null;
						Slot bestSlotForDestination = null;
						int minInterSlotTraffic = -1;
						for (Slot slotForSource : slotList) {
							for (Slot slotForDestination : slotList) {
								logger.debug("Assigning executor " + executorPair.getSource()
										+ " to slot " + slotForSource + " and executor "
										+ executorPair.getDestination() + " to slot "
										+ slotForDestination + "...");
								boolean assignmentOk = true;
								if (slotForSource.canAccept(executorPair.getSource())) {
									slotForSource.assign(executorPair.getSource());
								} else {
									logger.debug("Slot "
											+ slotForSource
											+ " is imbalanced, cannot be used to add more executors");
									assignmentOk = false;
								}

								if (slotForDestination.canAccept(executorPair.getDestination())) {
									slotForDestination.assign(executorPair.getDestination());
								} else {
									logger.debug("Slot "
											+ slotForDestination
											+ " is imbalanced, cannot be used to add more executors");
									assignmentOk = false;
								}

								if (assignmentOk) {
									int interSlotTraffic = TrafficManager.getInstance()
											.computeInterSlotTraffic(topologyID);
									logger.debug("...the inter-slot traffic is " + interSlotTraffic
											+ " tuple/s");
									if (minInterSlotTraffic == -1
											|| interSlotTraffic < minInterSlotTraffic) {
										bestSlotForSource = slotForSource;
										bestSlotForDestination = slotForDestination;
										minInterSlotTraffic = interSlotTraffic;
									}
								}
								if (slotForSource.contains(executorPair.getSource())) {
									slotForSource.remove(executorPair.getSource());
								}
								if (slotForDestination.contains(executorPair.getDestination())) {
									slotForDestination.remove(executorPair.getDestination());
								}
							}
						}
						if (bestSlotForSource == null || bestSlotForDestination == null) {
							throw new Exception("Cannot find a possible assignment of executors "
									+ executorPair.getSource() + " and "
									+ executorPair.getDestination() + " to slots "
									+ Utils.collectionToString(slotList));
						}
						logger.debug("The best assignment is executor " + executorPair.getSource()
								+ " to slot " + bestSlotForSource + " and executor "
								+ executorPair.getDestination() + " to slot "
								+ bestSlotForDestination + ", with inter-slot traffic "
								+ minInterSlotTraffic + " tuple/s");
						bestSlotForSource.assign(executorPair.getSource());
						bestSlotForDestination.assign(executorPair.getDestination());
						logger.info("Executor " + executorPair.getSource() + " assigned to slot "
								+ bestSlotForSource);
						logger.info("Executor " + executorPair.getDestination()
								+ " assigned to slot " + bestSlotForDestination);
					} /* end if (!slotList.isEmpty()) */

					logger.debug("Assignment of executors " + executorPair + " completed");

				} /* end for (ExecutorPair executorPair : executorPairList) */

				logger.info("Current assignment: " + Utils.collectionToString(topology.getSlots()));
				logger.info("Check for empty slots");
				List<Slot> emptySlotList = topology.getEmptySlots();
				if (emptySlotList.isEmpty()) {
					logger.info("No empty slots, the assignment is succesfully completed");
				} else {
					logger.info("Empty slots: " + Utils.collectionToString(emptySlotList));
					List<Slot> usedSlotList = topology.getUsedSlots();
					for (Slot emptySlot : emptySlotList) {
						logger.debug("Find an executor to assign to slot " + emptySlot);
						Executor bestExecutor = null;
						Slot bestSlot = null;
						int bestInterSlotTraffic = -1;
						for (Slot usedSlot : usedSlotList) {
							if (usedSlot.getExecutors().size() > 1) {
								logger.debug("Check the executors of slot " + usedSlot);
								List<Executor> executorList = new ArrayList<Executor>(
										usedSlot.getExecutors());
								for (Executor executor : executorList) {
									usedSlot.remove(executor);
									emptySlot.assign(executor);
									int interSlotTraffic = TrafficManager.getInstance()
											.computeInterSlotTraffic(topologyID);
									logger.debug("Moving executor " + executor
											+ ", the inter-slot traffic is " + interSlotTraffic
											+ " tuple/s");
									if (bestInterSlotTraffic == -1
											|| interSlotTraffic < bestInterSlotTraffic) {
										bestExecutor = executor;
										bestSlot = usedSlot;
										bestInterSlotTraffic = interSlotTraffic;
									}
									emptySlot.remove(executor);
									usedSlot.assign(executor);
								}
							}
						}
						if (bestSlot != null) {
							logger.debug("The best assignment is moving executor " + bestExecutor
									+ " from slot " + bestSlot + " to slot " + emptySlot
									+ " with an inter-slot traffic of " + bestInterSlotTraffic
									+ " tuple/s");
							bestSlot.remove(bestExecutor);
							emptySlot.assign(bestExecutor);
							logger.info("Executor " + bestExecutor + " moved from slot " + bestSlot
									+ " to slot " + emptySlot);
						} else {
							logger.warn("Cannot find an executor to move to slot " + emptySlot);
						}
					} /* end for (Slot emptySlot : emptySlotList) */
				}
				logger.info("Next assignment: " + Utils.collectionToString(topology.getSlots()));

				TrafficManager.getInstance().compileInterSlotTraffic();

			} /* end if (!executorPairList.isEmpty()) */

			logger.info("Assignments of executors for topology " + topologyID + " completed");

		} /* end for (String topologyID : dbTopologies) */
		logger.info("First phase completed!");
		List<SlotPair> interSlotTrafficList = TrafficManager.getInstance()
				.getInterSlotTrafficList();
		logger.info("Inter-slot traffic stats: " + Utils.collectionToString(interSlotTrafficList));

		// second phase
		logger.info("-- Second phase --");
		NodeManager nodeManager = new NodeManager(topologyList, cluster);
		if (nodeManager.getNodeCount() == 0) {
			logger.info("No nodes have been configured yet, cannot determine any scheduling");
		} else {
			for (SlotPair slotPair : interSlotTrafficList) {
				logger.info("Slot pair: " + slotPair);
				List<Node> nodeList = TrafficManager.getInstance().getContainingNodeList(
						slotPair.getFirst(), slotPair.getSecond());
				if (nodeList.isEmpty()) {
					logger.debug("Both slots have not been assigned yet, try to add them to the least loaded node");
					Node leastLoadedNode = nodeManager.getLeastLoadedNode(slotPair.getFirst(),
							slotPair.getSecond());
					if (leastLoadedNode != null) {
						logger.debug("Least loaded node able to get both the slots: "
								+ leastLoadedNode);
						leastLoadedNode.assign(slotPair.getFirst());
						leastLoadedNode.assign(slotPair.getSecond());
						logger.info("Slots "
								+ slotPair.getFirst()
								+ " and "
								+ slotPair.getSecond()
								+ " assigned to node "
								+ leastLoadedNode
								+ " (Slots of topology "
								+ slotPair.getFirst().getTopology().getTopologyID()
								+ " in this node: "
								+ leastLoadedNode.getTopologySlotCount(slotPair.getFirst()
										.getTopology().getTopologyID()));
					} else {
						logger.debug("No node exists that can get both the slots, assign them to distinct nodes");

						leastLoadedNode = nodeManager.getLeastLoadedNode(slotPair.getFirst());
						if (leastLoadedNode == null) {
							throw new RuntimeException(
									"Cannot find a node able to sustain the load of the slot "
											+ slotPair.getFirst());
						}
						logger.debug("Assign slot " + slotPair.getFirst() + " to node "
								+ leastLoadedNode);
						leastLoadedNode.assign(slotPair.getFirst());
						logger.debug("Least loaded node after such assignment: " + leastLoadedNode);
						logger.info("Slot "
								+ slotPair.getFirst()
								+ " assigned to node "
								+ leastLoadedNode
								+ " (Slots of topology "
								+ slotPair.getFirst().getTopology().getTopologyID()
								+ " in this node: "
								+ leastLoadedNode.getTopologySlotCount(slotPair.getFirst()
										.getTopology().getTopologyID()));

						leastLoadedNode = nodeManager.getLeastLoadedNode(slotPair.getSecond());
						if (leastLoadedNode == null) {
							throw new RuntimeException(
									"Cannot find a node able to sustain the load of the slot "
											+ slotPair.getSecond());
						}
						logger.debug("Assign slot " + slotPair.getSecond() + " to node "
								+ leastLoadedNode);
						leastLoadedNode.assign(slotPair.getSecond());
						logger.debug("Least loaded node after such assignment: " + leastLoadedNode);
						logger.info("Slot "
								+ slotPair.getSecond()
								+ " assigned to node "
								+ leastLoadedNode
								+ " (Slots of topology "
								+ slotPair.getSecond().getTopology().getTopologyID()
								+ " in this node: "
								+ leastLoadedNode.getTopologySlotCount(slotPair.getSecond()
										.getTopology().getTopologyID()));
					}
				} else {
					logger.debug("Either slot has been already assigned to nodes "
							+ Utils.collectionToString(nodeList));
					Node leastLoadedNode = nodeManager.getLeastLoadedNode(slotPair);
					logger.debug("The least loaded node able to sustain the traffic of the slot with the lowest traffic is "
							+ leastLoadedNode);
					if (leastLoadedNode != null && !nodeList.contains(leastLoadedNode)) {
						nodeList.add(leastLoadedNode);
					}
					logger.debug("Nodes to use: " + Utils.collectionToString(nodeList));
					logger.debug("Remove slots " + slotPair.getFirst() + " and "
							+ slotPair.getSecond() + " from these nodes");
					for (Node node : nodeList) {
						if (node.contains(slotPair.getFirst())) {
							node.remove(slotPair.getFirst());
						}
						if (node.contains(slotPair.getSecond())) {
							node.remove(slotPair.getSecond());
						}
					}
					logger.debug("Nodes after such removals: " + Utils.collectionToString(nodeList));

					logger.debug("Check every possible combination");
					Node bestNodeForFirst = null;
					Node bestNodeForSecond = null;
					int minInterNodeTraffic = -1;
					for (Node nodeForFirst : nodeList) {
						for (Node nodeForSecond : nodeList) {
							logger.debug("Assigning slot " + slotPair.getFirst() + " to node "
									+ nodeForFirst + " and slot " + slotPair.getSecond()
									+ " to node " + nodeForSecond + "...");
							boolean assignmentOK = true;

							if (nodeForFirst.canAssign(slotPair.getFirst())) {
								nodeForFirst.assign(slotPair.getFirst());
							} else {
								logger.debug("Cannot assign slot " + slotPair.getFirst()
										+ " to noe " + nodeForFirst);
								assignmentOK = false;
							}

							if (nodeForSecond.canAssign(slotPair.getSecond())) {
								nodeForSecond.assign(slotPair.getSecond());
							} else {
								logger.debug("Cannot assign slot " + slotPair.getSecond()
										+ " to node " + nodeForSecond);
								assignmentOK = false;
							}

							if (assignmentOK) {
								int tmpInterNodeTraffic = TrafficManager.getInstance()
										.computeInterNodeTraffic();
								logger.debug("...the inter-node traffic is " + tmpInterNodeTraffic
										+ " tuple/s");
								if (minInterNodeTraffic == -1
										|| tmpInterNodeTraffic < minInterNodeTraffic) {
									bestNodeForFirst = nodeForFirst;
									bestNodeForSecond = nodeForSecond;
									minInterNodeTraffic = tmpInterNodeTraffic;
								}
							}

							if (nodeForFirst.contains(slotPair.getFirst())) {
								nodeForFirst.remove(slotPair.getFirst());
							}
							if (nodeForSecond.contains(slotPair.getSecond())) {
								nodeForSecond.remove(slotPair.getSecond());
							}
						}
					}

					logger.debug("The best assignment is slot " + slotPair.getFirst() + " to node "
							+ bestNodeForFirst + " and slot " + slotPair.getSecond() + " to node "
							+ bestNodeForSecond + ", with inter-node traffic "
							+ minInterNodeTraffic + " tuple/s");
					bestNodeForFirst.assign(slotPair.getFirst());
					bestNodeForSecond.assign(slotPair.getSecond());
					logger.info("Slot "
							+ slotPair.getFirst()
							+ " assigned to node "
							+ bestNodeForFirst
							+ " (Slots of topology "
							+ slotPair.getFirst().getTopology().getTopologyID()
							+ " in this node: "
							+ bestNodeForFirst.getTopologySlotCount(slotPair.getFirst()
									.getTopology().getTopologyID()));
					logger.info("Slot "
							+ slotPair.getSecond()
							+ " assigned to node "
							+ bestNodeForSecond
							+ " (Slots of topology "
							+ slotPair.getSecond().getTopology().getTopologyID()
							+ " in this node: "
							+ bestNodeForSecond.getTopologySlotCount(slotPair.getSecond()
									.getTopology().getTopologyID()));

				} /* end if (!nodeList.isEmpty()) */

				logger.info("Assignment of slots " + slotPair + " completed");

			} /* end for (SlotPair slotPair : interSlotTrafficList) */

			if (TrafficManager.getInstance().getAssignments() != null) {
				logger.info("Intermediate assignment: "
						+ Utils.collectionToString(TrafficManager.getInstance().getAssignments()
								.keySet()));
			}

			/*
			 * ensure that the slots of a given topology are assigned to the
			 * proper number of nodes, otherwise the chances of
			 * parallelization/pipelining are not rightly exploited
			 */
			logger.info("Check whether all the topologies are using the desired number of nodes");
			for (Topology topology : topologyList) {
				// int numberOfNodesToUse = Math.min(topology.getSlots().size(),
				// nodeManager.getNodeCount());
				int numberOfNodesToUse = topology.getNumberOfNodesToUse(nodeManager.getNodeCount());
				List<Node> usedNodeList = null;
				while (TrafficManager.getInstance().getNodeList(topology) != null
						&& (usedNodeList = new ArrayList<Node>(TrafficManager.getInstance()
								.getNodeList(topology))).size() < numberOfNodesToUse) {
					logger.info("Topology " + topology + " is using " + usedNodeList.size()
							+ " nodes, while it should use " + numberOfNodesToUse);
					Node bestUsedNode = null;
					Node bestUnusedNode = null;
					Slot bestSlot = null;
					int bestTraffic = -1;
					for (Node usedNode : usedNodeList) {
						// check if this node has more than one slot for that
						// topology
						int topologySlotCount = 0;
						List<Slot> nodeSlotList = new ArrayList<Slot>(usedNode.getSlotList());
						for (Slot slot : nodeSlotList) {
							if (slot.getTopology() == topology) {
								topologySlotCount++;
							}
						}
						if (topologySlotCount > 1) {
							for (Slot slot : nodeSlotList) {
								Node unusedNode = nodeManager.getUnusedNode(usedNodeList, slot);
								if (unusedNode != null) {
									usedNode.remove(slot);
									unusedNode.assign(slot);
									int traffic = TrafficManager.getInstance()
											.computeInterNodeTraffic();
									logger.info("Moving slot " + slot + " from node " + usedNode
											+ " to node " + unusedNode + ", the traffic becomes "
											+ traffic + " tuple/s");
									if (bestUsedNode == null || traffic < bestTraffic) {
										bestUsedNode = usedNode;
										bestUnusedNode = unusedNode;
										bestSlot = slot;
										bestTraffic = traffic;
									}
									unusedNode.remove(slot);
									usedNode.assign(slot);
								} /* end if (unusedNode != null) */
							} /* end for (Slot slot : nodeSlotList) */
						} /* end if (topologySlotCount > 1) */
					} /* end for (Node usedNode : usedNodeList) */

					if (bestUnusedNode != null) {
						logger.info("The best is moving slot " + bestSlot + " from node "
								+ bestUsedNode + " to node " + bestUnusedNode
								+ ", with a traffic of " + bestTraffic + " tuple/s");
						bestUsedNode.remove(bestSlot);
						bestUnusedNode.assign(bestSlot);
					} else {
						logger.info("Cannot find a way to make topology " + topology
								+ " use the desired number of nodes");
						break;
					}

				} /* end while (number of used nodes < number of nodes to use */
			} /* end for (Topology topology : topologyList) */
		}

		logger.info("Second phase completed!");
		if (TrafficManager.getInstance().getAssignments() != null) {
			logger.info("Final assignment: "
					+ Utils.collectionToString(TrafficManager.getInstance().getAssignments()
							.keySet()));
		}
	}

}