import backtype.storm.scheduler.Cluster; //导入依赖的package包/类
private Cluster getSpyCluster(int numWorkers, int numExecutors) {
  Cluster spyCluster = getSpyCluster();

  List<MesosWorkerSlot> mesosWorkerSlots = this.generateMesosWorkerSlots(numWorkers);
  initializeMesosWorkerSlotMap(mesosWorkerSlots);

  Set<ExecutorDetails> executorsToAssign = this.generateExecutorDetailsSet(numExecutors);
  List<WorkerSlot> workerSlotList = getWorkerSlotFromMesosWorkerSlot(mesosWorkerSlots);
  topologyMap.put(sampleTopologyId, TestUtils.constructTopologyDetails(sampleTopologyId, numWorkers, 0.1, 100));
  this.topologies = new Topologies(topologyMap);

  doReturn(workerSlotList).when(spyCluster).getAvailableSlots();
  doReturn(executorsToAssign).when(spyCluster).getUnassignedExecutors(any(TopologyDetails.class));

  return spyCluster;
}
 

import backtype.storm.scheduler.Cluster; //导入依赖的package包/类
@Test
public void testScheduleWithOneWorkerSlot() {
  Cluster spyCluster = getSpyCluster();

  List<MesosWorkerSlot> mesosWorkerSlots = this.generateMesosWorkerSlots(1);
  initializeMesosWorkerSlotMap(mesosWorkerSlots);

  Set<ExecutorDetails> executorsToAssign = this.generateExecutorDetailsSet(4);
  List<WorkerSlot> workerSlotList = getWorkerSlotFromMesosWorkerSlot(mesosWorkerSlots);

  doReturn(workerSlotList).when(spyCluster).getAvailableSlots();
  doReturn(executorsToAssign).when(spyCluster).getUnassignedExecutors(any(TopologyDetails.class));

  defaultScheduler.schedule(topologies, spyCluster);

  Set<ExecutorDetails> assignedExecutors = spyCluster.getAssignmentById(sampleTopologyId).getExecutors();
  assertEquals(executorsToAssign, assignedExecutors);
}
 

import backtype.storm.scheduler.Cluster; //导入依赖的package包/类
@Override
public void schedule(Topologies topologies, Cluster cluster) {
	
	logger.info("Online Scheduler");
	logger.info("+++++++++++++++++++++++++++");
	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);
		else
			logger.info("It's not time to reschedule yet, " + elapsedTime + " seconds have passed, other " + (rescheduleTimeout - elapsedTime) + " seconds have to pass");
	}

	logger.info("---------------------------");
	
	logger.info("Calling EvenScheduler to schedule remaining executors...");
	new EvenScheduler().schedule(topologies, cluster);
	logger.info("Ok, EvenScheduler succesfully called");
	
	assignmentTracker.checkAssignment(topologies, cluster);
}
 

import backtype.storm.scheduler.Cluster; //导入依赖的package包/类
/**
 * Frees a single slot in this node
 * 
 * @param ws the slot to free
 * @param cluster the cluster to update
 */
public void free(WorkerSlot ws, Cluster cluster, boolean forceFree) {
    if (_freeSlots.contains(ws))
        return;
    boolean wasFound = false;
    for (Entry<String, Set<WorkerSlot>> entry : _topIdToUsedSlots.entrySet()) {
        Set<WorkerSlot> slots = entry.getValue();
        if (slots.remove(ws)) {
            cluster.freeSlot(ws);
            if (_isAlive) {
                _freeSlots.add(ws);
            }
            wasFound = true;
        }
    }
    if (!wasFound) {
        if (forceFree) {
            LOG.info("Forcefully freeing the " + ws);
            cluster.freeSlot(ws);
            _freeSlots.add(ws);
        } else {
            throw new IllegalArgumentException("Tried to free a slot that was not" + " part of this node " + _nodeId);
        }
    }
}
 

import backtype.storm.scheduler.Cluster; //导入依赖的package包/类
/**
 * Assign a free slot on the node to the following topology and executors. This will update the cluster too.
 * 
 * @param topId the topology to assign a free slot to.
 * @param executors the executors to run in that slot.
 * @param cluster the cluster to be updated
 */
public void assign(String topId, Collection<ExecutorDetails> executors, Cluster cluster) {
    if (!_isAlive) {
        throw new IllegalStateException("Trying to adding to a dead node " + _nodeId);
    }
    if (_freeSlots.isEmpty()) {
        throw new IllegalStateException("Trying to assign to a full node " + _nodeId);
    }
    if (executors.size() == 0) {
        LOG.warn("Trying to assign nothing from " + topId + " to " + _nodeId + " (Ignored)");
    } else {
        WorkerSlot slot = _freeSlots.iterator().next();
        cluster.assign(slot, topId, executors);
        assignInternal(slot, topId, false);
    }
}
 

import backtype.storm.scheduler.Cluster; //导入依赖的package包/类
/**
 * Frees a single slot in this node
 *
 * @param ws      the slot to free
 * @param cluster the cluster to update
 */
