MCPcopy Create free account
hub / github.com/apache/arrow / AddTask

Method AddTask

cpp/src/arrow/util/async_util.cc:303–330  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

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(); }

Callers

nothing calls this directly

Calls 11

ContinueTasksMethod · 0.95
TraceTaskQueuedFunction · 0.85
SubmitTaskClass · 0.85
EmptyMethod · 0.45
PushMethod · 0.45
costMethod · 0.45
CapacityMethod · 0.45
TryAcquireMethod · 0.45
getMethod · 0.45
AddCallbackMethod · 0.45
okMethod · 0.45

Tested by

no test coverage detected