| 40 | } |
| 41 | |
| 42 | void TaskScheduler::scheduleTaskAndWaitOrError(const std::shared_ptr<Task>& task, |
| 43 | processor::ExecutionContext* context, bool launchNewWorkerThread) { |
| 44 | for (auto& dependency : task->children) { |
| 45 | scheduleTaskAndWaitOrError(dependency, context); |
| 46 | if (dependency->terminate()) { |
| 47 | return; |
| 48 | } |
| 49 | } |
| 50 | std::thread newWorkerThread; |
| 51 | if (launchNewWorkerThread) { |
| 52 | // Note that newWorkerThread is not executing yet. However, we still call |
| 53 | // task->registerThread() function because the call in the next line will guarantee |
| 54 | // that the thread starts working on it. registerThread() function only increases the |
| 55 | // numThreadsRegistered field of the task, tt does not keep track of the thread ids or |
| 56 | // anything specific to the thread. |
| 57 | task->registerThread(); |
| 58 | newWorkerThread = std::thread(runTask, task.get()); |
| 59 | } |
| 60 | auto scheduledTask = pushTaskIntoQueue(task); |
| 61 | cv.notify_all(); |
| 62 | std::unique_lock<std::mutex> taskLck{task->taskMtx, std::defer_lock}; |
| 63 | while (true) { |
| 64 | taskLck.lock(); |
| 65 | bool timedWait = false; |
| 66 | auto timeout = 0u; |
| 67 | if (task->isCompletedNoLock()) { |
| 68 | // Note: we do not remove completed tasks from the queue in this function. They will be |
| 69 | // removed by the worker threads when they traverse down the queue for a task to work on |
| 70 | // (see getTaskAndRegister()). |
| 71 | taskLck.unlock(); |
| 72 | break; |
| 73 | } |
| 74 | if (context->clientContext->hasTimeout()) { |
| 75 | timeout = context->clientContext->getTimeoutRemainingInMS(); |
| 76 | if (timeout == 0) { |
| 77 | context->clientContext->interrupt(); |
| 78 | } else { |
| 79 | timedWait = true; |
| 80 | } |
| 81 | } else if (task->hasExceptionNoLock()) { |
| 82 | // Interrupt tasks that errored, so other threads can stop working on them early. |
| 83 | context->clientContext->interrupt(); |
| 84 | } |
| 85 | if (timedWait) { |
| 86 | task->cv.wait_for(taskLck, std::chrono::milliseconds(timeout)); |
| 87 | } else { |
| 88 | task->cv.wait(taskLck); |
| 89 | } |
| 90 | taskLck.unlock(); |
| 91 | } |
| 92 | if (launchNewWorkerThread) { |
| 93 | newWorkerThread.join(); |
| 94 | } |
| 95 | if (task->hasException()) { |
| 96 | removeErroringTask(scheduledTask->ID); |
| 97 | std::rethrow_exception(task->getExceptionPtr()); |
| 98 | } |
| 99 | } |
no test coverage detected