Is a job being starved for fair share for the given task type?
This is defined as being below half its fair share *and* having a
positive deficit.
320320
boolean isStarvedForFairShare(JobInfo info, TaskType type) {
    int desiredFairShare = (int) Math.floor(Math.min(
        (fairTasks(info, type) + 1) / 2, runnableTasks(info, type)));
    return (runningTasks(info, type) < desiredFairShare);
  }
564657
boolean takeWorkPermit(boolean takeTaskPermit)
    {
        int taskDelta = takeTaskPermit ? 1 : 0;
        while (true)
        {
            long current = permits.get();
            int workPermits = workPermits(current);
            int taskPermits = taskPermits(current);
            if (workPermits == 0 || taskPermits == 0)
                return false;
            if (permits.compareAndSet(current, combine(taskPermits - taskDelta, workPermits - 1)))
            {
                if (takeTaskPermit && taskPermits == maxTasksQueued && hasRoom.hasWaiters())
                    hasRoom.signalAll();
                return true;
            }
        }
    }
1125819
private double computeShare(JobInfo info, double w2sRatio,
      TaskType type, boolean considerMinMax) {
    if (!isRunnable(info)) {
      return 0;
    }
    double share = type == TaskType.MAP ? info.mapWeight : info.reduceWeight;
    share *= w2sRatio;
    if (considerMinMax) {
      int minSlots = type == TaskType.MAP ? info.minMaps : info.minReduces;
      share = Math.max(share, minSlots);
      int maxSlots = type == TaskType.MAP ? info.maxMaps : info.maxReduces;
      share = Math.min(share, maxSlots);
    }
    share = Math.min(share, runnableTasks(info, type));
    return share;
  }
661614
function () {
        if (job.pull_requests < worker_count) {
            return false;
        }
        for (var i = 0; i < worker_count; i++) {
            if (!job.worker_complete[i]) {
                return false;
            }
        }
        return true;
    }
473665
private boolean isRelated(Task task, List<Relation> list)
   {
      boolean result = false;
      for (Relation relation : list)
      {
         if (relation.getTargetTask().getUniqueID().intValue() == task.getUniqueID().intValue())
         {
            result = true;
            break;
         }
      }
      return result;
   }
550850
private void requirePending() {
		if (pending != null) {
			return;
		}

		long exclusionComparable = Long.MIN_VALUE;
		while (nInclusionsRemaining != 0 && !queue.isEmpty()) {
			//find a candidate that is not excluded
			HeapElement inclusion = null;
			do {
				HeapElement candidate = queue.poll();
				if (candidate.inclusion) {
					if (exclusionComparable != candidate.comparable()) {
						inclusion = candidate;
						break;
					}
				} else {
					exclusionComparable = candidate.comparable();
				}
				reattach(candidate);
				if (nInclusionsRemaining == 0) {
					return;
				}
			} while (!queue.isEmpty());
			if (inclusion == null) {
				return;
			}
			long inclusionComparable = inclusion.comparable();

			/*
			 * Check for any following exclusions and for duplicates. We could
			 * change the sort order so that exclusions always preceded
			 * inclusions, but that would be less efficient and would make the
			 * ordering different than the comparable value.
			 */
			boolean excluded = exclusionComparable == inclusionComparable;
			while (!queue.isEmpty() && queue.peek().comparable() == inclusionComparable) {
				HeapElement match = queue.poll();
				excluded |= !match.inclusion;
				reattach(match);
				if (nInclusionsRemaining == 0) {
					return;
				}
			}
			if (!excluded) {
				pending = inclusion;
				return;
			}
			reattach(inclusion);
		}
	}
107788
def ready(self, job):
        
        no_deps = len(job.depends_on) == 0
        all_complete = all(j.is_complete() for j in self.active_jobs
                if j.alias in job.depends_on)
        none_failed = not any(True for j in self.failed
                if j.alias in job.depends_on)
        queue_is_open = len(self.running) < self.MAX_CONCURRENT_JOBS
        return queue_is_open and (no_deps or (all_complete and none_failed))
381589
public Task pollTask(String taskType, String workerId, String domain) {
        Preconditions.checkArgument(StringUtils.isNotBlank(taskType), "Task type cannot be blank");
        Preconditions.checkArgument(StringUtils.isNotBlank(domain), "Domain cannot be blank");
        Preconditions.checkArgument(StringUtils.isNotBlank(workerId), "Worker id cannot be blank");

        TaskServicePb.PollResponse response = stub.poll(
                TaskServicePb.PollRequest.newBuilder()
                .setTaskType(taskType)
                .setWorkerId(workerId)
                .setDomain(domain)
                .build()
        );
        return protoMapper.fromProto(response.getTask());
    }
