| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package org.apache.hadoop.mapred; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.Comparator; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.mapred.AbstractQueue.AbstractQueueComparator; |
| import org.apache.hadoop.mapred.JobTracker.IllegalStateException; |
| import org.apache.hadoop.mapreduce.TaskType; |
| import org.apache.hadoop.mapreduce.server.jobtracker.TaskTracker; |
| import org.apache.hadoop.util.StringUtils; |
| |
| /** |
| * A {@link TaskScheduler} that implements the requirements in HADOOP-3421 |
| * and provides a HOD-less way to share large clusters. This scheduler |
| * provides the following features: |
| * * support for queues, where a job is submitted to a queue. |
| * * Queues are assigned a fraction of the capacity of the grid (their |
| * 'capacity') in the sense that a certain capacity of resources |
| * will be at their disposal. All jobs submitted to the queues of an Org |
| * will have access to the capacity to the Org. |
| * * Free resources can be allocated to any queue beyond its |
| * capacity. |
| * * Queues optionally support job priorities (disabled by default). |
| * * Within a queue, jobs with higher priority will have access to the |
| * queue's resources before jobs with lower priority. However, once a job |
| * is running, it will not be preempted for a higher priority job. |
| * * In order to prevent one or more users from monopolizing its resources, |
| * each queue enforces a limit on the percentage of resources allocated to a |
| * user at any given time, if there is competition for them. |
| * |
| */ |
| class CapacityTaskScheduler extends TaskScheduler { |
| |
| /** quick way to get qsc object given a queue name */ |
| private Map<String, QueueSchedulingContext> queueInfoMap = |
| new HashMap<String, QueueSchedulingContext>(); |
| |
| //Root level queue . It has all the |
| //cluster capacity at its disposal. |
| //Queues declared by users would |
| //be children of this queue. |
| //CS would have handle to root. |
| private AbstractQueue root = null; |
| |
| /** |
| * This class captures scheduling information we want to display or log. |
| */ |
| private static class SchedulingDisplayInfo { |
| private String queueName; |
| CapacityTaskScheduler scheduler; |
| |
| SchedulingDisplayInfo(String queueName, CapacityTaskScheduler scheduler) { |
| this.queueName = queueName; |
| this.scheduler = scheduler; |
| } |
| |
| @Override |
| public String toString(){ |
| // note that we do not call updateContextObjects() here for performance |
| // reasons. This means that the data we print out may be slightly |
| // stale. This data is updated whenever assignTasks() is called |
| // If this doesn't happen, the data gets stale. If we see |
| // this often, we may need to detect this situation and call |
| // updateContextObjects(), or just call it each time. |
| return scheduler.getDisplayInfo(queueName); |
| } |
| } |
| |
| |
| // this class encapsulates the result of a task lookup |
| private static class TaskLookupResult { |
| |
| static enum LookUpStatus { |
| TASK_FOUND, |
| NO_TASK_FOUND, |
| TASK_FAILING_MEMORY_REQUIREMENT, |
| } |
| // constant TaskLookupResult objects. Should not be accessed directly. |
| private static final TaskLookupResult NoTaskLookupResult = |
| new TaskLookupResult(null, TaskLookupResult.LookUpStatus.NO_TASK_FOUND); |
| private static final TaskLookupResult MemFailedLookupResult = |
| new TaskLookupResult(null, |
| TaskLookupResult.LookUpStatus.TASK_FAILING_MEMORY_REQUIREMENT); |
| |
| private LookUpStatus lookUpStatus; |
| private Task task; |
| |
| // should not call this constructor directly. use static factory methods. |
| private TaskLookupResult(Task t, LookUpStatus lUStatus) { |
| this.task = t; |
| this.lookUpStatus = lUStatus; |
| } |
| |
| static TaskLookupResult getTaskFoundResult(Task t) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Returning task " + t); |
| } |
| return new TaskLookupResult(t, LookUpStatus.TASK_FOUND); |
| } |
| static TaskLookupResult getNoTaskFoundResult() { |
| return NoTaskLookupResult; |
| } |
| static TaskLookupResult getMemFailedResult() { |
| return MemFailedLookupResult; |
| } |
| |
| |
| Task getTask() { |
| return task; |
| } |
| |
| LookUpStatus getLookUpStatus() { |
| return lookUpStatus; |
| } |
| } |
| |
| /** |
| * This class handles the scheduling algorithms. |
| * The algos are the same for both Map and Reduce tasks. |
| * There may be slight variations later, in which case we can make this |
| * an abstract base class and have derived classes for Map and Reduce. |
| */ |
| private static abstract class TaskSchedulingMgr { |
| |
| /** our TaskScheduler object */ |
| protected CapacityTaskScheduler scheduler; |
| protected TaskType type = null; |
| |
| abstract Task obtainNewTask(TaskTrackerStatus taskTracker, |
| JobInProgress job) throws IOException; |
| |
| abstract int getClusterCapacity(); |
| abstract TaskSchedulingContext getTSC( |
| QueueSchedulingContext qsc); |
| /** |
| * To check if job has a speculative task on the particular tracker. |
| * |
| * @param job job to check for speculative tasks. |
| * @param tts task tracker on which speculative task would run. |
| * @return true if there is a speculative task to run on the tracker. |
| */ |
| abstract boolean hasSpeculativeTask(JobInProgress job, |
| TaskTrackerStatus tts); |
| |
| /** |
| * Comparator to sort queues. |
| * For maps, we need to sort on QueueSchedulingContext.mapTSC. For |
| * reducers, we use reduceTSC. So we'll need separate comparators. |
| */ |
| private static abstract class QueueComparator |
| implements Comparator<AbstractQueue> { |
| abstract TaskSchedulingContext getTSC( |
| QueueSchedulingContext qsi); |
| public int compare(AbstractQueue q1, AbstractQueue q2) { |
| TaskSchedulingContext t1 = getTSC(q1.getQueueSchedulingContext()); |
| TaskSchedulingContext t2 = getTSC(q2.getQueueSchedulingContext()); |
| // look at how much capacity they've filled. Treat a queue with |
| // capacity=0 equivalent to a queue running at capacity |
| double r1 = (0 == t1.getCapacity())? 1.0f: |
| (double) t1.getNumSlotsOccupied() /(double) t1.getCapacity(); |
| double r2 = (0 == t2.getCapacity())? 1.0f: |
| (double) t2.getNumSlotsOccupied() /(double) t2.getCapacity(); |
| if (r1<r2) return -1; |
| else if (r1>r2) return 1; |
| else return 0; |
| } |
| } |
| // subclass for map and reduce comparators |
| private static final class MapQueueComparator extends QueueComparator { |
| TaskSchedulingContext getTSC(QueueSchedulingContext qsi) { |
| return qsi.getMapTSC(); |
| } |
| } |
| private static final class ReduceQueueComparator extends QueueComparator { |
| TaskSchedulingContext getTSC(QueueSchedulingContext qsi) { |
| return qsi.getReduceTSC(); |
| } |
| } |
| |
| // these are our comparator instances |
| protected final static MapQueueComparator mapComparator = |
| new MapQueueComparator(); |
| protected final static ReduceQueueComparator reduceComparator = |
| new ReduceQueueComparator(); |
| // and this is the comparator to use |
| protected QueueComparator queueComparator; |
| |
| // Returns queues sorted according to the QueueComparator. |
| // Mainly for testing purposes. |
| String[] getOrderedQueues() { |
| List<AbstractQueue> queueList = getOrderedJobQueues(); |
| List<String> queues = new ArrayList<String>(queueList.size()); |
| for (AbstractQueue q : queueList) { |
| queues.add(q.getName()); |
| } |
| return queues.toArray(new String[queues.size()]); |
| } |
| |
| /** |
| * Return an ordered list of {@link JobQueue}s wrapped as |
| * {@link AbstractQueue}s. Ordering is according to {@link QueueComparator}. |
| * To reflect the true ordering of the JobQueues, the complete hierarchy is |
| * sorted such that {@link AbstractQueue}s are ordered according to their |
| * needs at each level in the hierarchy, after which only the leaf level |
| * {@link JobQueue}s are returned. |
| * |
| * @return a list of {@link JobQueue}s wrapped as {@link AbstractQueue}s |
| * sorted by their needs. |
| */ |
| List<AbstractQueue> getOrderedJobQueues() { |
| scheduler.root.sort(queueComparator); |
| return scheduler.root.getDescendentJobQueues(); |
| } |
| |
| TaskSchedulingMgr(CapacityTaskScheduler sched) { |
| scheduler = sched; |
| } |
| |
| /** |
| * Ceil of result of dividing two integers. |
| * |
| * This is *not* a utility method. |
| * Neither <code>a</code> or <code>b</code> should be negative. |
| * |
| * @param a |
| * @param b |
| * @return ceil of the result of a/b |
| */ |
| private int divideAndCeil(int a, int b) { |
| if (b != 0) { |
| return (a + (b - 1)) / b; |
| } |
| |
| LOG.info("divideAndCeil called with a=" + a + " b=" + b); |
| return 0; |
| } |
| |
| private boolean isUserOverLimit(JobInProgress j, |
| QueueSchedulingContext qsc) { |
| // what is our current capacity? It is equal to the queue-capacity if |
| // we're running below capacity. If we're running over capacity, then its |
| // #running plus slotPerTask of the job (which is the number of extra |
| // slots we're getting). |
| int currentCapacity; |
| TaskSchedulingContext tsi = getTSC(qsc); |
| if (tsi.getNumSlotsOccupied() < tsi.getCapacity()) { |
| currentCapacity = tsi.getCapacity(); |
| } |
| else { |
| currentCapacity = |
| tsi.getNumSlotsOccupied() + |
| TaskDataView.getTaskDataView(type).getSlotsPerTask(j); |
| } |
| int limit = Math.max(divideAndCeil(currentCapacity, qsc.getNumJobsByUser().size()), |
| divideAndCeil(qsc.getUlMin() * currentCapacity, 100)); |
| String user = j.getProfile().getUser(); |
| if (tsi.getNumSlotsOccupiedByUser().get(user) >= limit) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("User " + user + " is over limit, num slots occupied = " + |
| tsi.getNumSlotsOccupiedByUser().get(user) + ", limit = " + limit); |
| } |
| return true; |
| } |
| else { |
| return false; |
| } |
| } |
| |
| /* |
| * This is the central scheduling method. |
| * It tries to get a task from jobs in a single queue. |
| * Always return a TaskLookupResult object. Don't return null. |
| */ |
| private TaskLookupResult getTaskFromQueue(TaskTracker taskTracker, |
| QueueSchedulingContext qsi) |
| throws IOException { |
| TaskTrackerStatus taskTrackerStatus = taskTracker.getStatus(); |
| // we only look at jobs in the running queues, as these are the ones |
| // who have been potentially initialized |
| |
| for (JobInProgress j : |
| scheduler.jobQueuesManager.getJobQueue(qsi.getQueueName()) |
| .getRunningJobs()) { |
| // only look at jobs that can be run. We ignore jobs that haven't |
| // initialized, or have completed but haven't been removed from the |
| // running queue. |
| |
| //Check queue for maximum capacity . |
| if(areTasksInQueueOverMaxCapacity(qsi,j.getNumSlotsPerTask(type))) { |
| continue; |
| } |
| |
| if (j.getStatus().getRunState() != JobStatus.RUNNING) { |
| continue; |
| } |
| // check if the job's user is over limit |
| if (isUserOverLimit(j, qsi)) { |
| continue; |
| } |
| //If this job meets memory requirements. Ask the JobInProgress for |
| //a task to be scheduled on the task tracker. |
| //if we find a job then we pass it on. |
| if (scheduler.memoryMatcher.matchesMemoryRequirements(j, type, |
| taskTrackerStatus)) { |
| // We found a suitable job. Get task from it. |
| Task t = obtainNewTask(taskTrackerStatus, j); |
| //if there is a task return it immediately. |
| if (t != null) { |
| // we're successful in getting a task |
| return TaskLookupResult.getTaskFoundResult(t); |
| } else { |
| //skip to the next job in the queue. |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Job " + j.getJobID().toString() |
| + " returned no tasks of type " + type); |
| } |
| } |
| } else { |
| // if memory requirements don't match then we check if the job has |
| // pending tasks and has insufficient number of 'reserved' |
| // tasktrackers to cover all pending tasks. If so we reserve the |
| // current tasktracker for this job so that high memory jobs are not |
| // starved |
| TaskDataView view = TaskDataView.getTaskDataView(type); |
| if ((view.getPendingTasks(j) != 0 && |
| !view.hasSufficientReservedTaskTrackers(j))) { |
| // Reserve all available slots on this tasktracker |
| LOG.info(j.getJobID() + ": Reserving " |
| + taskTracker.getTrackerName() |
| + " since memory-requirements don't match"); |
| taskTracker.reserveSlots(type, j, taskTracker |
| .getAvailableSlots(type)); |
| |
| // Block |
| return TaskLookupResult.getMemFailedResult(); |
| } |
| }//end of memory check block |
| // if we're here, this job has no task to run. Look at the next job. |
| }//end of for loop |
| |
| // if we're here, we haven't found any task to run among all jobs in |
| // the queue. This could be because there is nothing to run, or that |
| // the user limit for some user is too strict, i.e., there's at least |
| // one user who doesn't have enough tasks to satisfy his limit. If |
| // it's the latter case, re-look at jobs without considering user |
| // limits, and get a task from the first eligible job; however |
| // we do not 'reserve' slots on tasktrackers anymore since the user is |
| // already over the limit |
| // Note: some of the code from above is repeated here. This is on |
| // purpose as it improves overall readability. |
| // Note: we walk through jobs again. Some of these jobs, which weren't |
| // considered in the first pass, shouldn't be considered here again, |
| // but we still check for their viability to keep the code simple. In |
| // some cases, for high mem jobs that have nothing to run, we call |
| // obtainNewTask() unnecessarily. Should this be a problem, we can |
| // create a list of jobs to look at (those whose users were over |
| // limit) in the first pass and walk through that list only. |
| for (JobInProgress j : |
| scheduler.jobQueuesManager.getJobQueue(qsi.getQueueName()) |
| .getRunningJobs()) { |
| if (j.getStatus().getRunState() != JobStatus.RUNNING) { |
| continue; |
| } |
| |
| //Check for the maximum-capacity. |
| if(areTasksInQueueOverMaxCapacity(qsi,j.getNumSlotsPerTask(type))) { |
| continue; |
| } |
| |
| |
| if (scheduler.memoryMatcher.matchesMemoryRequirements(j, type, |
| taskTrackerStatus)) { |
| // We found a suitable job. Get task from it. |
| Task t = obtainNewTask(taskTrackerStatus, j); |
| //if there is a task return it immediately. |
| if (t != null) { |
| // we're successful in getting a task |
| return TaskLookupResult.getTaskFoundResult(t); |
| } else { |
| } |
| } else { |
| //if memory requirements don't match then we check if the |
| //job has either pending or speculative task. If the job |
| //has pending or speculative task we block till this job |
| //tasks get scheduled, so that high memory jobs are not |
| //starved |
| if (TaskDataView.getTaskDataView(type).getPendingTasks(j) != 0 || |
| hasSpeculativeTask(j, taskTrackerStatus)) { |
| return TaskLookupResult.getMemFailedResult(); |
| } |
| }//end of memory check block |
| }//end of for loop |
| |
| // found nothing for this queue, look at the next one. |
| if (LOG.isDebugEnabled()) { |
| String msg = "Found no task from the queue " + qsi.getQueueName(); |
| LOG.debug(msg); |
| } |
| return TaskLookupResult.getNoTaskFoundResult(); |
| } |
| |
| // Always return a TaskLookupResult object. Don't return null. |
| // The caller is responsible for ensuring that the QSC objects and the |
| // collections are up-to-date. |
| private TaskLookupResult assignTasks(TaskTracker taskTracker) |
| throws IOException { |
| TaskTrackerStatus taskTrackerStatus = taskTracker.getStatus(); |
| |
| printQSCs(); |
| |
| // Check if this tasktracker has been reserved for a job... |
| JobInProgress job = taskTracker.getJobForFallowSlot(type); |
| if (job != null) { |
| int availableSlots = taskTracker.getAvailableSlots(type); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug(job.getJobID() + ": Checking 'reserved' tasktracker " + |
| taskTracker.getTrackerName() + " with " + availableSlots + |
| " '" + type + "' slots"); |
| } |
| |
| if (availableSlots >= job.getNumSlotsPerTask(type)) { |
| // Unreserve |
| taskTracker.unreserveSlots(type, job); |
| |
| // We found a suitable job. Get task from it. |
| Task t = obtainNewTask(taskTrackerStatus, job); |
| //if there is a task return it immediately. |
| if (t != null) { |
| if (LOG.isDebugEnabled()) { |
| LOG.info(job.getJobID() + ": Got " + t.getTaskID() + |
| " for reserved tasktracker " + |
| taskTracker.getTrackerName()); |
| } |
| // we're successful in getting a task |
| return TaskLookupResult.getTaskFoundResult(t); |
| } |
| } else { |
| // Re-reserve the current tasktracker |
| taskTracker.reserveSlots(type, job, availableSlots); |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug(job.getJobID() + ": Re-reserving " + |
| taskTracker.getTrackerName()); |
| } |
| |
| return TaskLookupResult.getMemFailedResult(); |
| } |
| } |
| |
| for (AbstractQueue q : getOrderedJobQueues()) { |
| QueueSchedulingContext qsc = q.getQueueSchedulingContext(); |
| // we may have queues with capacity=0. We shouldn't look at jobs from |
| // these queues |
| if (0 == getTSC(qsc).getCapacity()) { |
| continue; |
| } |
| |
| //This call is important for optimization purposes , if we |
| //have reached the limit already no need for traversing the queue. |
| if(this.areTasksInQueueOverMaxCapacity(qsc,1)) { |
| continue; |
| } |
| |
| TaskLookupResult tlr = getTaskFromQueue(taskTracker, qsc); |
| TaskLookupResult.LookUpStatus lookUpStatus = tlr.getLookUpStatus(); |
| |
| if (lookUpStatus == TaskLookupResult.LookUpStatus.NO_TASK_FOUND) { |
| continue; // Look in other queues. |
| } |
| |
| // if we find a task, return |
| if (lookUpStatus == TaskLookupResult.LookUpStatus.TASK_FOUND) { |
| return tlr; |
| } |
| // if there was a memory mismatch, return |
| else if (lookUpStatus == |
| TaskLookupResult.LookUpStatus.TASK_FAILING_MEMORY_REQUIREMENT) { |
| return tlr; |
| } |
| } |
| |
| // nothing to give |
| return TaskLookupResult.getNoTaskFoundResult(); |
| } |
| |
| |
| /** |
| * Check if maximum-capacity is set for this queue. |
| * If set and greater than 0 , |
| * check if numofslotsoccupied+numSlotsPerTask is greater than |
| * maximum-Capacity ,if yes , implies this queue is over limit. |
| * |
| * Incase noOfSlotsOccupied is less than maximum-capacity ,but , |
| * numOfSlotsOccupied+noSlotsPerTask is more than maximum-capacity we still |
| * dont assign the task . This may lead to under utilization of very small |
| * set of slots. But this is ok ,as we strictly respect the maximum-capacity |
| * @param qsc |
| * @param noOfSlotsPerTask |
| * @return true if queue is over maximum-capacity |
| */ |
| private boolean areTasksInQueueOverMaxCapacity( |
| QueueSchedulingContext qsc,int noOfSlotsPerTask) { |
| TaskSchedulingContext tsi = getTSC(qsc); |
| //check for maximum-capacity |
| if(tsi.getMaxCapacity() >= 0) { |
| if ((tsi.getNumSlotsOccupied() + noOfSlotsPerTask) > |
| tsi.getMaxCapacity()) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug( |
| "Queue " + qsc.getQueueName() + " " + "has reached its max " + |
| type + "Capacity"); |
| LOG.debug("Current running tasks " + tsi.getCapacity()); |
| |
| } |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| |
| // for debugging. |
| private void printQSCs() { |
| if (LOG.isDebugEnabled()) { |
| StringBuffer s = new StringBuffer(); |
| for (AbstractQueue aq: getOrderedJobQueues()) { |
| QueueSchedulingContext qsi = aq.getQueueSchedulingContext(); |
| TaskSchedulingContext tsi = getTSC(qsi); |
| Collection<JobInProgress> runJobs = |
| scheduler.jobQueuesManager.getJobQueue(qsi.getQueueName()) |
| .getRunningJobs(); |
| s.append( |
| String.format( |
| " Queue '%s'(%s): runningTasks=%d, " |
| + "occupiedSlots=%d, capacity=%d, runJobs=%d maximumCapacity=%d ", |
| qsi.getQueueName(), |
| this.type, tsi.getNumRunningTasks(), |
| tsi.getNumSlotsOccupied(), tsi.getCapacity(), (runJobs.size()), |
| tsi.getMaxCapacity())); |
| } |
| LOG.debug(s); |
| } |
| } |
| |
| /** |
| * Check if one of the tasks have a speculative task to execute on the |
| * particular task tracker. |
| * |
| * @param tips tasks of a job |
| * @param tts task tracker status for which we are asking speculative tip |
| * @return true if job has a speculative task to run on particular TT. |
| */ |
| boolean hasSpeculativeTask( |
| TaskInProgress[] tips, |
| TaskTrackerStatus tts) { |
| long currentTime = System.currentTimeMillis(); |
| for(TaskInProgress tip : tips) { |
| if(tip.isRunning() |
| && !(tip.hasRunOnMachine(tts.getHost(), tts.getTrackerName())) |
| && tip.canBeSpeculated(currentTime)) { |
| return true; |
| } |
| } |
| return false; |
| } |
| } |
| |
| /** |
| * The scheduling algorithms for map tasks. |
| */ |
| private static class MapSchedulingMgr extends TaskSchedulingMgr { |
| |
| MapSchedulingMgr(CapacityTaskScheduler schedulr) { |
| super(schedulr); |
| type = TaskType.MAP; |
| queueComparator = mapComparator; |
| } |
| |
| @Override |
| Task obtainNewTask(TaskTrackerStatus taskTracker, JobInProgress job) |
| throws IOException { |
| synchronized (scheduler) { |
| ClusterStatus clusterStatus = scheduler.taskTrackerManager |
| .getClusterStatus(); |
| int numTaskTrackers = clusterStatus.getTaskTrackers(); |
| return job.obtainNewMapTask(taskTracker, numTaskTrackers, |
| scheduler.taskTrackerManager.getNumberOfUniqueHosts()); |
| } |
| } |
| |
| @Override |
| int getClusterCapacity() { |
| synchronized (scheduler) { |
| return scheduler.taskTrackerManager.getClusterStatus().getMaxMapTasks(); |
| } |
| } |
| |
| @Override |
| TaskSchedulingContext getTSC(QueueSchedulingContext qsi) { |
| return qsi.getMapTSC(); |
| } |
| |
| |
| @Override |
| boolean hasSpeculativeTask(JobInProgress job, TaskTrackerStatus tts) { |
| //Check if job supports speculative map execution first then |
| //check if job has speculative maps. |
| return (job.getMapSpeculativeExecution()) && ( |
| hasSpeculativeTask(job.getTasks(TaskType.MAP), |
| tts)); |
| } |
| |
| } |
| |
| /** |
| * The scheduling algorithms for reduce tasks. |
| */ |
| private static class ReduceSchedulingMgr extends TaskSchedulingMgr { |
| |
| ReduceSchedulingMgr(CapacityTaskScheduler schedulr) { |
| super(schedulr); |
| type = TaskType.REDUCE; |
| queueComparator = reduceComparator; |
| } |
| |
| @Override |
| Task obtainNewTask(TaskTrackerStatus taskTracker, JobInProgress job) |
| throws IOException { |
| synchronized (scheduler) { |
| ClusterStatus clusterStatus = scheduler.taskTrackerManager |
| .getClusterStatus(); |
| int numTaskTrackers = clusterStatus.getTaskTrackers(); |
| return job.obtainNewReduceTask(taskTracker, numTaskTrackers, |
| scheduler.taskTrackerManager.getNumberOfUniqueHosts()); |
| } |
| } |
| |
| @Override |
| int getClusterCapacity() { |
| synchronized (scheduler) { |
| return scheduler.taskTrackerManager.getClusterStatus() |
| .getMaxReduceTasks(); |
| } |
| } |
| |
| @Override |
| TaskSchedulingContext getTSC(QueueSchedulingContext qsi) { |
| return qsi.getReduceTSC(); |
| } |
| |
| @Override |
| boolean hasSpeculativeTask(JobInProgress job, TaskTrackerStatus tts) { |
| //check if the job supports reduce speculative execution first then |
| //check if the job has speculative tasks. |
| return (job.getReduceSpeculativeExecution()) && ( |
| hasSpeculativeTask(job.getTasks(TaskType.REDUCE), |
| tts)); |
| } |
| |
| } |
| |
| /** the scheduling mgrs for Map and Reduce tasks */ |
| protected TaskSchedulingMgr mapScheduler = new MapSchedulingMgr(this); |
| protected TaskSchedulingMgr reduceScheduler = new ReduceSchedulingMgr(this); |
| |
| MemoryMatcher memoryMatcher = new MemoryMatcher(); |
| |
| static final Log LOG = LogFactory.getLog(CapacityTaskScheduler.class); |
| protected JobQueuesManager jobQueuesManager; |
| |
| /** whether scheduler has started or not */ |
| private boolean started = false; |
| |
| /** |
| * A clock class - can be mocked out for testing. |
| */ |
| static class Clock { |
| long getTime() { |
| return System.currentTimeMillis(); |
| } |
| } |
| |
| private Clock clock; |
| private JobInitializationPoller initializationPoller; |
| |
| class CapacitySchedulerQueueRefresher extends QueueRefresher { |
| @Override |
| void refreshQueues(List<JobQueueInfo> newRootQueues) |
| throws Throwable { |
| if (!started) { |
| String msg = |
| "Capacity Scheduler is not in the 'started' state." |
| + " Cannot refresh queues."; |
| LOG.error(msg); |
| throw new IOException(msg); |
| } |
| CapacitySchedulerConf schedConf = new CapacitySchedulerConf(); |
| initializeQueues(newRootQueues, schedConf, true); |
| initializationPoller.refreshQueueInfo(schedConf); |
| } |
| } |
| |
| public CapacityTaskScheduler() { |
| this(new Clock()); |
| } |
| |
| // for testing |
| public CapacityTaskScheduler(Clock clock) { |
| this.jobQueuesManager = new JobQueuesManager(); |
| this.clock = clock; |
| } |
| |
| @Override |
| QueueRefresher getQueueRefresher() { |
| return new CapacitySchedulerQueueRefresher(); |
| } |
| |
| /** |
| * Only for testing. |
| * @param type |
| * @return |
| */ |
| String[] getOrderedQueues(TaskType type) { |
| if (type == TaskType.MAP) { |
| return mapScheduler.getOrderedQueues(); |
| } else if (type == TaskType.REDUCE) { |
| return reduceScheduler.getOrderedQueues(); |
| } |
| return null; |
| } |
| |
| @Override |
| public synchronized void start() throws IOException { |
| if (started) return; |
| super.start(); |
| |
| // Initialize MemoryMatcher |
| MemoryMatcher.initializeMemoryRelatedConf(conf); |
| |
| // read queue info from config file |
| QueueManager queueManager = taskTrackerManager.getQueueManager(); |
| |
| // initialize our queues from the config settings |
| CapacitySchedulerConf schedConf = new CapacitySchedulerConf(); |
| try { |
| initializeQueues(queueManager.getRoot().getJobQueueInfo().getChildren(), |
| schedConf, false); |
| } catch (Throwable e) { |
| LOG.error("Couldn't initialize queues because of the excecption : " |
| + StringUtils.stringifyException(e)); |
| throw new IOException(e); |
| } |
| |
| // Queues are ready. Now register jobQueuesManager with the JobTracker so as |
| // to listen to job changes |
| taskTrackerManager.addJobInProgressListener(jobQueuesManager); |
| |
| //Start thread for initialization |
| if (initializationPoller == null) { |
| this.initializationPoller = new JobInitializationPoller( |
| jobQueuesManager, taskTrackerManager); |
| } |
| initializationPoller.init(jobQueuesManager.getJobQueueNames(), schedConf); |
| initializationPoller.setDaemon(true); |
| initializationPoller.start(); |
| |
| started = true; |
| |
| LOG.info("Capacity scheduler started successfully"); |
| } |
| |
| /** |
| * Read the configuration and initialize the queues. This operation should be |
| * done only when either the scheduler is starting or a request is received |
| * from {@link QueueManager} to refresh the queue configuration. |
| * |
| * <p> |
| * |
| * Even in case of refresh, we do not explicitly destroy AbstractQueue items, |
| * or the info maps, they will be automatically garbage-collected. |
| * |
| * <p> |
| * |
| * We don't explicitly lock the scheduler completely. This method is called at |
| * two times. 1) When the scheduler is starting. During this time, the lock |
| * sequence is JT->scheduler and so we don't need any more locking here. 2) |
| * When refresh is issued to {@link QueueManager}. When this happens, parallel |
| * refreshes are guarded by {@link QueueManager} itself by taking its lock. |
| * |
| * @param newRootQueues |
| * @param schedConf |
| * @param refreshingQueues |
| * @throws Throwable |
| */ |
| private void initializeQueues(List<JobQueueInfo> newRootQueues, |
| CapacitySchedulerConf schedConf, boolean refreshingQueues) |
| throws Throwable { |
| |
| if (newRootQueues == null) { |
| throw new IOException( |
| "Cannot initialize the queues with null root-queues!"); |
| } |
| |
| // Sanity check: there should be at least one queue. |
| if (0 == newRootQueues.size()) { |
| throw new IllegalStateException("System has no queue configured!"); |
| } |
| |
| // Create a new queue-hierarchy builder and try loading the complete |
| // hierarchy of queues. |
| AbstractQueue newRootAbstractQueue; |
| try { |
| newRootAbstractQueue = |
| new QueueHierarchyBuilder().createHierarchy(newRootQueues, schedConf); |
| |
| } catch (Throwable e) { |
| LOG.error("Exception while tryign to (re)initializing queues : " |
| + StringUtils.stringifyException(e)); |
| LOG.info("(Re)initializing the queues with the new configuration " |
| + "failed, so keeping the old configuration."); |
| throw e; |
| } |
| |
| // New configuration is successfully validated and applied, set the new |
| // configuration to the current queue-hierarchy. |
| |
| if (refreshingQueues) { |
| // Scheduler is being refreshed. |
| |
| // Going to commit the changes to the hierarchy. Lock the scheduler. |
| synchronized (this) { |
| AbstractQueueComparator comparator = new AbstractQueueComparator(); |
| this.root.sort(comparator); |
| newRootAbstractQueue.sort(comparator); |
| root.validateAndCopyQueueContexts(newRootAbstractQueue); |
| } |
| } else { |
| // Scheduler is just starting. |
| |
| this.root = newRootAbstractQueue; |
| |
| // JobQueue objects are created. Inform the JobQueuesManager so that it |
| // can track the running/waiting jobs. JobQueuesManager is still not added |
| // as a listener to JobTracker, so no locking needed. |
| addJobQueuesToJobQueuesManager(); |
| } |
| |
| List<AbstractQueue> allQueues = new ArrayList<AbstractQueue>(); |
| allQueues.addAll(getRoot().getDescendantContainerQueues()); |
| allQueues.addAll(getRoot().getDescendentJobQueues()); |
| for (AbstractQueue queue : allQueues) { |
| if (!refreshingQueues) { |
| // Scheduler is just starting, create the display info also |
| createDisplayInfo(taskTrackerManager.getQueueManager(), queue.getName()); |
| } |
| |
| // QueueSchedulingContext objects are created/have changed. Put them |
| // (back) in the queue-info so as to be consumed by the UI. |
| addToQueueInfoMap(queue.getQueueSchedulingContext()); |
| } |
| } |
| |
| /** |
| * Inform the {@link JobQueuesManager} about the newly constructed |
| * {@link JobQueue}s. |
| */ |
| private void addJobQueuesToJobQueuesManager() { |
| List<AbstractQueue> allJobQueues = getRoot().getDescendentJobQueues(); |
| for (AbstractQueue jobQ : allJobQueues) { |
| jobQueuesManager.addQueue((JobQueue)jobQ); |
| } |
| } |
| |
| /** mostly for testing purposes */ |
| synchronized void setInitializationPoller(JobInitializationPoller p) { |
| this.initializationPoller = p; |
| } |
| |
| @Override |
| public synchronized void terminate() throws IOException { |
| if (!started) return; |
| if (jobQueuesManager != null) { |
| taskTrackerManager.removeJobInProgressListener( |
| jobQueuesManager); |
| } |
| started = false; |
| initializationPoller.terminate(); |
| super.terminate(); |
| } |
| |
| @Override |
| public synchronized void setConf(Configuration conf) { |
| super.setConf(conf); |
| } |
| |
| /** |
| * provided for the test classes |
| * lets you update the QSI objects and sorted collections |
| */ |
| synchronized void updateContextInfoForTests() { |
| ClusterStatus c = taskTrackerManager.getClusterStatus(); |
| int mapClusterCapacity = c.getMaxMapTasks(); |
| int reduceClusterCapacity = c.getMaxReduceTasks(); |
| // update the QSI objects |
| updateContextObjects(mapClusterCapacity, reduceClusterCapacity); |
| mapScheduler.scheduler.root.sort(mapScheduler.queueComparator); |
| reduceScheduler.scheduler.root.sort(reduceScheduler.queueComparator); |
| } |
| |
| /** |
| * Update individual QSC objects. |
| * We don't need exact information for all variables, just enough for us |
| * to make scheduling decisions. For example, we don't need an exact count |
| * of numRunningTasks. Once we count upto the grid capacity, any |
| * number beyond that will make no difference. |
| * |
| **/ |
| private synchronized void updateContextObjects(int mapClusterCapacity, |
| int reduceClusterCapacity) { |
| root.update(mapClusterCapacity,reduceClusterCapacity); |
| |
| } |
| |
| /* |
| * The grand plan for assigning a task. |
| * Always assigns 1 reduce and 1 map , if sufficient slots are |
| * available for each of types. |
| * If not , then which ever type of slots are available , that type of task is |
| * assigned. |
| * Next, pick a queue. We only look at queues that need a slot. Among these, |
| * we first look at queues whose (# of running tasks)/capacity is the least. |
| * Next, pick a job in a queue. we pick the job at the front of the queue |
| * unless its user is over the user limit. |
| * Finally, given a job, pick a task from the job. |
| * |
| */ |
| @Override |
| public synchronized List<Task> assignTasks(TaskTracker taskTracker) |
| throws IOException { |
| |
| TaskLookupResult tlr; |
| TaskTrackerStatus taskTrackerStatus = taskTracker.getStatus(); |
| List<Task> result = new ArrayList<Task>(); |
| |
| /* |
| * If TT has Map and Reduce slot free, we assign 1 map and 1 reduce |
| * We base decision on how much is needed |
| * versus how much is used |
| */ |
| ClusterStatus c = taskTrackerManager.getClusterStatus(); |
| int mapClusterCapacity = c.getMaxMapTasks(); |
| int reduceClusterCapacity = c.getMaxReduceTasks(); |
| int maxMapSlots = taskTrackerStatus.getMaxMapSlots(); |
| int currentMapSlots = taskTrackerStatus.countOccupiedMapSlots(); |
| int maxReduceSlots = taskTrackerStatus.getMaxReduceSlots(); |
| int currentReduceSlots = taskTrackerStatus.countOccupiedReduceSlots(); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("TT asking for task, max maps=" |
| + taskTrackerStatus.getMaxMapSlots() + |
| ", run maps=" + taskTrackerStatus.countMapTasks() + ", max reds=" + |
| taskTrackerStatus.getMaxReduceSlots() + ", run reds=" + |
| taskTrackerStatus.countReduceTasks() + ", map cap=" + |
| mapClusterCapacity + ", red cap = " + |
| reduceClusterCapacity); |
| } |
| |
| /* |
| * update all our QSC objects. |
| * This involves updating each qsC structure. This operation depends |
| * on the number of running jobs in a queue, and some waiting jobs. If it |
| * becomes expensive, do it once every few heartbeats only. |
| */ |
| updateContextObjects(mapClusterCapacity, reduceClusterCapacity); |
| // make sure we get our map or reduce scheduling object to update its |
| // collection of QSC objects too. |
| |
| if (maxReduceSlots > currentReduceSlots) { |
| //reduce slot available , try to get a |
| //reduce task |
| tlr = reduceScheduler.assignTasks(taskTracker); |
| if (TaskLookupResult.LookUpStatus.TASK_FOUND == |
| tlr.getLookUpStatus()) { |
| result.add(tlr.getTask()); |
| } |
| } |
| |
| if(maxMapSlots > currentMapSlots) { |
| //map slot available , try to get a map task |
| tlr = mapScheduler.assignTasks(taskTracker); |
| if (TaskLookupResult.LookUpStatus.TASK_FOUND == |
| tlr.getLookUpStatus()) { |
| result.add(tlr.getTask()); |
| } |
| } |
| |
| return (result.isEmpty()) ? null : result; |
| } |
| |
| |
| @Override |
| public synchronized Collection<JobInProgress> getJobs(String queueName) { |
| Collection<JobInProgress> jobCollection = new ArrayList<JobInProgress>(); |
| JobQueue jobQueue = jobQueuesManager.getJobQueue(queueName); |
| if (jobQueue == null) { |
| return jobCollection; |
| } |
| Collection<JobInProgress> runningJobs = |
| jobQueue.getRunningJobs(); |
| if (runningJobs != null) { |
| jobCollection.addAll(runningJobs); |
| } |
| Collection<JobInProgress> waitingJobs = |
| jobQueue.getWaitingJobs(); |
| Collection<JobInProgress> tempCollection = new ArrayList<JobInProgress>(); |
| if(waitingJobs != null) { |
| tempCollection.addAll(waitingJobs); |
| } |
| tempCollection.removeAll(runningJobs); |
| if(!tempCollection.isEmpty()) { |
| jobCollection.addAll(tempCollection); |
| } |
| return jobCollection; |
| } |
| |
| synchronized JobInitializationPoller getInitializationPoller() { |
| return initializationPoller; |
| } |
| |
| private synchronized String getDisplayInfo(String queueName) { |
| QueueSchedulingContext qsi = queueInfoMap.get(queueName); |
| if (null == qsi) { |
| return null; |
| } |
| return qsi.toString(); |
| } |
| |
| private synchronized void addToQueueInfoMap(QueueSchedulingContext qsc) { |
| queueInfoMap.put(qsc.getQueueName(), qsc); |
| } |
| |
| /** |
| * Create the scheduler information and set it in the {@link QueueManager}. |
| * this should be only called when the scheduler is starting. |
| * |
| * @param queueManager |
| * @param queueName |
| */ |
| private void createDisplayInfo(QueueManager queueManager, String queueName) { |
| if (queueManager != null) { |
| SchedulingDisplayInfo schedulingInfo = |
| new SchedulingDisplayInfo(queueName, this); |
| queueManager.setSchedulerInfo(queueName, schedulingInfo); |
| } |
| } |
| |
| |
| /** |
| * Use for testing purposes. |
| * returns the root |
| * @return |
| */ |
| AbstractQueue getRoot() { |
| return this.root; |
| } |
| |
| |
| /** |
| * This is used for testing purpose only |
| * Dont use this method. |
| * @param rt |
| */ |
| void setRoot(AbstractQueue rt) { |
| this.root = rt; |
| } |
| |
| } |
