store level then was executed for the given task to be used to compute correct the level inside the method below
192027
function() {
            // clean results of execution of a previous task

            this.$busy = 0;
            var pc = this.$taskCounter;
            if (this.$error !== null) {
                this.$taskCounter = 0;  // we have to count the tasks on this level
                this.$level       = level + 1;

                try {
                    if (typeof body === 'function') {
                        body.call(this, this.$error);
                    } else if (body === null) {

                    } else {
                        this.dumpError(this.$error);
                    }
                } catch(e) {
                    this.$level       = level; // restore level
                    this.$taskCounter = pc;    // restore counter
                    throw e;
                }
            }

            if (level === 0) {
                try {
                    this.$schedule();
                } catch(e) {
                    this.error(e);
                }
            } else {
                this.$schedule();
            }

            this.$level       = level; // restore level
            this.$taskCounter = pc;    // restore counter
        }
872168
private void updateLastMapLocalityLevel(JobInProgress job,
      Task mapTaskLaunched, TaskTrackerStatus tracker) {
    JobInfo info = infos.get(job);
    LocalityLevel localityLevel = localManager.taskToLocalityLevel(
        job, mapTaskLaunched, tracker);
    info.lastMapLocalityLevel = localityLevel;
    info.timeWaitedForLocalMap = 0;
  }
1063218
private boolean isTaskRunning(TaskStatus taskStatus) {
    TaskStatus.State state = taskStatus.getRunState();
    return (state == State.RUNNING || state == State.UNASSIGNED || 
            taskStatus.inTaskCleanupPhase());
  }
301523
@VisibleForTesting
  void checkSegmentsAndSubmitTasks()
  {
    synchronized (taskLock) {
      List<Interval> intervalsToRemove = new ArrayList<>();
      for (Map.Entry<Interval, HadoopIndexTask> entry : runningTasks.entrySet()) {
        Optional<TaskStatus> taskStatus = taskStorage.getStatus(entry.getValue().getId());
        if (!taskStatus.isPresent() || !taskStatus.get().isRunnable()) {
          intervalsToRemove.add(entry.getKey());
        }
      }
      for (Interval interval : intervalsToRemove) {
        runningTasks.remove(interval);
        runningVersion.remove(interval);
      }

      if (runningTasks.size() == maxTaskCount) {
        //if the number of running tasks reach the max task count, supervisor won't submit new tasks.
        return;
      }
      Pair<SortedMap<Interval, String>, Map<Interval, List<DataSegment>>> toBuildIntervalAndBaseSegments = checkSegments();
      SortedMap<Interval, String> sortedToBuildVersion = toBuildIntervalAndBaseSegments.lhs;
      Map<Interval, List<DataSegment>> baseSegments = toBuildIntervalAndBaseSegments.rhs;
      missInterval = sortedToBuildVersion.keySet();
      submitTasks(sortedToBuildVersion, baseSegments);
    }
  }
145336
public void submit(Task<? extends TaskResult> task) {

        // Increment the counter which indicates the number of results to
        // expect. This must be done BEFORE actually submitting the task for
        // execution.
        remainingTasks.incrementAndGet();

        // Increment the statistics and put the task on the correct queue.
        stats.incrementStartedTasks(task.resultType);
        executors.get(task.resultType).submit(task);

        // Make sure that the task gets added to the results queue.
        resultsQueue.add(task);
        stats.updateMemoryInfo();
    }
52861
def store_from(self, last_level_store):
        """"""
        assert isinstance(last_level_store, Cache), \
            "last_level needs to be a Cache object."
        assert last_level_store.store_to is None, \
            "last_level_store must be a last level cache (.store_to is None)."
        self.last_level_store = last_level_store
918714
def CreateTaskStorage(self, task):
    
    if task.identifier in self._task_storage_writers:
      raise IOError('Storage writer for task: {0:s} already exists.'.format(
          task.identifier))

    storage_writer = FakeStorageWriter(
        self._session, storage_type=definitions.STORAGE_TYPE_TASK, task=task)
    self._task_storage_writers[task.identifier] = storage_writer
    return storage_writer
636738
public void accept(Task task) {
    	if((activeIndex+1)%2==0)
    	{
    		if(!task.isInQueue0())
    		{
    			taskList[0].offer(task);
    			task.storedInQueue0();
    		}
    	}
    	else
    	{
    		if(!task.isInQueue1())
    		{
    			taskList[1].offer(task);
    			task.storedInQueue1();
    		}
    	}    	    	    
    }
676190
def _ProcessTask(self, task):
    
    logger.debug('Started processing task: {0:s}.'.format(task.identifier))

    if self._tasks_profiler:
      self._tasks_profiler.Sample(task, 'processing_started')

    self._task = task

    storage_writer = self._storage_writer.CreateTaskStorage(task)

    if self._serializers_profiler:
      storage_writer.SetSerializersProfiler(self._serializers_profiler)

    storage_writer.Open()

    self._parser_mediator.SetStorageWriter(storage_writer)

    storage_writer.WriteTaskStart()

    try:
      # TODO: add support for more task types.
      self._ProcessPathSpec(
          self._extraction_worker, self._parser_mediator, task.path_spec)
      self._number_of_consumed_sources += 1

      if self._guppy_memory_profiler:
        self._guppy_memory_profiler.Sample()

    finally:
      storage_writer.WriteTaskCompletion(aborted=self._abort)

      self._parser_mediator.SetStorageWriter(None)

      storage_writer.Close()

    try:
      self._storage_writer.FinalizeTaskStorage(task)
    except IOError:
      pass

    self._task = None

    if self._tasks_profiler:
      self._tasks_profiler.Sample(task, 'processing_completed')

    logger.debug('Completed processing task: {0:s}.'.format(task.identifier))
