* Worker entry point * * As long as worker thread is running, pull tasks off the task queue and * execute. */
| 230 | * execute. |
| 231 | */ |
| 232 | void run() override { |
| 233 | Guard g(manager_->mutex_); |
| 234 | |
| 235 | /** |
| 236 | * This method has three parts; one is to check for and account for |
| 237 | * admitting a task which happens under a lock. Then the lock is released |
| 238 | * and the task itself is executed. Finally we do some accounting |
| 239 | * under lock again when the task completes. |
| 240 | */ |
| 241 | |
| 242 | /** |
| 243 | * Admitting |
| 244 | */ |
| 245 | |
| 246 | /** |
| 247 | * Increment worker semaphore and notify manager if worker count reached |
| 248 | * desired max |
| 249 | */ |
| 250 | bool active = manager_->workerCount_ < manager_->workerMaxCount_; |
| 251 | if (active) { |
| 252 | if (++manager_->workerCount_ == manager_->workerMaxCount_) { |
| 253 | manager_->workerMonitor_.notify(); |
| 254 | } |
| 255 | } |
| 256 | |
| 257 | while (active) { |
| 258 | /** |
| 259 | * While holding manager monitor block for non-empty task queue (Also |
| 260 | * check that the thread hasn't been requested to stop). Once the queue |
| 261 | * is non-empty, dequeue a task, release monitor, and execute. If the |
| 262 | * worker max count has been decremented such that we exceed it, mark |
| 263 | * ourself inactive, decrement the worker count and notify the manager |
| 264 | * (technically we're notifying the next blocked thread but eventually |
| 265 | * the manager will see it. |
| 266 | */ |
| 267 | active = isActive(); |
| 268 | |
| 269 | while (active && manager_->tasks_.empty()) { |
| 270 | manager_->idleCount_++; |
| 271 | manager_->monitor_.wait(); |
| 272 | active = isActive(); |
| 273 | manager_->idleCount_--; |
| 274 | } |
| 275 | |
| 276 | shared_ptr<ThreadManager::Task> task; |
| 277 | |
| 278 | if (active) { |
| 279 | if (!manager_->tasks_.empty()) { |
| 280 | task = manager_->tasks_.front(); |
| 281 | manager_->tasks_.pop_front(); |
| 282 | if (task->state_ == ThreadManager::Task::WAITING) { |
| 283 | // If the state is changed to anything other than EXECUTING or TIMEDOUT here |
| 284 | // then the execution loop needs to be changed below. |
| 285 | task->state_ = |
| 286 | (task->getExpireTime() && *(task->getExpireTime()) < std::chrono::steady_clock::now()) ? |
| 287 | ThreadManager::Task::TIMEDOUT : |
| 288 | ThreadManager::Task::EXECUTING; |
| 289 | } |