Get the level of locality that a given task would have if launched on
a particular TaskTracker. Returns 0 if the task has data on that machine,
1 if it has data on the same rack, etc (depending on number of levels in
the network hierarchy).
327974
int getLocalityLevel(TaskInProgress tip, TaskTrackerStatus tts) {
    Node tracker = jobtracker.getNode(tts.getHost());
    int level = this.maxLevel;
    // find the right level across split locations
    for (String local : maps[tip.getIdWithinJob()].getSplitLocations()) {
      Node datanode = jobtracker.getNode(local);
      int newLevel = this.maxLevel;
      if (tracker != null && datanode != null) {
        newLevel = getMatchingLevelForNodes(tracker, datanode);
      }
      if (newLevel < level) {
        level = newLevel;
        // an optimization
        if (level == 0) {
          break;
        }
      }
    }
    return level;
  }
156975
def get_next_task(self):
        """"""
        if _debug: TaskManager._debug("get_next_task")

        # get the time
        now = _time()

        task = None
        delta = None

        if self.tasks:
            # look at the first task
            when, n, nxttask = self.tasks[0]
            if when <= now:
                # pull it off the list and mark that it's no longer scheduled
                heappop(self.tasks)
                task = nxttask
                task.isScheduled = False

                if self.tasks:
                    when, n, nxttask = self.tasks[0]
                    # peek at the next task, return how long to wait
                    delta = max(when - now, 0.0)
            else:
                delta = when - now

        # return the task to run and how long to wait for the next one
        return (task, delta)
1032356
List<Task> getSetupAndCleanupTasks(
    TaskTrackerStatus taskTracker) throws IOException {
    int maxMapTasks = taskScheduler.getMaxSlots(taskTracker, TaskType.MAP);
    int maxReduceTasks =
        taskScheduler.getMaxSlots(taskTracker, TaskType.REDUCE);
    int numMaps = taskTracker.countOccupiedMapSlots();
    int numReduces = taskTracker.countOccupiedReduceSlots();
    int numTaskTrackers = getClusterStatus().getTaskTrackers();
    int numUniqueHosts = getNumberOfUniqueHosts();

    List<JobInProgress> cachedJobs = new ArrayList<JobInProgress> ();

    // get a snapshot of all the jobs in the system
    synchronized (jobs) {
      cachedJobs.addAll(jobs.values());
    }

    Task t = null;

    if (numMaps < maxMapTasks) {
      for (JobInProgress job: cachedJobs) {
        t = job.obtainJobCleanupTask(taskTracker, numTaskTrackers,
                                     numUniqueHosts, true);
        if (t != null) {
          return Collections.singletonList(t);
        }
      }
      for (JobInProgress job: cachedJobs) {
        t = job.obtainTaskCleanupTask(taskTracker, true);
        if (t != null) {
          return Collections.singletonList(t);
        }
      }
      for (JobInProgress job: cachedJobs) {
        t = job.obtainJobSetupTask(taskTracker, numTaskTrackers,
                                   numUniqueHosts, true);
        if (t != null) {
          return Collections.singletonList(t);
        }
      }
    }
    if (numReduces < maxReduceTasks) {
      for (JobInProgress job: cachedJobs) {
        t = job.obtainJobCleanupTask(taskTracker, numTaskTrackers,
                                     numUniqueHosts, false);
        if (t != null) {
          return Collections.singletonList(t);
        }
      }
      for (JobInProgress job: cachedJobs) {
        t = job.obtainTaskCleanupTask(taskTracker, false);
        if (t != null) {
          return Collections.singletonList(t);
        }
      }
      for (JobInProgress job: cachedJobs) {
        t = job.obtainJobSetupTask(taskTracker, numTaskTrackers,
                                   numUniqueHosts, false);
        if (t != null) {
          return Collections.singletonList(t);
        }
      }
    }

    return null;
  }
