Method RunLoop

cpp/src/arrow/util/thread_pool.cc:257–291

#ifdef ARROW_ENABLE_THREADING

void SerialExecutor::RunLoop() {
  // This is called from the SerialExecutor's main thread, so the
  // state is guaranteed to be kept alive.
  std::unique_lock<std::mutex> lk(state_->mutex);
  state_->current_thread = std::this_thread::get_id();
  // If paused we break out immediately.  If finished we only break out
  // when all work is done.
  while (!state_->paused && !(state_->finished && state_->task_queue.empty())) {
    // The inner loop is to check if we need to sleep (e.g. while waiting on some
    // async task to finish from another thread pool).  We still need to check paused
    // because sometimes we will pause even with work leftover when processing
    // an async generator
    while (!state_->paused && !state_->task_queue.empty()) {
      Task task = std::move(const_cast<Task&>(state_->task_queue.top().task));
      state_->task_queue.pop();
      lk.unlock();
      if (!task.stop_token.IsStopRequested()) {
        std::move(task.callable)();
      } else {
        if (task.stop_callback) {
          std::move(task.stop_callback)(task.stop_token.Poll());
        }
        // Can't break here because there may be cleanup tasks down the chain we still
        // need to run.
      }
      lk.lock();
    }
    // In this case we must be waiting on work from external (e.g. I/O) executors.  Wait
    // for tasks to arrive (typically via transferred futures).
    state_->wait_for_tasks.wait(lk, [&] {
      return state_->paused || state_->finished || !state_->task_queue.empty();
    });
  }
  state_->current_thread = {};
}
#else   // ARROW_ENABLE_THREADING
bool SerialExecutor::RunTasksOnAllExecutors() {
  auto globalState = GetSerialExecutorGlobalState();