| 90 | } |
| 91 | |
| 92 | bool ExecutorTaskScheduler::waitForScheduling( |
| 93 | std::shared_ptr<Task>& task, |
| 94 | bool doSchedule) { |
| 95 | bool waitInQueue = false; |
| 96 | |
| 97 | // taskid deduplication |
| 98 | bool isInserted = tasksIds_.wlock()->emplace(task->taskId()).second; |
| 99 | (*allTasks_.wlock())[task->taskId()].emplace_back(task); |
| 100 | if (isInserted) { |
| 101 | task->setSourceFragmentUnlocked(); |
| 102 | } else { |
| 103 | return waitInQueue; |
| 104 | } |
| 105 | |
| 106 | std::lock_guard<std::mutex> l(mutex_); |
| 107 | ++numRunningTask_; |
| 108 | if (doSchedule && numRunningTask_ > currentConcurrency_) { |
| 109 | taskQueue_.push(task); |
| 110 | waitInQueue = true; |
| 111 | --numRunningTask_; |
| 112 | } |
| 113 | |
| 114 | if (!waitInQueue) { |
| 115 | startTaskTracking(task->taskId()); |
| 116 | } |
| 117 | |
| 118 | VLOG(1) << __FUNCTION__ << ": task " << task->taskId() |
| 119 | << (waitInQueue ? " in waiting queue" : " not in waiting queue") |
| 120 | << ", numRunningTask_ = " << numRunningTask_ |
| 121 | << ", taskQueue_.size() = " << taskQueue_.size(); |
| 122 | return waitInQueue; |
| 123 | } |
| 124 | |
| 125 | void ExecutorTaskScheduler::collectTaskStatsUnlocked( |
| 126 | const TaskStats& taskStats, |
no test coverage detected