1024866
private CompletableFuture<LogicalSlot> allocateSharedSlot(
		SlotRequestId slotRequestId,
		ScheduledUnit scheduledUnit,
		SlotProfile slotProfile,
		boolean allowQueuedScheduling,
		Time allocationTimeout) {
		// allocate slot with slot sharing
		final SlotSharingManager multiTaskSlotManager = slotSharingManagers.computeIfAbsent(
			scheduledUnit.getSlotSharingGroupId(),
			id -> new SlotSharingManager(
				id,
				slotPool,
				this));

		final SlotSharingManager.MultiTaskSlotLocality multiTaskSlotLocality;
		try {
			if (scheduledUnit.getCoLocationConstraint() != null) {
				multiTaskSlotLocality = allocateCoLocatedMultiTaskSlot(
					scheduledUnit.getCoLocationConstraint(),
					multiTaskSlotManager,
					slotProfile,
					allowQueuedScheduling,
					allocationTimeout);
			} else {
				multiTaskSlotLocality = allocateMultiTaskSlot(
					scheduledUnit.getJobVertexId(),
					multiTaskSlotManager,
					slotProfile,
					allowQueuedScheduling,
					allocationTimeout);
			}
		} catch (NoResourceAvailableException noResourceException) {
			return FutureUtils.completedExceptionally(noResourceException);
		}

		// sanity check
		Preconditions.checkState(!multiTaskSlotLocality.getMultiTaskSlot().contains(scheduledUnit.getJobVertexId()));

		final SlotSharingManager.SingleTaskSlot leaf = multiTaskSlotLocality.getMultiTaskSlot().allocateSingleTaskSlot(
			slotRequestId,
			scheduledUnit.getJobVertexId(),
			multiTaskSlotLocality.getLocality());
		return leaf.getLogicalSlotFuture();
	}
958673
@SuppressWarnings("deprecation")
  public void failedTask(
    TaskInProgress tip,
    TaskAttemptID taskid,
    String reason,
    TaskStatus.Phase phase,
    boolean isFailed,
    String trackerName,
    TaskTrackerInfo ttStatus) {
    TaskStatus.State state =
      isFailed ? TaskStatus.State.FAILED: TaskStatus.State.KILLED;
    TaskStatus status = TaskStatus.createTaskStatus(tip.isMapTask(),
                                                    taskid,
                                                    0.0f,
                                                    1,
                                                    state,
                                                    reason,
                                                    reason,
                                                    trackerName, phase,
                                                    new Counters());
    synchronized (lockObject) {
      // update the actual start-time of the attempt
      TaskStatus oldStatus = tip.getTaskStatus(taskid);
      long startTime = oldStatus == null
      ? JobTracker.getClock().getTime()
          : oldStatus.getStartTime();
      if (startTime < 0) {
        startTime = JobTracker.getClock().getTime();
      }
      status.setStartTime(startTime);
      status.setFinishTime(JobTracker.getClock().getTime());
      boolean wasComplete = tip.isComplete();
      updateTaskStatus(tip, status, ttStatus);
      boolean isComplete = tip.isComplete();
      if (wasComplete && !isComplete) { // mark a successful tip as failed
        String taskType = getTaskType(tip);
        JobHistory.Task.logFailed(tip.getTIPId(), taskType,
            tip.getExecFinishTime(), reason, taskid);
      }
    }
  }
904842
boolean reportSubtaskStats(JobVertexID jobVertexId, SubtaskStateStats subtask) {
		TaskStateStats taskStateStats = taskStats.get(jobVertexId);

		if (taskStateStats != null && taskStateStats.reportSubtaskStats(subtask)) {
			currentNumAcknowledgedSubtasks++;
			latestAcknowledgedSubtask = subtask;

			currentStateSize += subtask.getStateSize();

			long alignmentBuffered = subtask.getAlignmentBuffered();
			if (alignmentBuffered > 0) {
				currentAlignmentBuffered += alignmentBuffered;
			}

			return true;
		} else {
			return false;
		}
	}
354703
def lease(self, seconds=600, num_tasks=1, tag=None):
    
    tag = tag if tag else None
    tasks = self._api.lease(
      numTasks=num_tasks, 
      seconds=seconds,
      groupByTag=(tag is not None),
      tag=tag,
    )

    if not len(tasks):
      raise QueueEmpty

    task = tasks[0]
    return totask(task)
