import backtype.storm.scheduler.SupervisorDetails; //导入依赖的package包/类
/**
 * Method checks if all topologies that need assignment already have supervisor running on the node where the Offer
 * comes from. Required for more accurate available resource calculation where we can exclude supervisor's demand from
 * the Offer.
 * Unfortunately because of WorkerSlot type is not topology agnostic, we need to exclude supervisor's resources only
 * in case where ALL topologies in 'allSlotsAvailableForScheduling' method satisfy condition of supervisor existence
 *
 * @param offer                        Offer
 * @param existingSupervisors          Supervisors which already placed on the node for the Offer
 * @param topologiesMissingAssignments Topology ids required assignment
 * @return Boolean value indicating supervisor existence
 */
private boolean supervisorExists(
    Offer offer, Collection<SupervisorDetails> existingSupervisors, Set<String> topologiesMissingAssignments) {
  boolean alreadyExists = true;
  for (String topologyId : topologiesMissingAssignments) {
    String offerHost = offer.getHostname();
    boolean exists = false;
    for (SupervisorDetails d : existingSupervisors) {
      if (d.getId().equals(MesosCommon.supervisorId(offerHost, topologyId))) {
        exists = true;
      }
    }
    alreadyExists = (alreadyExists && exists);
  }
  return alreadyExists;
}
 

import backtype.storm.scheduler.SupervisorDetails; //导入依赖的package包/类
@Before
public void initialize() {
  defaultScheduler = new DefaultScheduler();
  Map<String, Object> mesosStormConf = new HashMap<>();
  defaultScheduler.prepare(mesosStormConf);

  rotatingMap = new RotatingMap<>(2);

  topologiesMissingAssignments = new HashSet<>();
  topologiesMissingAssignments.add("test-topology1-65-1442255385");
  topologiesMissingAssignments.add("test-topology1-65-1442255385");

  existingSupervisors = new ArrayList<>();
  existingSupervisors.add(new SupervisorDetails(MesosCommon.supervisorId(sampleHost, "test-topology1-65-1442255385"), sampleHost, null, null));
  existingSupervisors.add(new SupervisorDetails(MesosCommon.supervisorId(sampleHost, "test-topology10-65-1442255385"), sampleHost, null, null));

  topologyMap = new HashMap<>();
  topologyMap.put(sampleTopologyId, TestUtils.constructTopologyDetails(sampleTopologyId, 1, 0.1, 100));
  topologies = new Topologies(topologyMap);

  mesosWorkerSlotMap = new HashMap<>();
}
 

import backtype.storm.scheduler.SupervisorDetails; //导入依赖的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.SupervisorDetails; //导入依赖的package包/类
private Cluster getSpyCluster() {
  Map<String, SupervisorDetails> supervisors = new HashMap<>();
  Map<String, SchedulerAssignmentImpl> assignmentMap = new HashMap<>();

  for (SupervisorDetails supervisorDetails : existingSupervisors) {
    String nodeId = supervisorDetails.getHost();
    supervisors.put(nodeId, supervisorDetails);
  }

  return spy(new Cluster(null, supervisors, assignmentMap));
}
 

import backtype.storm.scheduler.SupervisorDetails; //导入依赖的package包/类
@Test
public void testSupervisorExists() throws Exception {
  Collection<SupervisorDetails> existingSupervisors = new ArrayList<>();
  String hostName = "host1.east";

  existingSupervisors.add(new SupervisorDetails(MesosCommon.supervisorId(hostName, "test-topology1-65-1442255385"), hostName, null));
  existingSupervisors.add(new SupervisorDetails(MesosCommon.supervisorId(hostName, "test-topology10-65-1442255385"), hostName, null));

  assertEquals(true, SchedulerUtils.supervisorExists(hostName, existingSupervisors, "test-topology1-65-1442255385"));
  assertEquals(false, SchedulerUtils.supervisorExists(hostName, existingSupervisors, "test-topology2-65-1442255385"));
}
 

import backtype.storm.scheduler.SupervisorDetails; //导入依赖的package包/类
@Override
public Collection<WorkerSlot> allSlotsAvailableForScheduling(
		Collection<SupervisorDetails> existingSupervisors,
		Topologies topologies, Set<String> topologiesMissingAssignments) {
	Collection<WorkerSlot> result = new HashSet<WorkerSlot>();
	for (SupervisorDetails detail : existingSupervisors) {
		for (Integer port : detail.getAllPorts())
			result.add(new WorkerSlot(detail.getId(), port));
	}
	return result;
}
 

import backtype.storm.scheduler.SupervisorDetails; //导入依赖的package包/类
@Override
public Collection<WorkerSlot> allSlotsAvailableForScheduling(
		Collection<SupervisorDetails> existingSupervisors,
		Topologies topologies, Set<String> topologiesMissingAssignments) {
	// TODO Auto-generated method stub
	Collection<WorkerSlot> result = new HashSet<WorkerSlot>();
	for (SupervisorDetails detail : existingSupervisors) {
		for (Integer port : detail.getAllPorts())
			result.add(new WorkerSlot(detail.getId(), port));
	}
	return result;
}
 