406463
@Override
    protected void executeImpl() throws TaskExecutionFailedException {

        try {
            String account = task.getAccount();
            String storeId = task.getStoreId();
            String spaceId = task.getSpaceId();
            String contentId = task.getContentId();
            String action = task.getAction();
            Map<String, String> props = task.getContentProperties();
            String acls = task.getSpaceACLs();
            Date timestamp = new Date(Long.valueOf(task.getDateTime()));

            auditLogStore.write(account,
                                storeId,
                                spaceId,
                                contentId,
                                task.getContentChecksum(),
                                task.getContentMimetype(),
                                task.getContentSize(),
                                task.getUserId(),
                                action,
                                props != null ? AuditLogStoreUtil.serialize(props) : null,
                                acls,
                                task.getSourceSpaceId(),
                                task.getSourceContentId(),
                                timestamp);

            log.debug("audit task successfully processed: {}", task);
        } catch (TransactionSystemException e) {
            log.error("failed to write item  ( account={} storeId={} spaceId={} contentId={} timestamp={} ) " +
                      "to the database due to a transactional error. Likely cause: duplicate entry. Details: {}. " +
                      "Ignoring...",
                      task.getAccount(),
                      task.getStoreId(),
                      task.getSpaceId(),
                      task.getContentId(),
                      new Date(Long.valueOf(task.getDateTime())),
                      e.getMessage());
        } catch (Exception e) {
            String message = "Failed to execute " + task + ": " + e.getMessage();
            log.debug(message, e);
            throw new TaskExecutionFailedException(message, e);
        }
    }
1366807
private static void handleDelayedSlotFill(DelayedSlotFillTask task) {
    Key slotKey = task.getSlotKey();
    Slot slot = querySlotOrAbandonTask(slotKey, true);
    Key rootJobKey = task.getRootJobKey();
    UpdateSpec updateSpec = new UpdateSpec(rootJobKey);
    slot.fill(null);
    updateSpec.getNonTransactionalGroup().includeSlot(slot);
    backEnd.save(updateSpec, task.getQueueSettings());
    // re-reading Slot (in handleSlotFilled) is needed (to capture slot fill after this one)
    handleSlotFilled(new HandleSlotFilledTask(slotKey, task.getQueueSettings()));
  }
1160405
private void calcChannelsStats(List<TaskRec> tasks) {
		for (ChannelWorkContext wc : channelContextMap.values()) {
			wc.lastGotCount = 0;
		}
		for (TaskRec taskRec : tasks) {
			ChannelWorkContext wc = channelContextMap.get(taskRec.channel);
			if (wc == null) {
				wc = new ChannelWorkContext(taskRec.channel);
				channelContextMap.put(taskRec.channel, wc);
			}
			wc.lastGotCount++;
		}
		// update next portion - double if found something, clear to minimum if not
		for (ChannelWorkContext wc : channelContextMap.values()) {
			if (wc.lastGotCount > 0) {
				wc.nextSlowLimit = Math.min(wc.nextSlowLimit * 2, MAX_TASK_COUNT);
				logger.debug("Channel " + wc.channelName + " nextSlowLimit=" + wc.nextSlowLimit);
			} else {
				wc.dropNextSlowLimit();
			}
		}
	}
53811
function queue(task, done) {
    var storage = this,
        key = this.keygen(task);

    // store task as JSON serialized string
    task = JSON.stringify(task);

    // first add the key to the queue
    this.redis().rpush(this.queueKey(), key, function(err) {
        if (err) done(err);

        // now store task data
        else storage.redis().set(key, task, function(err) {
            if (err) done(err);
            else done(null, key);
        });
    });
}
206335
def CheckTaskToMerge(self, task):
    
    with self._lock:
      is_abandoned = task.identifier in self._tasks_abandoned
      is_processing = task.identifier in self._tasks_processing
      is_queued = task.identifier in self._tasks_queued

      if not is_queued and not is_processing and not is_abandoned:
        raise KeyError('Status of task {0:s} is unknown.'.format(
            task.identifier))

      return is_queued or is_processing or is_abandoned and not task.has_retry
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;
  }
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))
421959
def run(self):
        '''
        '''
        if self.verbose:
            pbar = tqdm(total=100)
        while True:
            task_remain = self._task_queue.qsize()
            task_finished = int((float(self.total_task - task_remain) /
                                 float(self.total_task)) * 100)
            if task_finished % 20 == 0 and task_finished != self.current_state:
                self.current_state = task_finished
                logger.info('{0}% done'.format(task_finished))
                if self.verbose and task_finished > 0:
                    pbar.update(20)
            if task_remain == 0:
                break
        logger.debug('All task data cleared')