-----------------------------------------------------------------------------
| 119 | |
| 120 | //----------------------------------------------------------------------------- |
| 121 | void vtkThreadedCallbackQueue::SetNumberOfThreads(int numberOfThreads) |
| 122 | { |
| 123 | this->PushControl( |
| 124 | [this, numberOfThreads]() |
| 125 | { |
| 126 | int size = static_cast<int>(this->Threads.size()); |
| 127 | |
| 128 | std::lock_guard<std::mutex> destroyLock(this->DestroyMutex); |
| 129 | if (this->Destroying) |
| 130 | { |
| 131 | return; |
| 132 | } |
| 133 | if (size == numberOfThreads) |
| 134 | { |
| 135 | // Nothing to do |
| 136 | return; |
| 137 | } |
| 138 | // If we are expanding the number of threads, then we just need to spawn |
| 139 | // the missing threads. |
| 140 | else if (size < numberOfThreads) |
| 141 | { |
| 142 | this->NumberOfThreads = numberOfThreads; |
| 143 | |
| 144 | std::generate_n(std::back_inserter(this->Threads), numberOfThreads - size, |
| 145 | [this] |
| 146 | { |
| 147 | auto threadIndex = |
| 148 | std::make_shared<std::atomic_int>(static_cast<int>(this->Threads.size())); |
| 149 | auto thread = std::thread(ThreadWorker(this, threadIndex)); |
| 150 | { |
| 151 | std::lock_guard<std::mutex> threadIdLock(this->ThreadIdToIndexMutex); |
| 152 | this->ThreadIdToIndex.emplace(thread.get_id(), threadIndex); |
| 153 | } |
| 154 | return thread; |
| 155 | }); |
| 156 | } |
| 157 | // If we are shrinking the number of threads, let's notify all threads |
| 158 | // so the threads whose id is more than the updated NumberOfThreads terminate. |
| 159 | else |
| 160 | { |
| 161 | // If we have a thread index larger than the new number of threads, we swap ourself with |
| 162 | // thread 0. We now know we will live after this routine and can synchronize terminating |
| 163 | // threads ourselves. |
| 164 | { |
| 165 | std::unique_lock<std::mutex> lock(this->ThreadIdToIndexMutex); |
| 166 | std::atomic_int& threadIndex = *this->ThreadIdToIndex.at(std::this_thread::get_id()); |
| 167 | if (threadIndex && threadIndex >= numberOfThreads) |
| 168 | { |
| 169 | std::atomic_int& thread0Index = *this->ThreadIdToIndex.at(this->Threads[0].get_id()); |
| 170 | lock.unlock(); |
| 171 | |
| 172 | std::swap(this->Threads[threadIndex], this->Threads[0]); |
| 173 | |
| 174 | // Swapping the value of atomic ThreadIndex inside ThreadWorker. |
| 175 | int tmp = thread0Index; |
| 176 | thread0Index.exchange(threadIndex); |
| 177 | threadIndex = tmp; |
| 178 | } |