import backtype.storm.scheduler.SupervisorDetails; //导入依赖的package包/类
@Override
public Collection<WorkerSlot> allSlotsAvailableForScheduling(Collection<SupervisorDetails> existingSupervisors, Topologies topologies,
        Set<String> topologiesMissingAssignments) {
    // TODO Auto-generated method stub
    Collection<WorkerSlot> result = new HashSet<WorkerSlot>();
    for (SupervisorDetails detail : existingSupervisors) {
        for (Integer port : detail.getAllPorts())
            result.add(new WorkerSlot(detail.getId(), port));
    }
    return result;
}
 

import backtype.storm.scheduler.SupervisorDetails; //导入依赖的package包/类
@Override
public Collection<WorkerSlot> allSlotsAvailableForScheduling(Collection<SupervisorDetails> existingSupervisors,
                                                             Topologies topologies,
                                                             Set<String> topologiesMissingAssignments) {
    Collection<WorkerSlot> result = new HashSet<>();
    for (SupervisorDetails detail : existingSupervisors) {
        for (Integer port : detail.getAllPorts())
            result.add(new WorkerSlot(detail.getId(), port));
    }
    return result;
}
 

import backtype.storm.scheduler.SupervisorDetails; //导入依赖的package包/类
@Override
public String getHostName(Map<String, SupervisorDetails> map, String nodeId) {
  return nodeId;
}
 

import backtype.storm.scheduler.SupervisorDetails; //导入依赖的package包/类
@Override
public String getHostName(
		Map<String, SupervisorDetails> existingSupervisors, String nodeId) {
	return null;
}
 

import backtype.storm.scheduler.SupervisorDetails; //导入依赖的package包/类
@Override
public String getHostName(
		Map<String, SupervisorDetails> existingSupervisors, String nodeId) {
	// TODO Auto-generated method stub
	return null;
}
 

import backtype.storm.scheduler.SupervisorDetails; //导入依赖的package包/类
@Override
public String getHostName(Map<String, SupervisorDetails> existingSupervisors, String nodeId) {
    // TODO Auto-generated method stub
    return null;
}
 

import backtype.storm.scheduler.SupervisorDetails; //导入依赖的package包/类
@Override
public String getHostName(Map<String, SupervisorDetails> map, String nodeId) {
    return nodeId;
}
 

import backtype.storm.scheduler.SupervisorDetails; //导入依赖的package包/类
@Override
public Collection<WorkerSlot> allSlotsAvailableForScheduling(
        Collection<SupervisorDetails> existingSupervisors, Topologies topologies, Set<String> topologiesMissingAssignments) {
    synchronized (OFFERS_LOCK) {
        LOG.info("Currently have " + _offers.size() + " offers buffered");
        if (!topologiesMissingAssignments.isEmpty()) {
            LOG.info("Topologies that need assignments: " + topologiesMissingAssignments.toString());
        } else {
            LOG.info("Declining offers because no topologies need assignments");
            _offers.clear();
        }
    }

    Double cpu = null;
    Double mem = null;
    // TODO: maybe this isn't the best approach. if a topology raises #cpus keeps failing,
    // it will mess up scheduling on this cluster permanently
    for (String id : topologiesMissingAssignments) {
        TopologyDetails details = topologies.getById(id);
        double tcpu = MesosCommon.topologyCpu(_conf, details);
        double tmem = MesosCommon.topologyMem(_conf, details);
        if (cpu == null || tcpu > cpu) {
            cpu = tcpu;
        }
        if (mem == null || tmem > mem) {
            mem = tmem;
        }
    }

    // need access to how many slots are currently used to limit number of slots taken up
    List<WorkerSlot> allSlots = new ArrayList();

    if (cpu != null && mem != null) {
        synchronized (OFFERS_LOCK) {
            for (Offer offer : _offers.values()) {
                allSlots.addAll(toSlots(offer, cpu, mem));
            }
        }
    }

    LOG.info("Number of available slots: " + allSlots.size());
    return allSlots;
}
 

import backtype.storm.scheduler.SupervisorDetails; //导入依赖的package包/类
@Override
public String getHostName(Map<String, SupervisorDetails> existingSupervisors, String nodeId) {
    return null;
}
 

import backtype.storm.scheduler.SupervisorDetails; //导入依赖的package包/类
/**
 * This method is invoked by Nimbus when it wants to get a list of worker slots that are available for assigning the
 * topology workers. In Nimbus's view, a "WorkerSlot" is a host and port that it can use to assign a worker.
 * <p/>
 * TODO: Rotating Map itself needs to be refactored. Perhaps make RotatingMap inherit Map so users can pass in a
 * map? or Make the IMesosStormScheduler itself generic so _offers could be of any type?
 */
public List<WorkerSlot> allSlotsAvailableForScheduling(RotatingMap<Protos.OfferID, Protos.Offer> offers,
                                                       Collection<SupervisorDetails> existingSupervisors,
                                                       Topologies topologies, Set<String> topologiesMissingAssignments);