diff options
Diffstat (limited to 'src/scheduler.cpp')
-rw-r--r-- | src/scheduler.cpp | 80 |
1 files changed, 40 insertions, 40 deletions
diff --git a/src/scheduler.cpp b/src/scheduler.cpp index 927a3f3820..4cac5a54e0 100644 --- a/src/scheduler.cpp +++ b/src/scheduler.cpp @@ -5,33 +5,24 @@ #include <scheduler.h> #include <random.h> -#include <reverselock.h> #include <assert.h> #include <utility> -CScheduler::CScheduler() : nThreadsServicingQueue(0), stopRequested(false), stopWhenEmpty(false) +CScheduler::CScheduler() { } CScheduler::~CScheduler() { assert(nThreadsServicingQueue == 0); + if (stopWhenEmpty) assert(taskQueue.empty()); } -#if BOOST_VERSION < 105000 -static boost::system_time toPosixTime(const boost::chrono::system_clock::time_point& t) -{ - // Creating the posix_time using from_time_t loses sub-second precision. So rather than exporting the time_point to time_t, - // start with a posix_time at the epoch (0) and add the milliseconds that have passed since then. - return boost::posix_time::from_time_t(0) + boost::posix_time::milliseconds(boost::chrono::duration_cast<boost::chrono::milliseconds>(t.time_since_epoch()).count()); -} -#endif - void CScheduler::serviceQueue() { - boost::unique_lock<boost::mutex> lock(newTaskMutex); + WAIT_LOCK(newTaskMutex, lock); ++nThreadsServicingQueue; // newTaskMutex is locked throughout this loop EXCEPT @@ -40,7 +31,7 @@ void CScheduler::serviceQueue() while (!shouldStop()) { try { if (!shouldStop() && taskQueue.empty()) { - reverse_lock<boost::unique_lock<boost::mutex> > rlock(lock); + REVERSE_LOCK(lock); } while (!shouldStop() && taskQueue.empty()) { // Wait until there is something to do. @@ -50,21 +41,13 @@ void CScheduler::serviceQueue() // Wait until either there is a new task, or until // the time of the first item on the queue: -// wait_until needs boost 1.50 or later; older versions have timed_wait: -#if BOOST_VERSION < 105000 - while (!shouldStop() && !taskQueue.empty() && - newTaskScheduled.timed_wait(lock, toPosixTime(taskQueue.begin()->first))) { - // Keep waiting until timeout - } -#else - // Some boost versions have a conflicting overload of wait_until that returns void. - // Explicitly use a template here to avoid hitting that overload. while (!shouldStop() && !taskQueue.empty()) { - boost::chrono::system_clock::time_point timeToWaitFor = taskQueue.begin()->first; - if (newTaskScheduled.wait_until<>(lock, timeToWaitFor) == boost::cv_status::timeout) + std::chrono::system_clock::time_point timeToWaitFor = taskQueue.begin()->first; + if (newTaskScheduled.wait_until(lock, timeToWaitFor) == std::cv_status::timeout) { break; // Exit loop after timeout, it means we reached the time of the event + } } -#endif + // If there are multiple threads, the queue can empty while we're waiting (another // thread may service the task we were waiting on). if (shouldStop() || taskQueue.empty()) @@ -76,7 +59,7 @@ void CScheduler::serviceQueue() { // Unlock before calling f, so it can reschedule itself or another task // without deadlocking: - reverse_lock<boost::unique_lock<boost::mutex> > rlock(lock); + REVERSE_LOCK(lock); f(); } } catch (...) { @@ -91,7 +74,7 @@ void CScheduler::serviceQueue() void CScheduler::stop(bool drain) { { - boost::unique_lock<boost::mutex> lock(newTaskMutex); + LOCK(newTaskMutex); if (drain) stopWhenEmpty = true; else @@ -100,35 +83,52 @@ void CScheduler::stop(bool drain) newTaskScheduled.notify_all(); } -void CScheduler::schedule(CScheduler::Function f, boost::chrono::system_clock::time_point t) +void CScheduler::schedule(CScheduler::Function f, std::chrono::system_clock::time_point t) { { - boost::unique_lock<boost::mutex> lock(newTaskMutex); + LOCK(newTaskMutex); taskQueue.insert(std::make_pair(t, f)); } newTaskScheduled.notify_one(); } -void CScheduler::scheduleFromNow(CScheduler::Function f, int64_t deltaMilliSeconds) +void CScheduler::MockForward(std::chrono::seconds delta_seconds) { - schedule(f, boost::chrono::system_clock::now() + boost::chrono::milliseconds(deltaMilliSeconds)); + assert(delta_seconds.count() > 0 && delta_seconds < std::chrono::hours{1}); + + { + LOCK(newTaskMutex); + + // use temp_queue to maintain updated schedule + std::multimap<std::chrono::system_clock::time_point, Function> temp_queue; + + for (const auto& element : taskQueue) { + temp_queue.emplace_hint(temp_queue.cend(), element.first - delta_seconds, element.second); + } + + // point taskQueue to temp_queue + taskQueue = std::move(temp_queue); + } + + // notify that the taskQueue needs to be processed + newTaskScheduled.notify_one(); } -static void Repeat(CScheduler* s, CScheduler::Function f, int64_t deltaMilliSeconds) +static void Repeat(CScheduler& s, CScheduler::Function f, std::chrono::milliseconds delta) { f(); - s->scheduleFromNow(std::bind(&Repeat, s, f, deltaMilliSeconds), deltaMilliSeconds); + s.scheduleFromNow([=, &s] { Repeat(s, f, delta); }, delta); } -void CScheduler::scheduleEvery(CScheduler::Function f, int64_t deltaMilliSeconds) +void CScheduler::scheduleEvery(CScheduler::Function f, std::chrono::milliseconds delta) { - scheduleFromNow(std::bind(&Repeat, this, f, deltaMilliSeconds), deltaMilliSeconds); + scheduleFromNow([=] { Repeat(*this, f, delta); }, delta); } -size_t CScheduler::getQueueInfo(boost::chrono::system_clock::time_point &first, - boost::chrono::system_clock::time_point &last) const +size_t CScheduler::getQueueInfo(std::chrono::system_clock::time_point &first, + std::chrono::system_clock::time_point &last) const { - boost::unique_lock<boost::mutex> lock(newTaskMutex); + LOCK(newTaskMutex); size_t result = taskQueue.size(); if (!taskQueue.empty()) { first = taskQueue.begin()->first; @@ -138,7 +138,7 @@ size_t CScheduler::getQueueInfo(boost::chrono::system_clock::time_point &first, } bool CScheduler::AreThreadsServicingQueue() const { - boost::unique_lock<boost::mutex> lock(newTaskMutex); + LOCK(newTaskMutex); return nThreadsServicingQueue; } @@ -152,7 +152,7 @@ void SingleThreadedSchedulerClient::MaybeScheduleProcessQueue() { if (m_are_callbacks_running) return; if (m_callbacks_pending.empty()) return; } - m_pscheduler->schedule(std::bind(&SingleThreadedSchedulerClient::ProcessQueue, this)); + m_pscheduler->schedule(std::bind(&SingleThreadedSchedulerClient::ProcessQueue, this), std::chrono::system_clock::now()); } void SingleThreadedSchedulerClient::ProcessQueue() { |