483195
private void checkPendingRequests() {

		final Iterator<Map.Entry<JobID, PendingRequestsMap>> it = this.pendingRequestsOfJob.entrySet().iterator();
		while (it.hasNext()) {

			final List<AllocatedResource> allocatedResources = new ArrayList<AllocatedResource>();
			final Map.Entry<JobID, PendingRequestsMap> entry = it.next();
			final JobID jobID = entry.getKey();
			final PendingRequestsMap pendingRequestsMap = entry.getValue();
			final Iterator<Map.Entry<InstanceType, Integer>> it2 = pendingRequestsMap.iterator();
			while (it2.hasNext()) {

				final Map.Entry<InstanceType, Integer> entry2 = it2.next();
				final InstanceType requestedInstanceType = entry2.getKey();
				int numberOfPendingInstances = entry2.getValue().intValue();

				// Consistency check
				if (numberOfPendingInstances <= 0) {
					LOG.error("Inconsistency: Job " + jobID + " has " + numberOfPendingInstances
						+ " requests for instance type " + requestedInstanceType.getIdentifier());
					continue;
				}

				while (numberOfPendingInstances > 0) {

					if (LOG.isDebugEnabled()) {
						LOG.debug("Trying to allocate instance of type " + requestedInstanceType.getIdentifier());
					}

					// TODO: Introduce topology awareness here
					final AllocatedSlice slice = getSliceOfType(jobID, requestedInstanceType);
					if (slice == null) {
						break;
					} else {

						LOG.info("Allocated instance of type " + requestedInstanceType.getIdentifier()
							+ " as a result of pending request for job " + jobID);

						// Decrease number of pending instances
						--numberOfPendingInstances;
						pendingRequestsMap.decreaseNumberOfPendingInstances(requestedInstanceType);

						List<AllocatedSlice> allocatedSlices = this.slicesOfJobs.get(jobID);
						if (allocatedSlices == null) {
							allocatedSlices = new ArrayList<AllocatedSlice>();
							this.slicesOfJobs.put(jobID, allocatedSlices);
						}
						allocatedSlices.add(slice);

						allocatedResources.add(new AllocatedResource(slice.getHostingInstance(), slice.getType(), slice
							.getAllocationID()));
					}
				}
			}

			if (!allocatedResources.isEmpty() && this.instanceListener != null) {

				final ClusterInstanceNotifier clusterInstanceNotifier = new ClusterInstanceNotifier(
					this.instanceListener, jobID, allocatedResources);

				clusterInstanceNotifier.start();
			}
		}
	}
568166
private static boolean needIncrement(long ldivisor, int roundingMode,
                                         int qsign, MutableBigInteger mq, long r) {
        assert r != 0L;

        int cmpFracHalf;
        if (r <= HALF_LONG_MIN_VALUE || r > HALF_LONG_MAX_VALUE) {
            cmpFracHalf = 1; // 2 * r can't fit into long
        } else {
            cmpFracHalf = longCompareMagnitude(2 * r, ldivisor);
        }

        return commonNeedIncrement(roundingMode, qsign, cmpFracHalf, mq.isOdd());
    }
1009237
int moveTasksFromBuffers(Task[] tasks) {
        int maxTaskIndex = -1;
        for (int i = 0; i < buffers.length; i++) {
            int maxIndex = moveTasksFromBuffer(tasks, i);
            maxTaskIndex = Math.max(maxIndex, maxTaskIndex);
        }
        return maxTaskIndex;
    }
987236
@Service
    public Task poll(String taskType, String workerId, String domain) {
        LOGGER.debug("Task being polled: /tasks/poll/{}?{}&{}", taskType, workerId, domain);
        Task task = executionService.getLastPollTask(taskType, workerId, domain);
        if (task != null) {
            LOGGER.debug("The Task {} being returned for /tasks/poll/{}?{}&{}", task, taskType, workerId, domain);
        }
        Monitors.recordTaskPollCount(taskType, domain, 1);
        return task;
    }
995578
def HasPendingTasks(self):
    
    with self._lock:
      self._AbandonInactiveProcessingTasks()

      if self._tasks_processing:
        return True

      # There are no tasks being processed, but we might be
      # waiting for some tasks to be merged.
      if self._HasTasksPendingMerge():
        return True

      # There are no tasks processing or pending merge, but there may
      # still be some waiting to be retried, so we check that.
      if self._HasTasksPendingRetry():
        return True

      # It is possible that a worker has processed a task and the foreman has
      # not been informed about it, since there is no feedback from the worker
      # when it pops a task from the queue.

      # If we believe all the workers are idle for longer than the task
      # inactive time (timeout) abandon all queued tasks. This ensures
      # that processing actually stops when the foreman never gets an
      # update from a worker.

      if self._tasks_queued:
        inactive_time = time.time() - self._TASK_INACTIVE_TIME
        inactive_time = int(inactive_time * definitions.MICROSECONDS_PER_SECOND)

        if self._latest_task_processing_time < inactive_time:
          self._AbandonQueuedTasks()

      if self._tasks_queued:
        return True

      if self._tasks_merging:
        return True

    # There are no tasks pending any work.
    return False