Method RunTasksOnAllExecutors

cpp/src/arrow/util/thread_pool.cc:294–356

ARROW_ENABLE_THREADING

}
#else   // ARROW_ENABLE_THREADING
bool SerialExecutor::RunTasksOnAllExecutors() {
  auto globalState = GetSerialExecutorGlobalState();
  // if the previously called executor was deleted, ignore last_called_executor
  if (globalState->last_called_executor != NULL &&
      globalState->all_executors.count(globalState->last_called_executor) == 0) {
    globalState->last_called_executor = NULL;
  }
  bool run_task = true;
  bool keep_going = true;
  while (keep_going) {
    run_task = false;
    keep_going = false;
    for (auto it = globalState->all_executors.begin();
         it != globalState->all_executors.end(); ++it) {
      if (globalState->last_called_executor != NULL) {
        // always rerun loop if we have a last_called_executor, otherwise
        // we may drop out before every executor has been checked
        keep_going = true;
        if (globalState->all_executors.count(globalState->last_called_executor) == 0 ||
            globalState->last_called_executor == *it) {
          // found the last one (or it doesn't exist ih the set any more)
          // now we can start running things
          globalState->last_called_executor = NULL;
        }
        // skip until after we have seen the last executor we called
        // so that we do things nicely in turn
        continue;
      }
      auto exe = *it;
      // don't make more reentrant calls inside an
      // executor than the number of concurrent tasks set on a threadpool, or
      // 1 in the case of a serialexecutor -
      // this is because users will expect a serial executor not to be able to
      // run the next task until the current one is finished (and a threadpool
      // only to be able to run a certain number of tasks concurrently)
      if (exe->state_->tasks_running >= exe->state_->max_tasks_running) {
        continue;
      }
      if (exe->state_->paused == false && exe->state_->task_queue.empty() == false) {
        SerialExecutor* old_exe = globalState->current_executor;
        globalState->current_executor = exe;
        Task task = std::move(const_cast<Task&>(exe->state_->task_queue.top().task));
        exe->state_->task_queue.pop();
        run_task = true;
        exe->state_->tasks_running += 1;
        if (!task.stop_token.IsStopRequested()) {
          std::move(task.callable)();
        } else {
          if (task.stop_callback) {
            std::move(task.stop_callback)(task.stop_token.Poll());
          }
        }
        exe->state_->tasks_running -= 1;
        globalState->current_executor = old_exe;

        globalState->last_called_executor = exe;
        keep_going = false;
        break;