public void resourceOffers(SchedulerDriver d, List<Offer> offers) {
    try {
      synchronized(jobTracker) {

int numOffers = (int) offers.size();
double[] cpus = new double[numOffers];
double[] mem = new double[numOffers];

// Count up the amount of free CPUs and memory on each node
for (int i = 0; i < numOffers; i++) {
Offer offer = offers.get(i);
LOG.info("Got resource offer " + offer.getId());
cpus[i] = getResource(offer, "cpus");
mem[i] = getResource(offer, "mem");
}

// Assign tasks to the nodes in a round-robin manner, and stop when we
// are unable to assign a task to any node.
// We do this by keeping a linked list of indices of nodes for which
// we are still considering assigning tasks. Whenever we can't find a
// new task for a node, we remove it from the list. When the list is
// empty, no further assignments can be made. This algorithm was chosen
// because it minimizing the amount of scanning we need to do if we
// get a large set of offered nodes.
List<Integer> indices = new LinkedList<Integer>();
List<List<TaskInfo>> replies =
new ArrayList<List<TaskInfo>>(numOffers);
for (int i = 0; i < numOffers; i++) {
indices.add(i);
replies.add(new ArrayList<TaskInfo>());
}
while (indices.size() > 0) {
for (Iterator<Integer> it = indices.iterator(); it.hasNext();) {
int i = it.next();
Offer offer = offers.get(i);
TaskInfo task = findTask(
offer.getSlaveId(), offer.getHostname(), cpus[i], mem[i]);
if (task != null) {
cpus[i] -= getResource(task, "cpus");
mem[i] -= getResource(task, "mem");
replies.get(i).add(task);
} else {
it.remove();
}
}
}

for (int i = 0; i < numOffers; i++) {
OfferID offerId = offers.get(i).getId();
Status status = d.launchTasks(offerId, replies.get(i));
if (status != Status.DRIVER_RUNNING) {
LOG.warn("SchedulerDriver returned irregular status: " + status);
}
}
}
} catch(Exception e) {
LOG.error("Error in resourceOffer", e);
}
}

// Find a single task for a given node. Assumes JobTracker is locked.
  private TaskInfo findTask(
      SlaveID slaveId, String host, double cpus, double mem) {
    if (cpus < cpusPerTask || mem < memPerTask) {
      return null; // Too few resources are left on the node
    }

TaskTrackerInfo ttInfo = getTaskTrackerInfo(host, slaveId);

// Pick whether to launch a map or a reduce based on available tasks
String taskType = null;
boolean haveMaps = canLaunchMap(host);
boolean haveReduces = canLaunchReduce(host);
//LOG.info("Looking at " + host + ": haveMaps=" + haveMaps +
// ", haveReduces=" + haveReduces);
if (!haveMaps && !haveReduces) {
return null;
} else if (haveMaps && !haveReduces) {
taskType = "map";
} else if (haveReduces && !haveMaps) {
taskType = "reduce";
} else {
float mapToReduceRatio = 1;
if (ttInfo.reduces.size() < ttInfo.maps.size() / mapToReduceRatio)
taskType = "reduce";
else
taskType = "map";
}
//LOG.info("Task type chosen: " + taskType);

// Get a Mesos task ID for the new task
TaskID mesosId = newMesosTaskId();

// Remember that it is launched
boolean isMap = taskType.equals("map");
if (isMap) {
unassignedMaps++;
} else {
unassignedReduces++;
}
MesosTask nt = new MesosTask(isMap, mesosId, host);
mesosIdToMesosTask.put(mesosId, nt);
ttInfo.add(nt);

LOG.info("Launching Mesos task " + mesosId.getValue() +
" as " + taskType + " on " + host);

// Create a task description to pass back to Mesos.
return TaskInfo.newBuilder()
.setTaskId(mesosId)
.setSlaveId(slaveId)
.setName("task " + mesosId.getValue() + " (" + taskType + ")")
.addResources(makeResource("cpus", cpusPerTask))
.addResources(makeResource("mem", memPerTask))
.setExecutor(getExecutorInfo())
.build();
}

