如果 queue 为空
如果 task retry 次数小于系统设置的 DEFAULT_TASK_RESTY
把失败任务重新添加到队列中
543011
private synchronized void queueValid() throws InterruptedException {
        if(this.queue.isEmpty() && !this.faildQueue.isEmpty()){
            for (Task task : this.faildQueue) {
                this.push(task, false);
            }
            this.faildQueue.clear();
        }
        if (this.queue.isEmpty()) {
            logger.info(Thread.currentThread().getName() + " queue is empty");
        }
    }
644814
function retry(err) {
      if (retries < maxRetries) {
        return _Bluebird2['default'].delay(ms).then(function () {
          return loop(fn, retries + 1);
        });
      }

      return _Bluebird2['default'].reject(err);
    }
963996
def _insert_tasks(tasks, queue, transactional=False,
                  retry_transient_errors=True,
                  retry_delay=RETRY_SLEEP_SECS):
    
    from google.appengine.api import taskqueue

    if not tasks:
        return 0

    try:
        taskqueue.Queue(name=queue).add(tasks, transactional=transactional)
        return len(tasks)
    except (taskqueue.BadTaskStateError,
            taskqueue.TaskAlreadyExistsError,
            taskqueue.TombstonedTaskError):
        if len(tasks) <= 1:
            # Task has already been inserted, no reason to report an error here.
            return 0

        # If a list of more than one Tasks is given, a raised exception does
        # not guarantee that no tasks were added to the queue (unless
        # transactional is set to True). To determine which tasks were
        # successfully added when an exception is raised, check the
        # Task.was_enqueued property.
        reinsert = _tasks_to_reinsert(tasks, transactional)
        count = len(reinsert)
        inserted = len(tasks) - count
        inserted += _insert_tasks(reinsert[:count / 2], queue, transactional,
                                  retry_transient_errors, retry_delay)
        inserted += _insert_tasks(reinsert[count / 2:], queue, transactional,
                                  retry_transient_errors, retry_delay)

        return inserted
    except taskqueue.TransientError:
        # Always re-raise for transactional insert, or if specified by
        # options.
        if transactional or not retry_transient_errors:
            raise

        reinsert = _tasks_to_reinsert(tasks, transactional)

        # Retry with a delay, and then let any errors re-raise.
        time.sleep(retry_delay)

        taskqueue.Queue(name=queue).add(reinsert, transactional=transactional)
        return len(tasks)
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);
      }
    }
  }
533164
@Override
    public void requeue(Task task) {
        int attempts = task.getAttempts();
        task.incrementAttempts();
        try {
            deleteTask(task);
        } catch (TaskNotFoundException e) {
            log.error("unable to delete " + task + " ignoring - requeuing anyway");
        }

        put(task);
        log.warn("requeued {} after {} failed attempts.", task, attempts);

    }
671578
function checkRestartQueue() {
            // Kill one worker only if maximum count are working.
            if (restartQueue.length > 0 && listeningWorkersCount === workersCount && !currentRestartingPid) {
                var pid = restartQueue.shift();
                try {
                    // Store process id to wait for its finish.
                    currentRestartingPid = pid;
                    // Send SIGTERM signal to worker. SIGTERM starts graceful shutdown of worker inside it.
                    process.kill(pid);

                } catch(ex) {

                    // Reset current pid.
                    currentRestartingPid = null;

                    // If no process killed, try next in queue.
                    process.nextTick(checkRestartQueue);

                    // Fail silent on 'No such process'. May occur when kill message received after kill initiated but not finished.
                    if (ex.code !== 'ESRCH') {
                        throw ex;
                    }
                }
            }
        }
717157
def retry(self, index=None, params=None):
        
        return self.transport.perform_request(
            "POST", _make_path(index, "_ilm", "retry"), params=params
        )
812673
function(taskFn, taskParams) {
        var task = this._buildTask(taskFn, taskParams);

        if(task.params.weight > this._params.weightLimit) {
            throw Error('task with weight of ' +
                task.params.weight +
                ' can\'t be performed in queue with limit of ' +
                this._params.weightLimit);
        }

        this._enqueueTask(task);
        this._isStopped || this._scheduleRun();

        task.defer.promise().always(
            function() {
                this._stats.processingTasksCount--;
                this._stats.processedTasksCount++;
            },
            this);

        return task.defer.promise();
    }
1019556
def delay(self, func, args=None, kwargs=None, queue=None,
              hard_timeout=None, unique=None, lock=None, lock_key=None,
              when=None, retry=None, retry_on=None, retry_method=None,
              max_queue_size=None):
        

        task = Task(self, func, args=args, kwargs=kwargs, queue=queue,
                    hard_timeout=hard_timeout, unique=unique,
                    lock=lock, lock_key=lock_key,
                    retry=retry, retry_on=retry_on, retry_method=retry_method)

        task.delay(when=when, max_queue_size=max_queue_size)

        return task
486186
private Runnable decorateTask(Runnable task, boolean isRepeatingTask) {
        Runnable result = TaskUtils.decorateTaskWithErrorHandler(task, this.errorHandler, isRepeatingTask);
        if (this.enterpriseConcurrentScheduler) {
            result = ManagedTaskBuilder.buildManagedTask(result, task.toString());
        }
        return result;
    }
1345226
def retry(self, func, partition_id, retry_message, final_failure_message, max_retries, host_id):
        
        loop = asyncio.new_event_loop()
        loop.run_until_complete(self.retry_async(func, partition_id, retry_message,
                                                 final_failure_message, max_retries, host_id))
352651
def retry(self):
        

        logger.info('Job {0} retrying all failed tasks'.format(self.name))
        self.initialize_snapshot()

        failed_task_names = []
        for task_name, log in self.run_log['tasks'].items():
            if log.get('success', True) == False:
                failed_task_names.append(task_name)

        if len(failed_task_names) == 0:
            raise DagobahError('no failed tasks to retry')

        self._set_status('running')
        self.run_log['last_retry_time'] = datetime.utcnow()

        logger.debug('Job {0} seeding run logs'.format(self.name))
        for task_name in failed_task_names:
            self._put_task_in_run_log(task_name)
            self.tasks[task_name].start()

        self._commit_run_log()
607060
def retry_failed(FailAdmin, request, queryset):
    """"""
    for task in queryset:
        async_task(task.func, *task.args or (), hook=task.hook, **task.kwargs or {})
        task.delete()
361044
def reset(self):
        

        logger.debug('Resetting task {0}'.format(self.name))

        self.stdout_file = os.tmpfile()
        self.stderr_file = os.tmpfile()

        self.stdout = ""
        self.stderr = ""

        self.started_at = None
        self.completed_at = None
        self.successful = None

        self.terminate_sent = False
        self.kill_sent = False
        self.remote_failure = False
989933
@Service
    public String requeuePendingTask(String taskType) {
        return String.valueOf(executionService.requeuePendingTasks(taskType));
    }
651430
def update_status(self):
        
        task = self.make_request(
            TaskRunFailed,
            href=self.href)

        return Task(task)
409622
def _handle_task(self):
        """"""
        headers = self.request.headers

        message = None
        try:
            status_code, output = process_async_task(
                headers, self.request.body)
        except AbortAndRestart as restart:
            # Async retry status code
            status_code = 549
            message = 'Retry Async Task'
            output = str(restart)

        self.response.set_status(status_code, message)
        self.response.out.write(output)