| 853 | } |
| 854 | |
| 855 | void Task::createAndStartDrivers(uint32_t concurrentSplitGroups) { |
| 856 | checkExecutionMode(ExecutionMode::kParallel); |
| 857 | std::unique_lock<std::timed_mutex> l(mutex_); |
| 858 | BOLT_CHECK( |
| 859 | isRunningLocked(), |
| 860 | "Task {} has already been terminated before start: {}", |
| 861 | taskId_, |
| 862 | errorMessageLocked()); |
| 863 | BOLT_CHECK(!driverFactories_.empty()); |
| 864 | BOLT_CHECK_EQ(concurrentSplitGroups_, 1); |
| 865 | BOLT_CHECK(drivers_.empty()); |
| 866 | |
| 867 | concurrentSplitGroups_ = concurrentSplitGroups; |
| 868 | // Pre-allocates slots for maximum possible number of drivers. |
| 869 | if (numDriversPerSplitGroup_ > 0) { |
| 870 | drivers_.resize(numDriversPerSplitGroup_ * concurrentSplitGroups_); |
| 871 | } |
| 872 | |
| 873 | // First, create drivers for ungrouped execution. |
| 874 | if (numDriversUngrouped_ > 0) { |
| 875 | createSplitGroupStateLocked(kUngroupedGroupId); |
| 876 | // Create drivers. |
| 877 | std::vector<std::shared_ptr<Driver>> drivers = |
| 878 | createDriversLocked(kUngroupedGroupId); |
| 879 | if (pool_->reservedBytes() != 0) { |
| 880 | BOLT_FAIL( |
| 881 | "Unexpected memory pool allocations during task[{}] driver initialization: {}", |
| 882 | taskId_, |
| 883 | pool_->treeMemoryUsage()); |
| 884 | } |
| 885 | |
| 886 | // Prevent the connecting structures from being cleaned up before all |
| 887 | // split groups are finished during the grouped execution mode. |
| 888 | if (isGroupedExecution()) { |
| 889 | splitGroupStates_[kUngroupedGroupId].mixedExecutionMode = true; |
| 890 | } |
| 891 | |
| 892 | // Slots in the front are used by grouped execution drivers. Ungrouped |
| 893 | // execution drivers come after these. |
| 894 | if (drivers_.empty()) { |
| 895 | drivers_ = std::move(drivers); |
| 896 | } else { |
| 897 | drivers_.reserve(drivers_.size() + numDriversUngrouped_); |
| 898 | for (auto& driver : drivers) { |
| 899 | drivers_.emplace_back(std::move(driver)); |
| 900 | } |
| 901 | } |
| 902 | |
| 903 | // Set and start all Drivers together inside 'mutex_' so that |
| 904 | // cancellations and pauses have the well-defined timing. For example, do |
| 905 | // not pause and restart a task while it is still adding Drivers. |
| 906 | // |
| 907 | for (auto it = drivers_.end() - numDriversUngrouped_; it != drivers_.end(); |
| 908 | ++it) { |
| 909 | if (*it) { |
| 910 | ++numRunningDrivers_; |
| 911 | Driver::enqueue(*it); |
| 912 | } |
nothing calls this directly
no test coverage detected