import backtype.storm.scheduler.WorkerSlot; //导入依赖的package包/类
Map<String, List<MesosWorkerSlot>> getMesosWorkerSlotPerTopology(List<WorkerSlot> workerSlots) {
  HashMap<String, List<MesosWorkerSlot>> perTopologySlotList = new HashMap<>();

  for (WorkerSlot workerSlot : workerSlots) {
    if (workerSlot.getNodeId() == null) {
      log.warn("Unexpected: Node id is null for worker slot while scheduling");
      continue;
    }
    MesosWorkerSlot mesosWorkerSlot = mesosWorkerSlotMap.get(workerSlot.getNodeId() +
                                                             ":" + String.valueOf(workerSlot.getPort()));

    String topologyId = mesosWorkerSlot.getTopologyId();
    if (perTopologySlotList.get(topologyId) == null) {
      perTopologySlotList.put(topologyId, new ArrayList<MesosWorkerSlot>());
    }
    perTopologySlotList.get(topologyId).add(mesosWorkerSlot);
  }

  return  perTopologySlotList;
}
 

import backtype.storm.scheduler.WorkerSlot; //导入依赖的package包/类
@Override
public void assignSlots(Topologies topologies, Map<String, Collection<WorkerSlot>> slots) {
  if (slots.size() == 0) {
    LOG.debug("assignSlots: no slots passed in, nothing to do");
    return;
  }
  for (Map.Entry<String, Collection<WorkerSlot>> topologyToSlots : slots.entrySet()) {
    String topologyId = topologyToSlots.getKey();
    for (WorkerSlot slot : topologyToSlots.getValue()) {
      TopologyDetails details = topologies.getById(topologyId);
      LOG.debug("assignSlots: topologyId: {} worker being assigned to slot: {} with workerCpu: {} workerMem: {}",
                topologyId, slot, MesosCommon.topologyWorkerCpu(_conf, details), MesosCommon.topologyWorkerMem(_conf, details));
    }
  }
  synchronized (_offersLock) {
    computeLaunchList(topologies, slots);
  }
}
 

import backtype.storm.scheduler.WorkerSlot; //导入依赖的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.WorkerSlot; //导入依赖的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.WorkerSlot; //导入依赖的package包/类
private String mapToString(Map<String, List<WorkerSlot>> map) {
	if (map == null)
		return "null";
	if (map.keySet().isEmpty())
		return "<empty map>";
	StringBuffer sb = new  StringBuffer();
	int i = 0;
	for (Object key : map.keySet()) {
		sb.append(key.toString() + " -> (");
		sb.append(collectionToString(map.get(key)) + ")");
		if (i < map.keySet().size() - 1)
			sb.append(", ");
		i++;
	}
	return sb.toString();
}
 

import backtype.storm.scheduler.WorkerSlot; //导入依赖的package包/类
private boolean isOutTaskConnected(int taskId) {
    boolean ret = false;

    if (workerData.getInnerTaskTransfer().get(taskId) != null) {
        // Connections to inner tasks should be done after initialization. 
        // So return true here for all inner tasks.
        ret = true;
    } else {
        WorkerSlot slot = taskNodeport.get(taskId);
        if (slot != null) {
            IConnection connection = nodeportSocket.get(slot);
            if (connection != null) {
                ret = connection.available();
            }
        }
    }

    return ret;
}
 

import backtype.storm.scheduler.WorkerSlot; //导入依赖的package包/类
boolean assignInternal(WorkerSlot ws, String topId, boolean dontThrow) {
    validateSlot(ws);
    if (!_freeSlots.remove(ws)) {
        for (Entry<String, Set<WorkerSlot>> topologySetEntry : _topIdToUsedSlots.entrySet()) {
            if (topologySetEntry.getValue().contains(ws)) {
                if (dontThrow) {
                    LOG.warn("Worker slot [" + ws + "] can't be assigned to " + topId + ". Its already assigned to " + topologySetEntry.getKey() + ".");
                    return true;
                }
                throw new IllegalStateException("Worker slot [" + ws + "] can't be assigned to " + topId + ". Its already assigned to "
                        + topologySetEntry.getKey() + ".");
            }
        }
        LOG.warn("Adding Worker slot [" + ws + "] that was not reported in the supervisor heartbeats," + " but the worker is already running for topology "
                + topId + ".");
    }
    Set<WorkerSlot> usedSlots = _topIdToUsedSlots.get(topId);
    if (usedSlots == null) {
        usedSlots = new HashSet<WorkerSlot>();
        _topIdToUsedSlots.put(topId, usedSlots);
    }
    usedSlots.add(ws);
    return false;
}
 

import backtype.storm.scheduler.WorkerSlot; //导入依赖的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.WorkerSlot; //导入依赖的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.WorkerSlot; //导入依赖的package包/类
@Override
public void addTopology(TopologyDetails td) {
    String topId = td.getId();
    LOG.debug("Adding in Topology {}", topId);
    SchedulerAssignment assignment = _cluster.getAssignmentById(topId);
    Set<Node> assignedNodes = new HashSet<Node>();
    if (assignment != null) {
        for (WorkerSlot ws : assignment.getSlots()) {
            Node n = _nodeIdToNode.get(ws.getNodeId());
            assignedNodes.add(n);
        }
    }
    _usedNodes += assignedNodes.size();
    _topologyIdToNodes.put(topId, assignedNodes);
    _tds.put(topId, td);
    if (td.getConf().get(Config.TOPOLOGY_ISOLATED_MACHINES) != null) {
        _isolated.add(topId);
    }
}
 

