Try to offer as many runnable Tasks to the SiteTaskerQueue as possible.
@param txnId The transaction ID of the TransactionTask which is completing and causing the flush
@return the number of TransactionTasks queued to the SiteTaskerQueue
358530
synchronized int flush(long txnId)
    {
        if (tmLog.isDebugEnabled()) {
            tmLog.debug("Flush backlog with txnId:" + TxnEgo.txnIdToString(txnId) +
                    ", backlog head txnId is:" + (m_backlog.isEmpty()? "empty" : TxnEgo.txnIdToString(m_backlog.getFirst().getTxnId()))
                    );
        }

        int offered = 0;
        // If the first entry of the backlog is a completed transaction, clear it so it no longer
        // blocks the backlog then iterate the backlog for more work.
        //
        // Note the kooky corner case where a multi-part transaction can actually have multiple outstanding
        // tasks. At first glance you would think that because the relationship is request response there
        // can be only one outstanding task for a given multi-part transaction.
        //
        // That isn't true.
        //
        // A rollback can cause there to be a fragment task as well as a rollback
        // task. The rollback is generated asynchronously by another partition.
        // If we don't flush all the associated tasks now then flush won't be called again because it is waiting
        // for the complete transaction task that is languishing in the queue to do the flush post multi-part.
        // It can't be called eagerly because that would destructively flush single parts as well.
        if (m_backlog.isEmpty() || !m_backlog.getFirst().getTransactionState().isDone()) {
            return offered;
        }

        // Add a guard to protect the scenario that backlog been flushed multiple times for same txnId
        if (m_backlog.getFirst().getTxnId() != txnId) {
            return offered;
        }

        m_backlog.removeFirst();
        Iterator<TransactionTask> iter = m_backlog.iterator();
        while (iter.hasNext()) {
            TransactionTask task = iter.next();
            long lastQueuedTxnId = task.getTxnId();
            if (task.needCoordination() && m_scoreboardEnabled) {
                coordinatedTaskQueueOffer(task);
            } else {
                taskQueueOffer(task);
            }
            ++offered;
            if (task.getTransactionState().isSinglePartition()) {
                // single part can be immediately removed and offered
                iter.remove();
                continue;
            }
            else {
                // leave the mp fragment at the head of the backlog but
                // iterate and take care of the kooky case explained above.
                while (iter.hasNext()) {
                    task = iter.next();
                    if (task.getTxnId() == lastQueuedTxnId) {
                        iter.remove();
                        if (task.needCoordination() && m_scoreboardEnabled) {
                            coordinatedTaskQueueOffer(task);
                        } else {
                            taskQueueOffer(task);
                        }
                        ++offered;
                    }
                }
                break;
            }
        }
        return offered;
    }
488827
@Exported
    public int getIdleExecutors() {
        int r=0;
        for (Node n : getNodes()) {
            Computer c = n.toComputer();
            if(c!=null && (c.isOnline() || c.isConnecting()) && c.isAcceptingTasks())
                r += c.countIdle();
        }
        return r;
    }
1098911
@Override
    public void requeue(Task task) {
        this.inprocess.remove(task);
        task.incrementAttempts();
        this.queue.add(task);
    }
932378
public void quiesceStandby(long txId) throws IOException {
    if (currentAvatar != Avatar.STANDBY) {
      throw new IOException("This is not the standby avatar");
    }
    standby.quiesce(txId);
  }
362471
void completeWork(long txnId)
    {
        if (m_shuttingDown) {
            return;
        }

        MpRoSiteContext site = m_busySites.remove(txnId);
        if (site == null) {
            throw new RuntimeException("No busy site for txnID: " + txnId + " found, shouldn't happen.");
        }
        // check the catalog versions, only push back onto idle if the catalog hasn't changed
        // otherwise, just let it get garbage collected and let doWork() construct new ones for the
        // pool with the updated catalog.
        if (site.getCatalogCRC() == m_catalogContext.getCatalogCRC()
                && site.getCatalogVersion() == m_catalogContext.catalogVersion) {
            m_idleSites.push(site);
        }
        else {
            site.shutdown();
            m_allSites.remove(site);
        }
    }
1136133
public void incRunningTasks(String poolName, TaskType type, int inc) {
    Map<String, Integer> runningMap = (type == TaskType.MAP ?
                                       poolRunningMaps : poolRunningReduces);
    if (!runningMap.containsKey(poolName)) {
      runningMap.put(poolName, 0);
    }
    int runningTasks = runningMap.get(poolName) + inc;
    runningMap.put(poolName, runningTasks);
  }