512791
def is_previous_task_processing(self, *args, **kwargs):
        
        app = self._get_app()
        inspect = app.control.inspect()
        active = inspect.active() or {}
        scheduled = inspect.scheduled() or {}
        reserved = inspect.reserved() or {}
        uncompleted = sum(list(active.values()) + list(scheduled.values()) + reserved.values(), [])
        return any(self.is_equal(task, *args, **kwargs) for task in uncompleted)
468645
def getWorkerQte(hosts):
    """"""
    if "SLURM_NTASKS" in os.environ:
        return int(os.environ["SLURM_NTASKS"])
    elif "PBS_NP" in os.environ:
        return int(os.environ["PBS_NP"])
    elif "NSLOTS" in os.environ:
        return int(os.environ["NSLOTS"])
    else:
        return sum(host[1] for host in hosts)
740333
def SampleTaskStatus(self, task, status):
    
    if self._tasks_profiler:
      self._tasks_profiler.Sample(task, status)
81001
public TaskStatus getTaskStatus() {
    TaskStatusUpdate update = get(TaskStatusUpdate.class);
    if (update == null) {
      return null;
    }
    return update.getStatus();
  }
900312
public boolean isInLocalRack(InetSocketAddress addr) {
    if (dnsToSwitchMapping == null || this.localhostNetworkLocation == null) {
      return false;
    }
    ArrayList<String> tempList = new ArrayList<String>();
    tempList.add(addr.getAddress().getHostAddress());
    List<String> retList = dnsToSwitchMapping.resolve(tempList);
    if (retList != null && retList.size() > 0) {
      return retList.get(0).equals(this.localhostNetworkLocation);
    } else {
      return false;
    }
  }
1096397
void lostTaskTracker(TaskTracker taskTracker) {
    String trackerName = taskTracker.getTrackerName();
    LOG.info("Lost tracker '" + trackerName + "'");

    // remove the tracker from the local structures
    synchronized (trackerToJobsToCleanup) {
      trackerToJobsToCleanup.remove(trackerName);
    }

    synchronized (trackerToTasksToCleanup) {
      trackerToTasksToCleanup.remove(trackerName);
    }

    Set<TaskAttemptIDWithTip> lostTasks = trackerToTaskMap.get(trackerName);
    trackerToTaskMap.remove(trackerName);

    if (lostTasks != null) {
      // List of jobs which had any of their tasks fail on this tracker
      Set<JobInProgress> jobsWithFailures = new HashSet<JobInProgress>();
      for (TaskAttemptIDWithTip oneTask : lostTasks) {
        TaskAttemptID taskId = oneTask.attemptId;
        TaskInProgress tip = oneTask.tip;
        JobInProgress job = (JobInProgress) tip.getJob();

        // Completed reduce tasks never need to be failed, because
        // their outputs go to dfs
        // And completed maps with zero reducers of the job
        // never need to be failed.
        if (!tip.isComplete() ||
            (tip.isMapTask() && !tip.isJobSetupTask() &&
             job.desiredReduces() != 0)) {
          // if the job is done, we don't want to change anything
          if (job.getStatus().getRunState() == JobStatus.RUNNING ||
              job.getStatus().getRunState() == JobStatus.PREP) {
            // the state will be KILLED_UNCLEAN, if the task(map or reduce)
            // was RUNNING on the tracker
            TaskStatus.State killState = (tip.isRunningTask(taskId) &&
              !tip.isJobSetupTask() && !tip.isJobCleanupTask()) ?
              TaskStatus.State.KILLED_UNCLEAN : TaskStatus.State.KILLED;
            job.failedTask(tip, taskId,
                          ("Lost task tracker: " + trackerName +
                            " at " + new Date()),
                           (tip.isMapTask() ?
                               TaskStatus.Phase.MAP :
                               TaskStatus.Phase.REDUCE),
                            killState,
                            trackerName);
            jobsWithFailures.add(job);
          }
        } else {
          // Completed 'reduce' task and completed 'maps' with zero
          // reducers of the job, not failed;
          // only removed from data-structures.
          markCompletedTaskAttempt(trackerName, taskId);
        }
      }

      // Penalize this tracker for each of the jobs which
      // had any tasks running on it when it was 'lost'
      // Also, remove any reserved slots on this tasktracker
      for (JobInProgress job : jobsWithFailures) {
        String reason = "Tracker went down";
        job.addTrackerTaskFailure(trackerName, taskTracker, reason);
      }

      // Cleanup
      taskTracker.cancelAllReservations();

      // Purge 'marked' tasks, needs to be done
      // here to prevent hanging references!
      removeMarkedTasks(trackerName);
    }
  }
