| 24 | |
| 25 | |
| 26 | void CScheduler::serviceQueue() |
| 27 | { |
| 28 | SetSyscallSandboxPolicy(SyscallSandboxPolicy::SCHEDULER); |
| 29 | WAIT_LOCK(newTaskMutex, lock); |
| 30 | ++nThreadsServicingQueue; |
| 31 | |
| 32 | // newTaskMutex is locked throughout this loop EXCEPT |
| 33 | // when the thread is waiting or when the user's function |
| 34 | // is called. |
| 35 | while (!shouldStop()) { |
| 36 | try { |
| 37 | while (!shouldStop() && taskQueue.empty()) { |
| 38 | // Wait until there is something to do. |
| 39 | newTaskScheduled.wait(lock); |
| 40 | } |
| 41 | |
| 42 | // Wait until either there is a new task, or until |
| 43 | // the time of the first item on the queue: |
| 44 | |
| 45 | while (!shouldStop() && !taskQueue.empty()) { |
| 46 | std::chrono::system_clock::time_point timeToWaitFor = taskQueue.begin()->first; |
| 47 | if (newTaskScheduled.wait_until(lock, timeToWaitFor) == std::cv_status::timeout) { |
| 48 | break; // Exit loop after timeout, it means we reached the time of the event |
| 49 | } |
| 50 | } |
| 51 | |
| 52 | // If there are multiple threads, the queue can empty while we're waiting (another |
| 53 | // thread may service the task we were waiting on). |
| 54 | if (shouldStop() || taskQueue.empty()) |
| 55 | continue; |
| 56 | |
| 57 | Function f = taskQueue.begin()->second; |
| 58 | taskQueue.erase(taskQueue.begin()); |
| 59 | |
| 60 | { |
| 61 | // Unlock before calling f, so it can reschedule itself or another task |
| 62 | // without deadlocking: |
| 63 | REVERSE_LOCK(lock); |
| 64 | f(); |
| 65 | } |
| 66 | } catch (...) { |
| 67 | --nThreadsServicingQueue; |
| 68 | throw; |
| 69 | } |
| 70 | } |
| 71 | --nThreadsServicingQueue; |
| 72 | newTaskScheduled.notify_one(); |
| 73 | } |
| 74 | |
| 75 | void CScheduler::schedule(CScheduler::Function f, std::chrono::system_clock::time_point t) |
| 76 | { |