Method AddTask

cpp/src/arrow/util/async_util.cc:302–329

  }

  bool AddTask(std::unique_ptr<Task> task) override {
    std::unique_lock lk(mutex_);
    // If the queue isn't empty then don't even try and acquire the throttle
    // We can safely assume it is either blocked or in the middle of trying to
    // alert a queued task.
    if (!queue_->Empty()) {
      queue_->Push(std::move(task));
      return true;
    }
    int latched_cost = std::min(task->cost(), throttle_->Capacity());
    std::optional<Future<>> maybe_backoff = throttle_->TryAcquire(latched_cost);
    if (maybe_backoff) {
#ifdef ARROW_WITH_OPENTELEMETRY
      TraceTaskQueued(task.get(), span());
#endif
      queue_->Push(std::move(task));
      lk.unlock();
      maybe_backoff->AddCallback([weak_self = weak_from_this()](const Status& st) {
        if (auto self = weak_self.lock(); self && st.ok()) {
          self->ContinueTasks();
        }
      });
      return true;
    } else {
      lk.unlock();
      return SubmitTask(std::move(task), latched_cost, /*in_continue=*/false);
    }
  }

  void Pause() override { throttle_->Pause(); }
  void Resume() override { throttle_->Resume(); }