| 769 | } |
| 770 | |
| 771 | void Task::start(uint32_t maxDrivers, uint32_t concurrentSplitGroups) { |
| 772 | bytedance::bolt::process::ThreadDebugInfo threadDebugInfo{ |
| 773 | queryCtx()->queryId(), taskId_, nullptr}; |
| 774 | bytedance::bolt::process::ScopedThreadDebugInfo scopedInfo(threadDebugInfo); |
| 775 | checkExecutionMode(ExecutionMode::kParallel); |
| 776 | |
| 777 | try { |
| 778 | BOLT_CHECK_GE( |
| 779 | maxDrivers, |
| 780 | 1, |
| 781 | "maxDrivers parameter must be greater then or equal to 1"); |
| 782 | BOLT_CHECK_GE( |
| 783 | concurrentSplitGroups, |
| 784 | 1, |
| 785 | "concurrentSplitGroups parameter must be greater then or equal to 1"); |
| 786 | |
| 787 | { |
| 788 | std::unique_lock<std::timed_mutex> l(mutex_); |
| 789 | taskStats_.executionStartTimeMs = getCurrentTimeMs(); |
| 790 | if (!isRunningLocked()) { |
| 791 | LOG(WARNING) << "Task " << taskId_ |
| 792 | << " has already been terminated before start: " |
| 793 | << errorMessageLocked(); |
| 794 | return; |
| 795 | } |
| 796 | createDriverFactoriesLocked(maxDrivers); |
| 797 | } |
| 798 | initializePartitionOutput(); |
| 799 | createAndStartDrivers(concurrentSplitGroups); |
| 800 | } catch (const std::exception&) { |
| 801 | if (isRunning()) { |
| 802 | setError(std::current_exception()); |
| 803 | } else { |
| 804 | maybeRemoveFromOutputBufferManager(); |
| 805 | { |
| 806 | // NOTE: the async task error might be triggered in the middle of task |
| 807 | // start processing, and we need to mark all the drivers have been |
| 808 | // finished. |
| 809 | std::unique_lock<std::timed_mutex> l(mutex_); |
| 810 | BOLT_CHECK_EQ(numRunningDrivers_, 0); |
| 811 | BOLT_CHECK_EQ(numFinishedDrivers_, 0); |
| 812 | numFinishedDrivers_ = numTotalDrivers_; |
| 813 | } |
| 814 | } |
| 815 | throw; |
| 816 | } |
| 817 | } |
| 818 | |
| 819 | void Task::checkExecutionMode(ExecutionMode mode) { |
| 820 | BOLT_CHECK_EQ(mode, mode_, "Inconsistent task execution mode."); |