Method SubmitTask

cpp/src/arrow/util/async_util.cc:340–368

 private:
  bool SubmitTask(std::unique_ptr<Task> task, int latched_cost, bool in_continue) {
    // Wrap the task with a wrapper that runs it and then checks to see if there are any
    // queued tasks
    std::string_view name = task->name();
    return target_->AddSimpleTask(
        [latched_cost, in_continue, inner_task = std::move(task),
         self = shared_from_this()]() mutable -> Result<Future<>> {
          ARROW_ASSIGN_OR_RAISE(Future<> inner_fut, (*inner_task)());
          if (!inner_fut.TryAddCallback([&] {
                return [latched_cost,
                        weak_self = self->weak_from_this()](const Status& st) -> void {
                  if (auto self = weak_self.lock(); self && st.ok()) {
                    self->throttle_->Release(latched_cost);
                    self->ContinueTasks();
                  }
                };
              })) {
            // If the task is already finished then don't run ContinueTasks
            // if we are already running it so we can avoid stack overflow
            self->throttle_->Release(latched_cost);
            inner_task.reset();
            if (!in_continue) {
              self->ContinueTasks();
            }
          }
          return inner_fut;
        },
        name);
  }

  void ContinueTasks() {
    std::unique_lock lk(mutex_);