private boolean canLaunchMap(String host) {
    // Check whether the TT is saturated on maps
    TaskTrackerInfo ttInfo = ttInfos.get(host);
    if (ttInfo == null) {
      throw new RuntimeException("Expecting TaskTrackerInfo for host " + host);
    }

if (ttInfo.maps.size() >= ttInfo.maxMaps) {
return false;
}

// Compute the total demand for maps to make sure we don't exceed it
Collection<JobInProgress> jobs = jobTracker.jobs.values();
int neededMaps = 0;
for (JobInProgress job : jobs) {
if (job.getStatus().getRunState() == JobStatus.RUNNING) {
neededMaps += job.pendingMaps();
}
}
// TODO (!!!): Count speculatable tasks and add them to neededMaps
// For now, we just add 1
if (jobs.size() > 0)
neededMaps += 1;

if (unassignedMaps < neededMaps) {
// 0. check for a failed map task to place. These tasks are not included
// in the "normal" lists of tasks in the JobInProgress object.
for (JobInProgress job: jobs) {
int state = job.getStatus().getRunState();
if (job.failedMaps != null && state == JobStatus.RUNNING) {
for (TaskInProgress tip : job.failedMaps) {
if (!tip.hasFailedOnMachine(host)) {
return true;
}
}
}
}

int maxLevel = Integer.MAX_VALUE;
// Look for a map with the required level
for (JobInProgress job: jobs) {
int state = job.getStatus().getRunState();
if (state == JobStatus.RUNNING) {
int availLevel = availableMapLevel(job, host, maxLevel);
if (availLevel != -1) {
lastMapWasLocal = (availLevel == 0);
return true;
}
}
}
}

private static class TaskTrackerInfo {
    SlaveID mesosSlaveId;
    List<MesosTask> maps = new LinkedList<MesosTask>();
    List<MesosTask> reduces = new LinkedList<MesosTask>();
    int maxMaps = 1;
    int maxReduces = 1;

public TaskTrackerInfo(SlaveID mesosSlaveId) {
this.mesosSlaveId = mesosSlaveId;
}

void add(MesosTask nt) {
if (nt.isMap)
maps.add(nt);
else
reduces.add(nt);
}

public void remove(MesosTask nt) {
if (nt.isMap)
maps.remove(nt);
else
reduces.remove(nt);
}

void Master::launchTasks(const FrameworkID& frameworkId,
                         const OfferID& offerId,
                         const vector<TaskInfo>& tasks,
                         const Filters& filters)
{
  Framework* framework = getFramework(frameworkId);
  if (framework != NULL) {
    // TODO(benh): Support offer "hoarding" and allow multiple offers
    // *from the same slave* to be used to launch tasks. This can be
    // accomplished rather easily by collecting and merging all offers
    // into a mega-offer and passing that offer to
    // Master::processTasks.
    Offer* offer = getOffer(offerId);
    if (offer != NULL) {
      CHECK(offer->framework_id() == frameworkId);
      Slave* slave = getSlave(offer->slave_id());
      CHECK(slave != NULL) << "An offer should not outlive a slave!";
      processTasks(offer, framework, slave, tasks, filters);
    } else {
      // The offer is gone (possibly rescinded, lost slave, re-reply
      // to same offer, etc). Report all tasks in it as failed.
      // TODO: Consider adding a new task state TASK_INVALID for
      // situations like these.
      LOG(WARNING) << "Offer " << offerId << " is no longer valid";
      foreach (const TaskInfo& task, tasks) {
        StatusUpdateMessage message;
        StatusUpdate* update = message.mutable_update();
        update->mutable_framework_id()->MergeFrom(frameworkId);
        TaskStatus* status = update->mutable_status();
        status->mutable_task_id()->MergeFrom(task.task_id());
        status->set_state(TASK_LOST);
        status->set_message("Task launched with invalid offer");
        update->set_timestamp(Clock::now());
        update->set_uuid(UUID::random().toBytes());
        send(framework->pid, message);
      }
    }
  }
}