scheduler: switch from boost to std

ajtowns · random-zebra · commit 55ac6de5cc7f · 2021-08-18T17:49:50.000+02:00
Changes from boost::chrono to std::chrono, boost::condition_var to
std::condition_var, boost::mutex to sync.h Mutex, and reverselock.h to
sync.h REVERSE_LOCK. Also adds threadsafety annotations to CScheduler
members.
diff --git a/src/scheduler.cpp b/src/scheduler.cpp
@@ -6,7 +6,6 @@
 #include "scheduler.h"
 
 #include "random.h"
-#include "reverselock.h"
 
 #include <assert.h>
 #include <utility>
@@ -22,7 +21,7 @@ CScheduler::~CScheduler()
 
 void CScheduler::serviceQueue()
 {
-    boost::unique_lock<boost::mutex> lock(newTaskMutex);
+    WAIT_LOCK(newTaskMutex, lock);
     ++nThreadsServicingQueue;
 
     // newTaskMutex is locked throughout this loop EXCEPT
@@ -31,7 +30,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.
@@ -41,12 +40,13 @@ void CScheduler::serviceQueue()
             // Wait until either there is a new task, or until
             // the time of the first item on the queue:
 
-            // 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() &&
-                   newTaskScheduled.wait_until<>(lock, taskQueue.begin()->first) != boost::cv_status::timeout) {
-                // Keep waiting until timeout
+            while (!shouldStop() && !taskQueue.empty()) {
+                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
+                }
             }
+
             // 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())
@@ -58,7 +58,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 (...) {
@@ -73,7 +73,7 @@ void CScheduler::serviceQueue()
 void CScheduler::stop(bool drain)
 {
     {
-        boost::unique_lock<boost::mutex> lock(newTaskMutex);
+        LOCK(newTaskMutex);
         if (drain)
             stopWhenEmpty = true;
         else
@@ -82,18 +82,18 @@ 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.emplace(t, f);
     }
     newTaskScheduled.notify_one();
 }
 
 void CScheduler::scheduleFromNow(CScheduler::Function f, int64_t deltaMilliSeconds)
 {
-    schedule(f, boost::chrono::system_clock::now() + boost::chrono::milliseconds(deltaMilliSeconds));
+    schedule(f, std::chrono::system_clock::now() + std::chrono::milliseconds(deltaMilliSeconds));
 }
 
 static void Repeat(CScheduler* s, CScheduler::Function f, int64_t deltaMilliSeconds)
@@ -107,10 +107,10 @@ void CScheduler::scheduleEvery(CScheduler::Function f, int64_t deltaMilliSeconds
     scheduleFromNow(std::bind(&Repeat, this, f, deltaMilliSeconds), deltaMilliSeconds);
 }
 
-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;
@@ -120,7 +120,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;
 }
 
