| 301 | } |
| 302 | |
| 303 | bool AddTask(std::unique_ptr<Task> task) override { |
| 304 | std::unique_lock lk(mutex_); |
| 305 | // If the queue isn't empty then don't even try and acquire the throttle |
| 306 | // We can safely assume it is either blocked or in the middle of trying to |
| 307 | // alert a queued task. |
| 308 | if (!queue_->Empty()) { |
| 309 | queue_->Push(std::move(task)); |
| 310 | return true; |
| 311 | } |
| 312 | int latched_cost = std::min(task->cost(), throttle_->Capacity()); |
| 313 | std::optional<Future<>> maybe_backoff = throttle_->TryAcquire(latched_cost); |
| 314 | if (maybe_backoff) { |
| 315 | #ifdef ARROW_WITH_OPENTELEMETRY |
| 316 | TraceTaskQueued(task.get(), span()); |
| 317 | #endif |
| 318 | queue_->Push(std::move(task)); |
| 319 | lk.unlock(); |
| 320 | maybe_backoff->AddCallback([weak_self = weak_from_this()](const Status& st) { |
| 321 | if (auto self = weak_self.lock(); self && st.ok()) { |
| 322 | self->ContinueTasks(); |
| 323 | } |
| 324 | }); |
| 325 | return true; |
| 326 | } else { |
| 327 | lk.unlock(); |
| 328 | return SubmitTask(std::move(task), latched_cost, /*in_continue=*/false); |
| 329 | } |
| 330 | } |
| 331 | |
| 332 | void Pause() override { throttle_->Pause(); } |
| 333 | void Resume() override { throttle_->Resume(); } |
nothing calls this directly
no test coverage detected