| 2041 | } |
| 2042 | |
| 2043 | ContinueFuture Task::terminate(TaskState terminalState) { |
| 2044 | std::vector<std::shared_ptr<Driver>> offThreadDrivers; |
| 2045 | EventCompletionNotifier taskCompletionNotifier; |
| 2046 | EventCompletionNotifier stateChangeNotifier; |
| 2047 | std::vector<std::shared_ptr<ExchangeClient>> exchangeClients; |
| 2048 | { |
| 2049 | std::lock_guard<std::timed_mutex> l(mutex_); |
| 2050 | if (taskStats_.executionEndTimeMs == 0) { |
| 2051 | taskStats_.executionEndTimeMs = getCurrentTimeMs(); |
| 2052 | } |
| 2053 | if (queryCtx_->queryConfig().enableDynamicConcurrencyAdjustment()) { |
| 2054 | ExecutorTaskScheduler::instance().scheduleNewTasksIfAny( |
| 2055 | shared_from_this(), |
| 2056 | (terminalState == TaskState::kFinished), |
| 2057 | queryCtx_->queryConfig().enableBoltTaskScheduling()); |
| 2058 | } |
| 2059 | if (not isRunningLocked()) { |
| 2060 | return makeFinishFutureLocked("Task::terminate"); |
| 2061 | } |
| 2062 | |
| 2063 | state_ = terminalState; |
| 2064 | BOLT_CHECK_EQ( |
| 2065 | taskStats_.terminationTimeMs, |
| 2066 | 0, |
| 2067 | "Termination time has already been set, this should only happen once."); |
| 2068 | taskStats_.terminationTimeMs = getCurrentTimeMs(); |
| 2069 | if (state_ == TaskState::kCanceled || state_ == TaskState::kAborted) { |
| 2070 | try { |
| 2071 | BOLT_FAIL( |
| 2072 | state_ == TaskState::kCanceled ? "Cancelled" |
| 2073 | : "Aborted for external error"); |
| 2074 | } catch (const std::exception&) { |
| 2075 | exception_ = std::current_exception(); |
| 2076 | } |
| 2077 | } |
| 2078 | |
| 2079 | LOG(INFO) << "Terminating task " << taskId() << " with state " |
| 2080 | << taskStateString(state_) << " after running for " |
| 2081 | << succinctMillis(timeSinceStartMsLocked()); |
| 2082 | |
| 2083 | taskCompletionNotifier.activate( |
| 2084 | std::move(taskCompletionPromises_), [&]() { onTaskCompletion(); }); |
| 2085 | stateChangeNotifier.activate(std::move(stateChangePromises_)); |
| 2086 | |
| 2087 | // Update the total number of drivers if we were cancelled. |
| 2088 | numTotalDrivers_ = seenSplitGroups_.size() * numDriversPerSplitGroup_ + |
| 2089 | numDriversUngrouped_; |
| 2090 | // Drivers that are on thread will see this at latest when they go off |
| 2091 | // thread. |
| 2092 | terminateRequested_ = true; |
| 2093 | // The drivers that are on thread will go off thread in time and |
| 2094 | // 'numRunningDrivers_' is cleared here so that this is 0 right |
| 2095 | // after terminate as tests expect. |
| 2096 | numRunningDrivers_ = 0; |
| 2097 | for (auto& driver : drivers_) { |
| 2098 | if (driver) { |
| 2099 | if (enterForTerminateLocked(driver->state()) == |
| 2100 | StopReason::kTerminate) { |