public void free(WorkerSlot ws, Cluster cluster, boolean forceFree) {
    if (_freeSlots.contains(ws))
        return;
    boolean wasFound = false;
    for (Entry<String, Set<WorkerSlot>> entry : _topIdToUsedSlots.entrySet()) {
        Set<WorkerSlot> slots = entry.getValue();
        if (slots.remove(ws)) {
            cluster.freeSlot(ws);
            if (_isAlive) {
                _freeSlots.add(ws);
            }
            wasFound = true;
        }
    }
    if (!wasFound) {
        if (forceFree) {
            LOG.info("Forcefully freeing the " + ws);
            cluster.freeSlot(ws);
            _freeSlots.add(ws);
        } else {
            throw new IllegalArgumentException("Tried to free a slot that was not" + " part of this node " + _nodeId);
        }
    }
}
 

import backtype.storm.scheduler.Cluster; //导入依赖的package包/类
/**
 * Assign a free slot on the node to the following topology and executors. This will update the cluster too.
 *
 * @param topId     the topology to assign a free slot to.
 * @param executors the executors to run in that slot.
 * @param cluster   the cluster to be updated
 */
public void assign(String topId, Collection<ExecutorDetails> executors, Cluster cluster) {
    if (!_isAlive) {
        throw new IllegalStateException("Trying to adding to a dead node " + _nodeId);
    }
    if (_freeSlots.isEmpty()) {
        throw new IllegalStateException("Trying to assign to a full node " + _nodeId);
    }
    if (executors.size() == 0) {
        LOG.warn("Trying to assign nothing from " + topId + " to " + _nodeId + " (Ignored)");
    } else {
        WorkerSlot slot = _freeSlots.iterator().next();
        cluster.assign(slot, topId, executors);
        assignInternal(slot, topId, false);
    }
}
 

import backtype.storm.scheduler.Cluster; //导入依赖的package包/类
/**
 * Schedule function looks in the "mesosWorkerSlotMap" to determine which topology owns the particular
 * WorkerSlot and assigns the executors accordingly.
 */
@Override
public void schedule(Topologies topologies, Cluster cluster) {
  List<WorkerSlot> workerSlots = cluster.getAvailableSlots();
  Map<String, List<MesosWorkerSlot>> perTopologySlotList = getMesosWorkerSlotPerTopology(workerSlots);

  // So far we know how many MesosSlots each of the topologies have got. Lets assign executors for each of them
  for (String topologyId : perTopologySlotList.keySet()) {
    TopologyDetails topologyDetails = topologies.getById(topologyId);
    List<MesosWorkerSlot> mesosWorkerSlots = perTopologySlotList.get(topologyId);

    int countSlotsRequested = topologyDetails.getNumWorkers();
    int countSlotsAssigned = cluster.getAssignedNumWorkers(topologyDetails);

    if (mesosWorkerSlots.size() == 0) {
      log.warn("No slots found for topology {} while scheduling", topologyId);
      continue;
    }

    int countSlotsAvailable = Math.min(mesosWorkerSlots.size(), (countSlotsRequested - countSlotsAssigned));

    List<List<ExecutorDetails>> executorsPerWorkerList = executorsPerWorkerList(cluster, topologyDetails, countSlotsAvailable);

    for (int i = 0; i < countSlotsAvailable; i++) {
      cluster.assign(mesosWorkerSlots.remove(0), topologyId, executorsPerWorkerList.remove(0));
    }
  }
  mesosWorkerSlotMap.clear();
}
 

import backtype.storm.scheduler.Cluster; //导入依赖的package包/类
@Test
public void testScheduleWithMultipleSlotsOnSameHost() {
  Cluster spyCluster = this.getSpyCluster(3, 3);
  defaultScheduler.schedule(topologies, spyCluster);
  SchedulerAssignment schedulerAssignment = spyCluster.getAssignments()
                                                       .get(sampleTopologyId);
  Map<ExecutorDetails, WorkerSlot> executorDetailsWorkerSlotMap = schedulerAssignment.getExecutorToSlot();
  /* We expect the three unassigned executors to be spread
     across the three available worker slots */
  assertEquals(executorDetailsWorkerSlotMap.keySet().size(), 3);
  assertEquals(executorDetailsWorkerSlotMap.values().size(), 3);

  spyCluster = this.getSpyCluster(3, 6);
  defaultScheduler.schedule(topologies, spyCluster);
  executorDetailsWorkerSlotMap = spyCluster.getAssignments()
                                            .get(sampleTopologyId)
                                            .getExecutorToSlot();
  /* We expect all executors to be scheduled across the three
     available slots */
  assertEquals(executorDetailsWorkerSlotMap.keySet().size(), 6);
  int workerSlotsUsed = new HashSet<>(executorDetailsWorkerSlotMap.values()).size() ;
  assertEquals(workerSlotsUsed, 3);

  /* Lets make sure that the executors are evenly spread
     across the worker slots in a round robin fashion */
  schedulerAssignment = spyCluster.getAssignments()
                                                       .get(sampleTopologyId);
  executorDetailsWorkerSlotMap = schedulerAssignment.getExecutorToSlot();

  Map<WorkerSlot, List<ExecutorDetails>> workerSlotExecutorDetailsMap = this.getworkerSlotExecutorDetailsMap(executorDetailsWorkerSlotMap);

  for (WorkerSlot workerSlot : workerSlotExecutorDetailsMap.keySet()) {
    List<ExecutorDetails> executorDetails = workerSlotExecutorDetailsMap.get(workerSlot);
    assertEquals(3, Math.abs(executorDetails.get(0).getStartTask() - executorDetails.get(1).getEndTask()));
  }
}
 