405708
public TransactionInfoBaseMessage nextTask() throws IOException {
        if (!hasMoreEntries()) {
            return null;
        }

        ByteBuffer bb = m_container.b();

        int position = bb.position();
        int length = bb.getInt();
        long sourceHSId = bb.getLong();
        VoltDbMessageFactory factory = new VoltDbMessageFactory();
        /*
         * create a new buffer that just contains the message, deserialization
         * of the messsage may assert on the capacity of the buffer
         */
        final int oldLimit = bb.limit();
        bb.limit(bb.position() + length);
        ByteBuffer slice = bb.slice();
        bb.limit(oldLimit);
        VoltMessage msg = factory.createMessageFromBuffer(slice, sourceHSId);
        // createMessageFromBuffer() doesn't move the position pointer, set it here
        bb.position(position + length + 8 + 4); // sourceHSId + buf len
        return (TransactionInfoBaseMessage) msg;
    }
989933
@Service
    public String requeuePendingTask(String taskType) {
        return String.valueOf(executionService.requeuePendingTasks(taskType));
    }
1195001
protected ResultQueueEntry<FlatRow> getNext() throws IOException {
    ResultQueueEntry<FlatRow> queueEntry;
    try {
      queueEntry = resultQueue.take();
      // Should never happen
      if (queueEntry == null) {
        throw new IllegalStateException("Timed out awaiting next sync rows");
      }
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new IOException("Interrupted while waiting for next result", e);
    }

    return queueEntry;
  }
801960
private List<TaskDef> findAllTaskDefs(Connection tx) {
        final String READ_ALL_TASKDEF_QUERY = "SELECT json_data FROM meta_task_def";

        return query(tx, READ_ALL_TASKDEF_QUERY, q -> q.executeAndFetch(TaskDef.class));
    }
1140490
protected final boolean runAllTasksFrom(Queue<Runnable> taskQueue) {
        Runnable task = pollTaskFrom(taskQueue);
        if (task == null) {
            return false;
        }
        for (;;) {
            safeExecute(task);
            task = pollTaskFrom(taskQueue);
            if (task == null) {
                return true;
            }
        }
    }
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;
  }
1156866
def n_executions(self):
        
        pipeline = self.tiger.connection.pipeline()
        pipeline.exists(self.tiger._key('task', self.id))
        pipeline.llen(self.tiger._key('task', self.id, 'executions'))
        exists, n_executions = pipeline.execute()
        if not exists:
            raise TaskNotFound('Task {} not found.'.format(
                self.id
            ))
        return n_executions
556638
private void refreshTaskDefs() {
        try {
            withTransaction(tx -> {
                Map<String, TaskDef> map = new HashMap<>();
                findAllTaskDefs(tx).forEach(taskDef -> map.put(taskDef.getName(), taskDef));

                synchronized (taskDefCache) {
                    taskDefCache.clear();
                    taskDefCache.putAll(map);
                }

                if (logger.isTraceEnabled()) {
                    logger.trace("Refreshed {} TaskDefs", taskDefCache.size());
                }
            });
        } catch (Exception e){
            Monitors.error(className, "refreshTaskDefs");
            logger.error("refresh TaskDefs failed ", e);
        }
    }
253605
@Override
    protected void nibble(Queue<StorageStatsMorsel> queue) {
        StorageStatsMorsel morsel = queue.peek();
        String storeId = morsel.getStoreId();

        StorageProvider store = getStorageProvider(morsel.getAccount(), storeId);

        int maxTaskQueueSize = getMaxTaskQueueSize();
        int taskQueueSize = getTaskQueue().size();

        if (taskQueueSize >= maxTaskQueueSize) {
            log.info("Task queue size ({}) has reached or exceeded max size ({}).",
                     taskQueueSize,
                     maxTaskQueueSize);
        } else {

            addTask(morsel, store);
            queue.remove(morsel);
        }
    }
125862
protected boolean runAllTasks(long timeoutNanos) {
        fetchFromScheduledTaskQueue();
        Runnable task = pollTask();
        if (task == null) {
            afterRunningAllTasks();
            return false;
        }

        final long deadline = ScheduledFutureTask.nanoTime() + timeoutNanos;
        long runTasks = 0;
        long lastExecutionTime;
        for (;;) {
            safeExecute(task);

            runTasks ++;

            // Check timeout every 64 tasks because nanoTime() is relatively expensive.
            // XXX: Hard-coded value - will make it configurable if it is really a problem.
            if ((runTasks & 0x3F) == 0) {
                lastExecutionTime = ScheduledFutureTask.nanoTime();
                if (lastExecutionTime >= deadline) {
                    break;
                }
            }

            task = pollTask();
            if (task == null) {
                lastExecutionTime = ScheduledFutureTask.nanoTime();
                break;
            }
        }

        afterRunningAllTasks();
        this.lastExecutionTime = lastExecutionTime;
        return true;
    }
743328
public void abortTransactionFailed(String scope, String streamName, String txnId) {
        DYNAMIC_LOGGER.incCounterValue(globalMetricName(ABORT_TRANSACTION_FAILED), 1);
        DYNAMIC_LOGGER.incCounterValue(ABORT_TRANSACTION_FAILED, 1, streamTags(scope, streamName));
        DYNAMIC_LOGGER.incCounterValue(ABORT_TRANSACTION_FAILED, 1, transactionTags(scope, streamName, txnId));
    }