| 195 | } |
| 196 | |
| 197 | void DoSubmitTask(std::unique_ptr<Task> task) { |
| 198 | Result<Future<>> submit_result = (*task)(); |
| 199 | if (!submit_result.ok()) { |
| 200 | std::unique_lock<std::mutex> lk(mutex_); |
| 201 | running_tasks_--; |
| 202 | AbortUnlocked(submit_result.status(), std::move(lk)); |
| 203 | return; |
| 204 | } |
| 205 | // Capture `task` to keep it alive until finished |
| 206 | if (!submit_result->TryAddCallback([this, task_inner = std::move(task)]() mutable { |
| 207 | return [this, task_inner2 = std::move(task_inner)](const Status& st) mutable { |
| 208 | #ifdef ARROW_WITH_OPENTELEMETRY |
| 209 | TraceTaskFinished(task_inner2.get()); |
| 210 | #endif |
| 211 | // OnTaskFinished might trigger the scheduler to end. We want to ensure that |
| 212 | // is the very last thing that happens after all task destructors have run so |
| 213 | // we eagerly destroy the task first. |
| 214 | task_inner2.reset(); |
| 215 | OnTaskFinished(st); |
| 216 | }; |
| 217 | })) { |
| 218 | return OnTaskFinished(submit_result->status()); |
| 219 | } |
| 220 | } |
| 221 | |
| 222 | void MaybeEndUnlocked(std::unique_lock<std::mutex>&& lk) { |
| 223 | if (IsFullyFinished()) { |
nothing calls this directly
no test coverage detected