import backtype.storm.scheduler.Cluster; //导入依赖的package包/类
@Override
public void init(Cluster cluster, Map<String, Node> nodeIdToNode) {
    super.init(cluster, nodeIdToNode);
    for (Node n : nodeIdToNode.values()) {
        if (n.isTotallyFree() && n.isAlive()) {
            _nodes.add(n);
            _totalSlots += n.totalSlotsFree();
        }
    }
    LOG.debug("Found {} nodes with {} slots", _nodes.size(), _totalSlots);
}
 

import backtype.storm.scheduler.Cluster; //导入依赖的package包/类
/**
 * Free all slots on this node. This will update the Cluster too.
 * 
 * @param cluster the cluster to be updated
 */
public void freeAllSlots(Cluster cluster) {
    if (!_isAlive) {
        LOG.warn("Freeing all slots on a dead node {} ", _nodeId);
    }
    for (Entry<String, Set<WorkerSlot>> entry : _topIdToUsedSlots.entrySet()) {
        cluster.freeSlots(entry.getValue());
        if (_isAlive) {
            _freeSlots.addAll(entry.getValue());
        }
    }
    _topIdToUsedSlots = new HashMap<String, Set<WorkerSlot>>();
}
 

import backtype.storm.scheduler.Cluster; //导入依赖的package包/类
/**
 * Frees all the slots for a topology.
 * 
 * @param topId the topology to free slots for
 * @param cluster the cluster to update
 */
public void freeTopology(String topId, Cluster cluster) {
    Set<WorkerSlot> slots = _topIdToUsedSlots.get(topId);
    if (slots == null || slots.isEmpty())
        return;
    for (WorkerSlot ws : slots) {
        cluster.freeSlot(ws);
        if (_isAlive) {
            _freeSlots.add(ws);
        }
    }
    _topIdToUsedSlots.remove(topId);
}
 

import backtype.storm.scheduler.Cluster; //导入依赖的package包/类
/**
 * Free all slots on this node. This will update the Cluster too.
 *
 * @param cluster the cluster to be updated
 */
public void freeAllSlots(Cluster cluster) {
    if (!_isAlive) {
        LOG.warn("Freeing all slots on a dead node {} ", _nodeId);
    }
    for (Entry<String, Set<WorkerSlot>> entry : _topIdToUsedSlots.entrySet()) {
        cluster.freeSlots(entry.getValue());
        if (_isAlive) {
            _freeSlots.addAll(entry.getValue());
        }
    }
    _topIdToUsedSlots = new HashMap<>();
}
 

import backtype.storm.scheduler.Cluster; //导入依赖的package包/类
/**
 * Frees all the slots for a topology.
 *
 * @param topId   the topology to free slots for
 * @param cluster the cluster to update
 */
public void freeTopology(String topId, Cluster cluster) {
    Set<WorkerSlot> slots = _topIdToUsedSlots.get(topId);
    if (slots == null || slots.isEmpty())
        return;
    for (WorkerSlot ws : slots) {
        cluster.freeSlot(ws);
        if (_isAlive) {
            _freeSlots.add(ws);
        }
    }
    _topIdToUsedSlots.remove(topId);
}
 

import backtype.storm.scheduler.Cluster; //导入依赖的package包/类
@Override
public void schedule(Topologies topologies, Cluster cluster) {
	evenScheduler.schedule(topologies, cluster);
	assignmentTracker.checkAssignment(topologies, cluster);
}
 

import backtype.storm.scheduler.Cluster; //导入依赖的package包/类
@Override
public void schedule(Topologies topologies, Cluster cluster) {
    LOG.trace("scheduling topologies...");



    this.evenScheduler.schedule(topologies, cluster);
}
 

import backtype.storm.scheduler.Cluster; //导入依赖的package包/类
/**
 * Initialize the pool.
 * 
 * @param cluster the cluster
 * @param nodeIdToNode the mapping of node id to nodes
 */
public void init(Cluster cluster, Map<String, Node> nodeIdToNode) {
    _cluster = cluster;
    _nodeIdToNode = nodeIdToNode;
}
 

import backtype.storm.scheduler.Cluster; //导入依赖的package包/类
/**
 * Initialize the pool.
 *
 * @param cluster      the cluster
 * @param nodeIdToNode the mapping of node id to nodes
 */
public void init(Cluster cluster, Map<String, Node> nodeIdToNode) {
    _cluster = cluster;
    _nodeIdToNode = nodeIdToNode;
}