Tasks whose status:not_done and givenUp:zero
@param ds
@return
960457
public List<StatusTrail> getIncompleteTasks(DataSource ds) {
        EntityManager em = getEntityManager();
        try {        
            return em.createQuery("SELECT st FROM StatusTrail st WHERE st.dataSourceId = :dataSourceId "
                    + "AND st.status = 'not_done' AND st.givenUp = 0 ORDER BY st.id DESC", 
                    StatusTrail.class).
                    setParameter("dataSourceId", ds.getId()).getResultList();        
        } finally {
            em.close();
        }
    }
418440
def task_done(self, task_id, result):
        """"""
        try:
            task = self.tasks.pop(task_id)
        except KeyError:
            return  # result of previously timeout Task
        else:
            if task.future.cancelled():
                task.set_running_or_notify_cancel()
            elif isinstance(result, BaseException):
                task.future.set_exception(result)
            else:
                task.future.set_result(result)

            self.task_done_callback()
1175589
public int getRunningTasks(String poolName, TaskType type) {
    Map<String, Integer> runningMap = (type == TaskType.MAP ?
                                       poolRunningMaps : poolRunningReduces);
    return (runningMap.containsKey(poolName) ?
            runningMap.get(poolName) : 0);
  }
81001
public TaskStatus getTaskStatus() {
    TaskStatusUpdate update = get(TaskStatusUpdate.class);
    if (update == null) {
      return null;
    }
    return update.getStatus();
  }
530039
public synchronized Map<String, List<Integer>> getList() {
    final Map<String, List<Integer>> tasks = new LinkedHashMap<>();

    tasks.put("Running", getTaskIdList(runningTasks));
    tasks.put("Waiting", getTaskIdList(taskQueue));
    tasks.put("Finished", getTaskIdList(finishedTasks));
    tasks.put("Canceled", getTaskIdList(canceledTasks));

    return tasks;
  }
327800
public void run(Collection<Runnable> tasks) {
        // Create a semphore that the wrapped runnables will execute
        int numTasks = tasks.size();
        CountDownLatch latch = new CountDownLatch(numTasks);
        for (Runnable r : tasks) {
            if (r == null)
                throw new NullPointerException("Cannot run null tasks");
            workQueue.offer(new CountingRunnable(r, latch));
        }
        try {
            // Wait until all the tasks have finished
            latch.await();
        }
        catch (InterruptedException ie) {
            throw new IllegalStateException("Not all tasks finished", ie);
        }
    }
852539
@Override
    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, long delay) {
        ScheduledExecutorService executor = getScheduledExecutor();
        try {
            return executor.scheduleWithFixedDelay(errorHandlingTask(task, true), 0, delay, TimeUnit.MILLISECONDS);
        } catch (RejectedExecutionException ex) {
            throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
        }
    }
104429
@Override
  public ListenableFuture<TaskStatus> run(final Task task)
  {
    final RemoteTaskRunnerWorkItem completeTask, runningTask, pendingTask;
    if ((pendingTask = pendingTasks.get(task.getId())) != null) {
      log.info("Assigned a task[%s] that is already pending!", task.getId());
      runPendingTasks();
      return pendingTask.getResult();
    } else if ((runningTask = runningTasks.get(task.getId())) != null) {
      ZkWorker zkWorker = findWorkerRunningTask(task.getId());
      if (zkWorker == null) {
        log.warn("Told to run task[%s], but no worker has started running it yet.", task.getId());
      } else {
        log.info("Task[%s] already running on %s.", task.getId(), zkWorker.getWorker().getHost());
        TaskAnnouncement announcement = zkWorker.getRunningTasks().get(task.getId());
        if (announcement.getTaskStatus().isComplete()) {
          taskComplete(runningTask, zkWorker, announcement.getTaskStatus());
        }
      }
      return runningTask.getResult();
    } else if ((completeTask = completeTasks.get(task.getId())) != null) {
      return completeTask.getResult();
    } else {
      return addPendingTask(task).getResult();
    }
  }