1124641
def cur_depth_level(self):
        
        cur_level = self.initial_depth
        if not self.am_i_main_thread:
            cur_level += len(self.get_tasks(thread_name='MainThread'))

        return cur_level + len(self.tasks)
1144474
def get_task(self, task=None):
        
        # Iterate over the grindstone tasks
        for t in self.grindstone['tasks']:
            # if they key matches the task
            if key_of(t) == task:
                # Return this task
                return t
        # Otherwise return nothing
        return None
1329137
def check(self):
		
		if self.__next_start is not None:
			utc_now = utc_datetime()
			if utc_now >= self.__next_start:
				result = []

				for task_source in self.__next_sources:
					records = task_source.has_records()
					if records is not None:
						result.extend(records)

				self.__update_all()

				if len(result) > 0:
					return tuple(result)
544047
private String getTaskCacheDirectory(TaskControllerContext context) {
    // In the case of JVM reuse, the task specific directory
    // is different from what is set with respect with
    // env.workDir. Hence building this from the taskId everytime.
    String taskId = context.task.getTaskID().toString();
    File cacheDirForJob = context.env.workDir.getParentFile().getParentFile();
    if(context.task.isTaskCleanupTask()) {
      taskId = taskId + TaskTracker.TASK_CLEANUP_SUFFIX;
    }
    return new File(cacheDirForJob, taskId).getAbsolutePath(); 
  }
499617
public int getAvailableSlots(TaskType taskType) {
    int availableSlots = 0;
    if (taskType == TaskType.MAP) {
      if (LOG.isDebugEnabled()) {
        LOG.debug(trackerName + " getAvailSlots:" +
        		     " max(m)=" + status.getMaxMapSlots() + 
        		     " occupied(m)=" + status.countOccupiedMapSlots());
      }
      availableSlots = status.getAvailableMapSlots();
    } else {
      if (LOG.isDebugEnabled()) {
        LOG.debug(trackerName + " getAvailSlots:" +
                  " max(r)=" + status.getMaxReduceSlots() + 
                  " occupied(r)=" + status.countOccupiedReduceSlots());
      }
      availableSlots = status.getAvailableReduceSlots();
    }
    return availableSlots;
  }
1055739
def view_task_info(token, dstore):
    
    args = token.split(':')[1:]  # called as task_info:task_name
    if args:
        [task] = args
        array = dstore['task_info/' + task].value
        rduration = array['duration'] / array['weight']
        data = util.compose_arrays(rduration, array, 'rduration')
        data.sort(order='duration')
        return rst_table(data)

    data = ['operation-duration mean stddev min max outputs'.split()]
    for task in dstore['task_info']:
        val = dstore['task_info/' + task]['duration']
        if len(val):
            data.append(stats(task, val))
    if len(data) == 1:
        return 'Not available'
    return rst_table(data)
1139636
def _get_task_priority(tasks, task_priority):
    
    if task_priority is None:
        return None
    if is_integer(task_priority):
        return task_priority
    if isinstance(task_priority, basestring):
        if task_priority in tasks:
            return tasks[task_priority].priority

    raise ValueError("Unrecognized task priority '{}'".format(task_priority))
935509
def ncores_allocated(self):
        
        return sum(task.manager.num_cores for task in self if task.status in [task.S_SUB, task.S_RUN])