import backtype.storm.scheduler.WorkerSlot; //导入依赖的package包/类
private OfferID findOffer(WorkerSlot worker) {
    int port = worker.getPort();
    ArrayList<Offer> offers = new ArrayList(_offers.values());
    Collections.shuffle(offers);
    for (Offer offer : offers) {
        if (offer.getHostname().equals(worker.getNodeId())) {
            for (Resource r : offer.getResourcesList()) {
                if (r.getName().equals("ports")) {
                    for (Range range : r.getRanges().getRangeList()) {
                        if (port >= range.getBegin() && port <= range.getEnd()) {
                            return offer.getId();
                        }
                    }
                }
            }
        }
    }
    return null;
}
 

import backtype.storm.scheduler.WorkerSlot; //导入依赖的package包/类
private boolean isOutTaskConnected(int taskId) {
    boolean ret = false;

    if (workerData.getInnerTaskTransfer().get(taskId) != null) {
        // Connections to inner tasks should be done after initialization.
        // So return true here for all inner tasks.
        ret = true;
    } else {
        WorkerSlot slot = taskToNodePort.get(taskId);
        if (slot != null) {
            IConnection connection = nodePortToSocket.get(slot);
            if (connection != null) {
                ret = connection.available(taskId);
            }
        }
    }

    return ret;
}
 

import backtype.storm.scheduler.WorkerSlot; //导入依赖的package包/类
private boolean isOutboundTaskAvailable(int taskId) {
    boolean ret = false;
    DisruptorQueue targetQueue = workerData.getInnerTaskTransfer().get(taskId);

    if (targetQueue != null) {
        float queueLoadRatio = targetQueue.pctFull();
        if (queueLoadRatio < loadMark) {
            ret = true;
        }
    } else {
        WorkerSlot slot = taskNodePort.get(taskId);
        if (slot != null) {
            IConnection connection = nodePortToSocket.get(slot);
            if (connection != null) {
                ret = connection.available(taskId);
            }
        }
    }

    if (!ret) {
        LOG.debug("taskId:{} is unavailable", taskId);
    }

    return ret;
}
 

import backtype.storm.scheduler.WorkerSlot; //导入依赖的package包/类
boolean assignInternal(WorkerSlot ws, String topId, boolean dontThrow) {
    validateSlot(ws);
    if (!_freeSlots.remove(ws)) {
        for (Entry<String, Set<WorkerSlot>> topologySetEntry : _topIdToUsedSlots.entrySet()) {
            if (topologySetEntry.getValue().contains(ws)) {
                if (dontThrow) {
                    LOG.warn("Worker slot [" + ws + "] can't be assigned to " + topId + ". Its already assigned to " + topologySetEntry.getKey() + ".");
                    return true;
                }
                throw new IllegalStateException("Worker slot [" + ws + "] can't be assigned to " + topId + ". Its already assigned to "
                        + topologySetEntry.getKey() + ".");
            }
        }
        LOG.warn("Adding Worker slot [" + ws + "] that was not reported in the supervisor heartbeats," + " but the worker is already running for topology "
                + topId + ".");
    }
    Set<WorkerSlot> usedSlots = _topIdToUsedSlots.get(topId);
    if (usedSlots == null) {
        usedSlots = new HashSet<>();
        _topIdToUsedSlots.put(topId, usedSlots);
    }
    usedSlots.add(ws);
    return false;
}
 

import backtype.storm.scheduler.WorkerSlot; //导入依赖的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.WorkerSlot; //导入依赖的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.WorkerSlot; //导入依赖的package包/类
@Override
public void addTopology(TopologyDetails td) {
    String topId = td.getId();
    LOG.debug("Adding in topology {}", topId);
    SchedulerAssignment assignment = _cluster.getAssignmentById(topId);
    Set<Node> assignedNodes = new HashSet<>();
    if (assignment != null) {
        for (WorkerSlot ws : assignment.getSlots()) {
            Node n = _nodeIdToNode.get(ws.getNodeId());
            assignedNodes.add(n);
        }
    }
    _usedNodes += assignedNodes.size();
    _topologyIdToNodes.put(topId, assignedNodes);
    _tds.put(topId, td);
    if (td.getConf().get(Config.TOPOLOGY_ISOLATED_MACHINES) != null) {
        _isolated.add(topId);
    }
}
 

import backtype.storm.scheduler.WorkerSlot; //导入依赖的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.WorkerSlot; //导入依赖的package包/类
@Override
public Collection<WorkerSlot> allSlotsAvailableForScheduling(
        Collection<SupervisorDetails> existingSupervisors, Topologies topologies, Set<String> topologiesMissingAssignments) {
  synchronized (_offersLock) {
    return _mesosStormScheduler.allSlotsAvailableForScheduling(
            _offers,
            existingSupervisors,
            topologies,
            topologiesMissingAssignments);
  }
}
 

import backtype.storm.scheduler.WorkerSlot; //导入依赖的package包/类
private OfferID findOffer(WorkerSlot worker) {
  int port = worker.getPort();
  for (Offer offer : _offers.values()) {
    if (offer.getHostname().equals(worker.getNodeId())) {
      List<Resource> r = getResourcesRange(offer.getResourcesList(), port, "ports");
      if (r != null) return offer.getId();
    }
  }
  // Still haven't found the slot? Maybe it's an offer we already used.
  return null;
}