458527
public Task obtainTaskCleanupTask(TaskTrackerStatus tts,
                                                 boolean isMapSlot)
  throws IOException {
    if (!tasksInited.get()) {
      return null;
    }

    if (this.status.getRunState() != JobStatus.RUNNING ||
        jobFailed || jobKilled) {
      return null;
    }

    if (isMapSlot) {
      if (mapCleanupTasks.isEmpty())
        return null;
    } else {
      if (reduceCleanupTasks.isEmpty())
        return null;
    }

    synchronized (this) {
      if (this.status.getRunState() != JobStatus.RUNNING ||
          jobFailed || jobKilled) {
        return null;
      }
      String taskTracker = tts.getTrackerName();
      if (!shouldRunOnTaskTracker(taskTracker)) {
        return null;
      }
      TaskAttemptID taskid = null;
      TaskInProgress tip = null;
      if (isMapSlot) {
        if (!mapCleanupTasks.isEmpty()) {
          taskid = mapCleanupTasks.remove(0);
          tip = maps[taskid.getTaskID().getId()];
        }
      } else {
        if (!reduceCleanupTasks.isEmpty()) {
          taskid = reduceCleanupTasks.remove(0);
          tip = reduces[taskid.getTaskID().getId()];
        }
      }
      if (tip != null) {
        return tip.addRunningTask(taskid, taskTracker, true);
      }
      return null;
    }
  }
496967
public CompositeOperation<TaskExecution> getTasks() {
        List<TaskExecution> taskExecutions = CollectionUtils.sort(tasks.values(), slowestFirst());
        return new CompositeOperation<TaskExecution>(taskExecutions);
    }
1207621
public synchronized int cancelTask(final int taskId) throws UnsuccessfulException, NotFoundException {
    if (getTask(taskId, runningTasks) != null) {
      throw new UnsuccessfulException("The task " + taskId + " is running");
    } else if (getTask(taskId, finishedTasks) != null) {
      throw new UnsuccessfulException("The task " + taskId + " has finished");
    }

    final TaskEntity task = getTask(taskId, taskQueue);
    if (task == null) {
      final String message =
          new StringBuilder().append("Task with ID ").append(taskId).append(" is not found").toString();
      throw new NotFoundException(message);
    } else {
      taskQueue.remove(task);
      canceledTasks.add(task);
      return taskId;
    }
  }
934526
public static long getProcessedCount(List<TaskState> taskStates) {
    long value = 0;

    for (TaskState taskState : taskStates) {
      value += taskState.getPropAsLong(ConfigurationKeys.WRITER_RECORDS_WRITTEN, 0);
    }

    return value;
  }
584751
void runTasks(Task[] tasks, int maxTaskIndex) {
        for (int i = 0; i <= maxTaskIndex; i++) {
            runTasksInChain(tasks[i]);
            tasks[i] = null;
        }
    }
722922
private Task taskToBeRescheduled(Task task) {
        Task taskToBeRetried = task.copy();
        taskToBeRetried.setTaskId(IDGenerator.generate());
        taskToBeRetried.setRetriedTaskId(task.getTaskId());
        taskToBeRetried.setStatus(SCHEDULED);
        taskToBeRetried.setRetryCount(task.getRetryCount() + 1);
        taskToBeRetried.setRetried(false);
        taskToBeRetried.setPollCount(0);
        taskToBeRetried.setCallbackAfterSeconds(0);
        task.setRetried(true);
        return taskToBeRetried;
    }
55518
protected final boolean hasScheduledTasks() {
        Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
        ScheduledFutureTask<?> scheduledTask = scheduledTaskQueue == null ? null : scheduledTaskQueue.peek();
        return scheduledTask != null && scheduledTask.deadlineNanos() <= nanoTime();
    }
765997
private List<Task> receiveTasks() {
		synchronized (mutex) {
			if (tasks.isEmpty()) {
				return Collections.emptyList();
			} else {
				List<Task> currentTasks = tasks;
				tasks = new ArrayList<Task>();
				return currentTasks;
			}
		}
	}
884627
def CompleteTask(self, task):
    
    with self._lock:
      if task.identifier not in self._tasks_merging:
        raise KeyError('Task {0:s} was not merging.'.format(task.identifier))

      self.SampleTaskStatus(task, 'completed')

      del self._tasks_merging[task.identifier]

      logger.debug('Completed task {0:s}.'.format(task.identifier))