| 146 | |
| 147 | #ifdef ARROW_ENABLE_THREADING |
| 148 | Status SerialExecutor::SpawnReal(TaskHints hints, FnOnce<void()> task, |
| 149 | StopToken stop_token, StopCallback&& stop_callback) { |
| 150 | # ifdef ARROW_WITH_OPENTELEMETRY |
| 151 | // Wrap the task to propagate a parent tracing span to it |
| 152 | // XXX should there be a generic utility in tracing_internal.h for this? |
| 153 | task = [func = std::move(task), |
| 154 | active_span = |
| 155 | ::arrow::internal::tracing::GetTracer()->GetCurrentSpan()]() mutable { |
| 156 | auto scope = ::arrow::internal::tracing::GetTracer()->WithActiveSpan(active_span); |
| 157 | std::move(func)(); |
| 158 | }; |
| 159 | # endif |
| 160 | // While the SerialExecutor runs tasks synchronously on its main thread, |
| 161 | // SpawnReal may be called from external threads (e.g. when transferring back |
| 162 | // from blocking I/O threads), so we need to keep the state alive *and* to |
| 163 | // lock its contents. |
| 164 | // |
| 165 | // Note that holding the lock while notifying the condition variable may |
| 166 | // not be sufficient, as some exit paths in the main thread are unlocked. |
| 167 | auto state = state_; |
| 168 | { |
| 169 | std::lock_guard<std::mutex> lk(state->mutex); |
| 170 | if (state_->finished) { |
| 171 | return Status::Invalid( |
| 172 | "Attempt to schedule a task on a serial executor that has already finished or " |
| 173 | "been abandoned"); |
| 174 | } |
| 175 | state->task_queue.push( |
| 176 | QueuedTask{{std::move(task), std::move(stop_token), std::move(stop_callback)}, |
| 177 | hints.priority, |
| 178 | state_->spawned_tasks_count_++}); |
| 179 | } |
| 180 | state->wait_for_tasks.notify_one(); |
| 181 | return Status::OK(); |
| 182 | } |
| 183 | |
| 184 | void SerialExecutor::Finish() { |
| 185 | auto state = state_; |