@@ -133,7 +133,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() {
diff --git a/src/scheduler.h b/src/scheduler.h
@@ -8,11 +8,12 @@
 
 //
 // NOTE:
-// boost::thread / boost::chrono should be ported to std::thread / std::chrono
+// boost::thread should be ported to std::thread
 // when we support C++11.
 //
-#include <boost/chrono/chrono.hpp>
-#include <boost/thread.hpp>
+#include <condition_variable>
+#include <functional>
+#include <list>
 #include <map>
 
 #include "sync.h"
@@ -28,8 +29,8 @@
 // s->scheduleFromNow(std::bind(Class::func, this, argument), 3);
 // boost::thread* t = new boost::thread(std::bind(CScheduler::serviceQueue, s));
 //
-// ... then at program shutdown, clean up the thread running serviceQueue:
-// t->interrupt();
+// ... then at program shutdown, make sure to call stop() to clean up the thread(s) running serviceQueue:
+// s->stop();
 // t->join();
 // delete t;
 // delete s; // Must be done after thread is interrupted/joined.
@@ -44,7 +45,7 @@ class CScheduler
     typedef std::function<void(void)> Function;
 
     // Call func at/after time t
-    void schedule(Function f, boost::chrono::system_clock::time_point t=boost::chrono::system_clock::now());
+    void schedule(Function f, std::chrono::system_clock::time_point t);
 
     // Convenience method: call f once deltaMilliSeconds from now
     void scheduleFromNow(Function f, int64_t deltaMilliSeconds);
@@ -69,20 +70,20 @@ class CScheduler
 
     // Returns number of tasks waiting to be serviced,
     // and first and last task times
-    size_t getQueueInfo(boost::chrono::system_clock::time_point &first,
-                        boost::chrono::system_clock::time_point &last) const;
+    size_t getQueueInfo(std::chrono::system_clock::time_point &first,
+                        std::chrono::system_clock::time_point &last) const;
 
     // Returns true if there are threads actively running in serviceQueue()
     bool AreThreadsServicingQueue() const;
 
 private:
-    std::multimap<boost::chrono::system_clock::time_point, Function> taskQueue;
-    boost::condition_variable newTaskScheduled;
-    mutable boost::mutex newTaskMutex;
-    int nThreadsServicingQueue;
-    bool stopRequested;
-    bool stopWhenEmpty;
-    bool shouldStop() const { return stopRequested || (stopWhenEmpty && taskQueue.empty()); }
+    mutable Mutex newTaskMutex;
+    std::condition_variable newTaskScheduled;
+    std::multimap<std::chrono::system_clock::time_point, Function> taskQueue GUARDED_BY(newTaskMutex);
+    int nThreadsServicingQueue GUARDED_BY(newTaskMutex);
+    bool stopRequested GUARDED_BY(newTaskMutex);
+    bool stopWhenEmpty GUARDED_BY(newTaskMutex);
+    bool shouldStop() const EXCLUSIVE_LOCKS_REQUIRED(newTaskMutex) { return stopRequested || (stopWhenEmpty && taskQueue.empty()); }
 };
 
 /**
diff --git a/src/test/scheduler_tests.cpp b/src/test/scheduler_tests.cpp
@@ -5,6 +5,7 @@
 
 #include "random.h"
 #include "scheduler.h"
+#include "utiltime.h"
 #if defined(HAVE_CONFIG_H)
 #include "config/pivx-config.h"
 #endif
@@ -16,24 +17,19 @@
 
 BOOST_AUTO_TEST_SUITE(scheduler_tests)
 
-static void microTask(CScheduler& s, boost::mutex& mutex, int& counter, int delta, boost::chrono::system_clock::time_point rescheduleTime)
+static void microTask(CScheduler& s, boost::mutex& mutex, int& counter, int delta, std::chrono::system_clock::time_point rescheduleTime)
 {
     {
         boost::unique_lock<boost::mutex> lock(mutex);
         counter += delta;
     }
-    boost::chrono::system_clock::time_point noTime = boost::chrono::system_clock::time_point::min();
+    std::chrono::system_clock::time_point noTime = std::chrono::system_clock::time_point::min();
     if (rescheduleTime != noTime) {
         CScheduler::Function f = std::bind(&microTask, std::ref(s), std::ref(mutex), std::ref(counter), -delta + 1, noTime);
         s.schedule(f, rescheduleTime);
     }
 }
 
-static void MicroSleep(uint64_t n)
-{
-    boost::this_thread::sleep_for(boost::chrono::microseconds(n));
-}
-
 BOOST_AUTO_TEST_CASE(manythreads)
 {
     // Stress test: hundreds of microsecond-scheduled tasks,
@@ -55,15 +51,15 @@ BOOST_AUTO_TEST_CASE(manythreads)
     boost::random::uniform_int_distribution<> randomMsec(-11, 1000);
     boost::random::uniform_int_distribution<> randomDelta(-1000, 1000);
 
-    boost::chrono::system_clock::time_point start = boost::chrono::system_clock::now();
-    boost::chrono::system_clock::time_point now = start;
-    boost::chrono::system_clock::time_point first, last;
+    std::chrono::system_clock::time_point start = std::chrono::system_clock::now();
+    std::chrono::system_clock::time_point now = start;
+    std::chrono::system_clock::time_point first, last;
     size_t nTasks = microTasks.getQueueInfo(first, last);
     BOOST_CHECK(nTasks == 0);
 
     for (int i = 0; i < 100; ++i) {
-        boost::chrono::system_clock::time_point t = now + boost::chrono::microseconds(randomMsec(rng));
-        boost::chrono::system_clock::time_point tReschedule = now + boost::chrono::microseconds(500 + randomMsec(rng));
+        std::chrono::system_clock::time_point t = now + std::chrono::microseconds(randomMsec(rng));
+        std::chrono::system_clock::time_point tReschedule = now + std::chrono::microseconds(500 + randomMsec(rng));
         int whichCounter = zeroToNine(rng);
         CScheduler::Function f = std::bind(&microTask, std::ref(microTasks),
                                              std::ref(counterMutex[whichCounter]), std::ref(counter[whichCounter]),
@@ -80,15 +76,15 @@ BOOST_AUTO_TEST_CASE(manythreads)
     for (int i = 0; i < 5; i++)
         microThreads.create_thread(std::bind(&CScheduler::serviceQueue, &microTasks));
 
-    MicroSleep(600);
-    now = boost::chrono::system_clock::now();
+    UninterruptibleSleep(std::chrono::microseconds{600});
+    now = std::chrono::system_clock::now();
 
     // More threads and more tasks:
     for (int i = 0; i < 5; i++)
         microThreads.create_thread(std::bind(&CScheduler::serviceQueue, &microTasks));
     for (int i = 0; i < 100; i++) {
-        boost::chrono::system_clock::time_point t = now + boost::chrono::microseconds(randomMsec(rng));
-        boost::chrono::system_clock::time_point tReschedule = now + boost::chrono::microseconds(500 + randomMsec(rng));
+        std::chrono::system_clock::time_point t = now + std::chrono::microseconds(randomMsec(rng));
+        std::chrono::system_clock::time_point tReschedule = now + std::chrono::microseconds(500 + randomMsec(rng));
         int whichCounter = zeroToNine(rng);
         CScheduler::Function f = std::bind(&microTask, std::ref(microTasks),
                                              std::ref(counterMutex[whichCounter]), std::ref(counter[whichCounter]),

Original file line number	Diff line number	Diff line change
`@@ -6,7 +6,6 @@`
`6`	`6`	`#include "scheduler.h"`
`7`	`7`
`8`	`8`	`#include "random.h"`
`9`		`-#include "reverselock.h"`
`10`	`9`
`11`	`10`	`#include <assert.h>`
`12`	`11`	`#include <utility>`
`@@ -22,7 +21,7 @@ CScheduler::~CScheduler()`
`22`	`21`
`23`	`22`	`void CScheduler::serviceQueue()`
`24`	`23`	`{`
`25`		`- boost::unique_lock<boost::mutex> lock(newTaskMutex);`
	`24`	`+ WAIT_LOCK(newTaskMutex, lock);`
`26`	`25`	`++nThreadsServicingQueue;`
`27`	`26`
`28`	`27`	`// newTaskMutex is locked throughout this loop EXCEPT`
`@@ -31,7 +30,7 @@ void CScheduler::serviceQueue()`
`31`	`30`	`while (!shouldStop()) {`
`32`	`31`	`try {`
`33`	`32`	`if (!shouldStop() && taskQueue.empty()) {`
`34`		`- reverse_lock<boost::unique_lock<boost::mutex> > rlock(lock);`
	`33`	`+ REVERSE_LOCK(lock);`
`35`	`34`	`}`
`36`	`35`	`while (!shouldStop() && taskQueue.empty()) {`
`37`	`36`	`// Wait until there is something to do.`
`@@ -41,12 +40,13 @@ void CScheduler::serviceQueue()`
`41`	`40`	`// Wait until either there is a new task, or until`
`42`	`41`	`// the time of the first item on the queue:`
`43`	`42`
`44`		`- // Some boost versions have a conflicting overload of wait_until that returns void.`
`45`		`- // Explicitly use a template here to avoid hitting that overload.`
`46`		`- while (!shouldStop() && !taskQueue.empty() &&`
`47`		`- newTaskScheduled.wait_until<>(lock, taskQueue.begin()->first) != boost::cv_status::timeout) {`
`48`		`- // Keep waiting until timeout`
	`43`	`+ while (!shouldStop() && !taskQueue.empty()) {`
	`44`	`+ std::chrono::system_clock::time_point timeToWaitFor = taskQueue.begin()->first;`
	`45`	`+ if (newTaskScheduled.wait_until(lock, timeToWaitFor) == std::cv_status::timeout) {`
	`46`	`+ break; // Exit loop after timeout, it means we reached the time of the event`
	`47`	`+ }`
`49`	`48`	`}`
	`49`	`+`
`50`	`50`	`// If there are multiple threads, the queue can empty while we're waiting (another`
`51`	`51`	`// thread may service the task we were waiting on).`
`52`	`52`	`if (shouldStop() \|\| taskQueue.empty())`
`@@ -58,7 +58,7 @@ void CScheduler::serviceQueue()`
`58`	`58`	`{`
`59`	`59`	`// Unlock before calling f, so it can reschedule itself or another task`
`60`	`60`	`// without deadlocking:`
`61`		`- reverse_lock<boost::unique_lock<boost::mutex> > rlock(lock);`
	`61`	`+ REVERSE_LOCK(lock);`
`62`	`62`	`f();`
`63`	`63`	`}`
`64`	`64`	`} catch (...) {`
`@@ -73,7 +73,7 @@ void CScheduler::serviceQueue()`
`73`	`73`	`void CScheduler::stop(bool drain)`
`74`	`74`	`{`
`75`	`75`	`{`
`76`		`- boost::unique_lock<boost::mutex> lock(newTaskMutex);`
	`76`	`+ LOCK(newTaskMutex);`
`77`	`77`	`if (drain)`
`78`	`78`	`stopWhenEmpty = true;`
`79`	`79`	`else`
`@@ -82,18 +82,18 @@ void CScheduler::stop(bool drain)`
`82`	`82`	`newTaskScheduled.notify_all();`
`83`	`83`	`}`
`84`	`84`
`85`		`-void CScheduler::schedule(CScheduler::Function f, boost::chrono::system_clock::time_point t)`
	`85`	`+void CScheduler::schedule(CScheduler::Function f, std::chrono::system_clock::time_point t)`
`86`	`86`	`{`
`87`	`87`	`{`
`88`		`- boost::unique_lock<boost::mutex> lock(newTaskMutex);`
	`88`	`+ LOCK(newTaskMutex);`
`89`	`89`	`taskQueue.emplace(t, f);`
`90`	`90`	`}`
`91`	`91`	`newTaskScheduled.notify_one();`
`92`	`92`	`}`
`93`	`93`
`94`	`94`	`void CScheduler::scheduleFromNow(CScheduler::Function f, int64_t deltaMilliSeconds)`
`95`	`95`	`{`
`96`		`- schedule(f, boost::chrono::system_clock::now() + boost::chrono::milliseconds(deltaMilliSeconds));`
	`96`	`+ schedule(f, std::chrono::system_clock::now() + std::chrono::milliseconds(deltaMilliSeconds));`
`97`	`97`	`}`
`98`	`98`
`99`	`99`	`static void Repeat(CScheduler* s, CScheduler::Function f, int64_t deltaMilliSeconds)`
`@@ -107,10 +107,10 @@ void CScheduler::scheduleEvery(CScheduler::Function f, int64_t deltaMilliSeconds`
`107`	`107`	`scheduleFromNow(std::bind(&Repeat, this, f, deltaMilliSeconds), deltaMilliSeconds);`
`108`	`108`	`}`
`109`	`109`
`110`		`-size_t CScheduler::getQueueInfo(boost::chrono::system_clock::time_point &first,`
`111`		`- boost::chrono::system_clock::time_point &last) const`
	`110`	`+size_t CScheduler::getQueueInfo(std::chrono::system_clock::time_point &first,`
	`111`	`+ std::chrono::system_clock::time_point &last) const`
`112`	`112`	`{`
`113`		`- boost::unique_lock<boost::mutex> lock(newTaskMutex);`
	`113`	`+ LOCK(newTaskMutex);`
`114`	`114`	`size_t result = taskQueue.size();`
`115`	`115`	`if (!taskQueue.empty()) {`
`116`	`116`	`first = taskQueue.begin()->first;`
`@@ -120,7 +120,7 @@ size_t CScheduler::getQueueInfo(boost::chrono::system_clock::time_point &first,`
`120`	`120`	`}`
`121`	`121`
`122`	`122`	`bool CScheduler::AreThreadsServicingQueue() const {`
`123`		`- boost::unique_lock<boost::mutex> lock(newTaskMutex);`
	`123`	`+ LOCK(newTaskMutex);`
`124`	`124`	`return nThreadsServicingQueue;`
`125`	`125`	`}`
`126`	`126`
`@@ -133,7 +133,7 @@ void SingleThreadedSchedulerClient::MaybeScheduleProcessQueue() {`
`133`	`133`	`if (m_are_callbacks_running) return;`
`134`	`134`	`if (m_callbacks_pending.empty()) return;`
`135`	`135`	`}`
`136`		`- m_pscheduler->schedule(std::bind(&SingleThreadedSchedulerClient::ProcessQueue, this));`
	`136`	`+ m_pscheduler->schedule(std::bind(&SingleThreadedSchedulerClient::ProcessQueue, this), std::chrono::system_clock::now());`
`137`	`137`	`}`
`138`	`138`
`139`	`139`	`void SingleThreadedSchedulerClient::ProcessQueue() {`