Skip to content

Commit d0ebd93

Browse files
ajtownsajtowns
committed
scheduler: switch from boost to std
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.
1 parent b9c4260 commit d0ebd93

File tree

5 files changed

+52
-70
lines changed

5 files changed

+52
-70
lines changed

‎src/rpc/misc.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -393,7 +393,7 @@ static UniValue mockscheduler(const JSONRPCRequest& request)
393393
// protect against null pointer dereference
394394
CHECK_NONFATAL(g_rpc_node);
395395
CHECK_NONFATAL(g_rpc_node->scheduler);
396-
g_rpc_node->scheduler->MockForward(boost::chrono::seconds(delta_seconds));
396+
g_rpc_node->scheduler->MockForward(std::chrono::seconds(delta_seconds));
397397

398398
return NullUniValue;
399399
}

‎src/scheduler.cpp‎

Lines changed: 20 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@
55
#include <scheduler.h>
66

77
#include <random.h>
8-
#include <reverselock.h>
98

109
#include <assert.h>
1110
#include <utility>
@@ -20,18 +19,9 @@ CScheduler::~CScheduler()
2019
}
2120

2221

23-
#if BOOST_VERSION < 105000
24-
static boost::system_time toPosixTime(const boost::chrono::system_clock::time_point& t)
25-
{
26-
// Creating the posix_time using from_time_t loses sub-second precision. So rather than exporting the time_point to time_t,
27-
// start with a posix_time at the epoch (0) and add the milliseconds that have passed since then.
28-
return boost::posix_time::from_time_t(0) + boost::posix_time::milliseconds(boost::chrono::duration_cast<boost::chrono::milliseconds>(t.time_since_epoch()).count());
29-
}
30-
#endif
31-
3222
void CScheduler::serviceQueue()
3323
{
34-
boost::unique_lock<boost::mutex> lock(newTaskMutex);
24+
WAIT_LOCK(newTaskMutex, lock);
3525
++nThreadsServicingQueue;
3626

3727
// newTaskMutex is locked throughout this loop EXCEPT
@@ -40,7 +30,7 @@ void CScheduler::serviceQueue()
4030
while (!shouldStop()) {
4131
try {
4232
if (!shouldStop() && taskQueue.empty()) {
43-
reverse_lock<boost::unique_lock<boost::mutex> > rlock(lock);
33+
REVERSE_LOCK(lock);
4434
}
4535
while (!shouldStop() && taskQueue.empty()) {
4636
// Wait until there is something to do.
@@ -50,21 +40,13 @@ void CScheduler::serviceQueue()
5040
// Wait until either there is a new task, or until
5141
// the time of the first item on the queue:
5242

53-
// wait_until needs boost 1.50 or later; older versions have timed_wait:
54-
#if BOOST_VERSION < 105000
55-
while (!shouldStop() && !taskQueue.empty() &&
56-
newTaskScheduled.timed_wait(lock, toPosixTime(taskQueue.begin()->first))) {
57-
// Keep waiting until timeout
58-
}
59-
#else
60-
// Some boost versions have a conflicting overload of wait_until that returns void.
61-
// Explicitly use a template here to avoid hitting that overload.
6243
while (!shouldStop() && !taskQueue.empty()) {
63-
boost::chrono::system_clock::time_point timeToWaitFor = taskQueue.begin()->first;
64-
if (newTaskScheduled.wait_until<>(lock, timeToWaitFor) == boost::cv_status::timeout)
44+
std::chrono::system_clock::time_point timeToWaitFor = taskQueue.begin()->first;
45+
if (newTaskScheduled.wait_until(lock, timeToWaitFor) == std::cv_status::timeout) {
6546
break; // Exit loop after timeout, it means we reached the time of the event
47+
}
6648
}
67-
#endif
49+
6850
// If there are multiple threads, the queue can empty while we're waiting (another
6951
// thread may service the task we were waiting on).
7052
if (shouldStop() || taskQueue.empty())
@@ -76,7 +58,7 @@ void CScheduler::serviceQueue()
7658
{
7759
// Unlock before calling f, so it can reschedule itself or another task
7860
// without deadlocking:
79-
reverse_lock<boost::unique_lock<boost::mutex> > rlock(lock);
61+
REVERSE_LOCK(lock);
8062
f();
8163
}
8264
} catch (...) {
@@ -91,7 +73,7 @@ void CScheduler::serviceQueue()
9173
void CScheduler::stop(bool drain)
9274
{
9375
{
94-
boost::unique_lock<boost::mutex> lock(newTaskMutex);
76+
LOCK(newTaskMutex);
9577
if (drain)
9678
stopWhenEmpty = true;
9779
else
@@ -100,29 +82,29 @@ void CScheduler::stop(bool drain)
10082
newTaskScheduled.notify_all();
10183
}
10284

103-
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)
10486
{
10587
{
106-
boost::unique_lock<boost::mutex> lock(newTaskMutex);
88+
LOCK(newTaskMutex);
10789
taskQueue.insert(std::make_pair(t, f));
10890
}
10991
newTaskScheduled.notify_one();
11092
}
11193

11294
void CScheduler::scheduleFromNow(CScheduler::Function f, int64_t deltaMilliSeconds)
11395
{
114-
schedule(f, boost::chrono::system_clock::now() + boost::chrono::milliseconds(deltaMilliSeconds));
96+
schedule(f, std::chrono::system_clock::now() + std::chrono::milliseconds(deltaMilliSeconds));
11597
}
11698

117-
void CScheduler::MockForward(boost::chrono::seconds delta_seconds)
99+
void CScheduler::MockForward(std::chrono::seconds delta_seconds)
118100
{
119-
assert(delta_seconds.count() > 0 && delta_seconds < boost::chrono::hours{1});
101+
assert(delta_seconds.count() > 0 && delta_seconds < std::chrono::hours{1});
120102

121103
{
122-
boost::unique_lock<boost::mutex> lock(newTaskMutex);
104+
LOCK(newTaskMutex);
123105

124106
// use temp_queue to maintain updated schedule
125-
std::multimap<boost::chrono::system_clock::time_point, Function> temp_queue;
107+
std::multimap<std::chrono::system_clock::time_point, Function> temp_queue;
126108

127109
for (const auto& element : taskQueue) {
128110
temp_queue.emplace_hint(temp_queue.cend(), element.first - delta_seconds, element.second);
@@ -147,10 +129,10 @@ void CScheduler::scheduleEvery(CScheduler::Function f, int64_t deltaMilliSeconds
147129
scheduleFromNow(std::bind(&Repeat, this, f, deltaMilliSeconds), deltaMilliSeconds);
148130
}
149131

150-
size_t CScheduler::getQueueInfo(boost::chrono::system_clock::time_point &first,
151-
boost::chrono::system_clock::time_point &last) const
132+
size_t CScheduler::getQueueInfo(std::chrono::system_clock::time_point &first,
133+
std::chrono::system_clock::time_point &last) const
152134
{
153-
boost::unique_lock<boost::mutex> lock(newTaskMutex);
135+
LOCK(newTaskMutex);
154136
size_t result = taskQueue.size();
155137
if (!taskQueue.empty()) {
156138
first = taskQueue.begin()->first;
@@ -160,7 +142,7 @@ size_t CScheduler::getQueueInfo(boost::chrono::system_clock::time_point &first,
160142
}
161143

162144
bool CScheduler::AreThreadsServicingQueue() const {
163-
boost::unique_lock<boost::mutex> lock(newTaskMutex);
145+
LOCK(newTaskMutex);
164146
return nThreadsServicingQueue;
165147
}
166148

@@ -174,7 +156,7 @@ void SingleThreadedSchedulerClient::MaybeScheduleProcessQueue() {
174156
if (m_are_callbacks_running) return;
175157
if (m_callbacks_pending.empty()) return;
176158
}
177-
m_pscheduler->schedule(std::bind(&SingleThreadedSchedulerClient::ProcessQueue, this));
159+
m_pscheduler->schedule(std::bind(&SingleThreadedSchedulerClient::ProcessQueue, this), std::chrono::system_clock::now());
178160
}
179161

180162
void SingleThreadedSchedulerClient::ProcessQueue() {

‎src/scheduler.h‎

Lines changed: 17 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -7,11 +7,12 @@
77

88
//
99
// NOTE:
10-
// boost::thread / boost::chrono should be ported to std::thread / std::chrono
10+
// boost::thread should be ported to std::thread
1111
// when we support C++11.
1212
//
13-
#include <boost/chrono/chrono.hpp>
14-
#include <boost/thread.hpp>
13+
#include <condition_variable>
14+
#include <functional>
15+
#include <list>
1516
#include <map>
1617

1718
#include <sync.h>
@@ -27,8 +28,8 @@
2728
// s->scheduleFromNow(std::bind(Class::func, this, argument), 3);
2829
// boost::thread* t = new boost::thread(std::bind(CScheduler::serviceQueue, s));
2930
//
30-
// ... then at program shutdown, clean up the thread running serviceQueue:
31-
// t->interrupt();
31+
// ... then at program shutdown, make sure to call stop() to clean up the thread(s) running serviceQueue:
32+
// s->stop();
3233
// t->join();
3334
// delete t;
3435
// delete s; // Must be done after thread is interrupted/joined.
@@ -43,7 +44,7 @@ class CScheduler
4344
typedef std::function<void()> Function;
4445

4546
// Call func at/after time t
46-
void schedule(Function f, boost::chrono::system_clock::time_point t=boost::chrono::system_clock::now());
47+
void schedule(Function f, std::chrono::system_clock::time_point t);
4748

4849
// Convenience method: call f once deltaMilliSeconds from now
4950
void scheduleFromNow(Function f, int64_t deltaMilliSeconds);
@@ -60,7 +61,7 @@ class CScheduler
6061
* Iterates through items on taskQueue and reschedules them
6162
* to be delta_seconds sooner.
6263
*/
63-
void MockForward(boost::chrono::seconds delta_seconds);
64+
void MockForward(std::chrono::seconds delta_seconds);
6465

6566
// To keep things as simple as possible, there is no unschedule.
6667

@@ -75,20 +76,20 @@ class CScheduler
7576

7677
// Returns number of tasks waiting to be serviced,
7778
// and first and last task times
78-
size_t getQueueInfo(boost::chrono::system_clock::time_point &first,
79-
boost::chrono::system_clock::time_point &last) const;
79+
size_t getQueueInfo(std::chrono::system_clock::time_point &first,
80+
std::chrono::system_clock::time_point &last) const;
8081

8182
// Returns true if there are threads actively running in serviceQueue()
8283
bool AreThreadsServicingQueue() const;
8384

8485
private:
85-
std::multimap<boost::chrono::system_clock::time_point, Function> taskQueue;
86-
boost::condition_variable newTaskScheduled;
87-
mutable boost::mutex newTaskMutex;
88-
int nThreadsServicingQueue;
89-
bool stopRequested;
90-
bool stopWhenEmpty;
91-
bool shouldStop() const { return stopRequested || (stopWhenEmpty && taskQueue.empty()); }
86+
mutable Mutex newTaskMutex;
87+
std::condition_variable newTaskScheduled;
88+
std::multimap<std::chrono::system_clock::time_point, Function> taskQueue GUARDED_BY(newTaskMutex);
89+
int nThreadsServicingQueue GUARDED_BY(newTaskMutex);
90+
bool stopRequested GUARDED_BY(newTaskMutex);
91+
bool stopWhenEmpty GUARDED_BY(newTaskMutex);
92+
bool shouldStop() const EXCLUSIVE_LOCKS_REQUIRED(newTaskMutex) { return stopRequested || (stopWhenEmpty && taskQueue.empty()); }
9293
};
9394

9495
/**

‎src/test/scheduler_tests.cpp‎

Lines changed: 14 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -11,13 +11,13 @@
1111

1212
BOOST_AUTO_TEST_SUITE(scheduler_tests)
1313

14-
static void microTask(CScheduler& s, boost::mutex& mutex, int& counter, int delta, boost::chrono::system_clock::time_point rescheduleTime)
14+
static void microTask(CScheduler& s, boost::mutex& mutex, int& counter, int delta, std::chrono::system_clock::time_point rescheduleTime)
1515
{
1616
{
1717
boost::unique_lock<boost::mutex> lock(mutex);
1818
counter += delta;
1919
}
20-
boost::chrono::system_clock::time_point noTime = boost::chrono::system_clock::time_point::min();
20+
std::chrono::system_clock::time_point noTime = std::chrono::system_clock::time_point::min();
2121
if (rescheduleTime != noTime) {
2222
CScheduler::Function f = std::bind(&microTask, std::ref(s), std::ref(mutex), std::ref(counter), -delta + 1, noTime);
2323
s.schedule(f, rescheduleTime);
@@ -45,15 +45,15 @@ BOOST_AUTO_TEST_CASE(manythreads)
4545
auto randomMsec = [](FastRandomContext& rc) -> int { return -11 + (int)rc.randrange(1012); }; // [-11, 1000]
4646
auto randomDelta = [](FastRandomContext& rc) -> int { return -1000 + (int)rc.randrange(2001); }; // [-1000, 1000]
4747

48-
boost::chrono::system_clock::time_point start = boost::chrono::system_clock::now();
49-
boost::chrono::system_clock::time_point now = start;
50-
boost::chrono::system_clock::time_point first, last;
48+
std::chrono::system_clock::time_point start = std::chrono::system_clock::now();
49+
std::chrono::system_clock::time_point now = start;
50+
std::chrono::system_clock::time_point first, last;
5151
size_t nTasks = microTasks.getQueueInfo(first, last);
5252
BOOST_CHECK(nTasks == 0);
5353

5454
for (int i = 0; i < 100; ++i) {
55-
boost::chrono::system_clock::time_point t = now + boost::chrono::microseconds(randomMsec(rng));
56-
boost::chrono::system_clock::time_point tReschedule = now + boost::chrono::microseconds(500 + randomMsec(rng));
55+
std::chrono::system_clock::time_point t = now + std::chrono::microseconds(randomMsec(rng));
56+
std::chrono::system_clock::time_point tReschedule = now + std::chrono::microseconds(500 + randomMsec(rng));
5757
int whichCounter = zeroToNine(rng);
5858
CScheduler::Function f = std::bind(&microTask, std::ref(microTasks),
5959
std::ref(counterMutex[whichCounter]), std::ref(counter[whichCounter]),
@@ -71,14 +71,14 @@ BOOST_AUTO_TEST_CASE(manythreads)
7171
microThreads.create_thread(std::bind(&CScheduler::serviceQueue, &microTasks));
7272

7373
UninterruptibleSleep(std::chrono::microseconds{600});
74-
now = boost::chrono::system_clock::now();
74+
now = std::chrono::system_clock::now();
7575

7676
// More threads and more tasks:
7777
for (int i = 0; i < 5; i++)
7878
microThreads.create_thread(std::bind(&CScheduler::serviceQueue, &microTasks));
7979
for (int i = 0; i < 100; i++) {
80-
boost::chrono::system_clock::time_point t = now + boost::chrono::microseconds(randomMsec(rng));
81-
boost::chrono::system_clock::time_point tReschedule = now + boost::chrono::microseconds(500 + randomMsec(rng));
80+
std::chrono::system_clock::time_point t = now + std::chrono::microseconds(randomMsec(rng));
81+
std::chrono::system_clock::time_point tReschedule = now + std::chrono::microseconds(500 + randomMsec(rng));
8282
int whichCounter = zeroToNine(rng);
8383
CScheduler::Function f = std::bind(&microTask, std::ref(microTasks),
8484
std::ref(counterMutex[whichCounter]), std::ref(counter[whichCounter]),
@@ -157,14 +157,14 @@ BOOST_AUTO_TEST_CASE(mockforward)
157157
scheduler.scheduleFromNow(dummy, 8*min_in_milli);
158158
159159
// check taskQueue
160-
boost::chrono::system_clock::time_point first, last;
160+
std::chrono::system_clock::time_point first, last;
161161
size_t num_tasks = scheduler.getQueueInfo(first, last);
162162
BOOST_CHECK_EQUAL(num_tasks, 3ul);
163163
164164
std::thread scheduler_thread([&]() { scheduler.serviceQueue(); });
165165
166166
// bump the scheduler forward 5 minutes
167-
scheduler.MockForward(boost::chrono::seconds(5*60));
167+
scheduler.MockForward(std::chrono::seconds(5*60));
168168
169169
// ensure scheduler has chance to process all tasks queued for before 1 ms from now.
170170
scheduler.scheduleFromNow([&scheduler]{ scheduler.stop(false); }, 1);
@@ -178,8 +178,8 @@ BOOST_AUTO_TEST_CASE(mockforward)
178178
BOOST_CHECK_EQUAL(counter, 2);
179179
180180
// check that the time of the remaining job has been updated
181-
boost::chrono::system_clock::time_point now = boost::chrono::system_clock::now();
182-
int delta = boost::chrono::duration_cast<boost::chrono::seconds>(first - now).count();
181+
std::chrono::system_clock::time_point now = std::chrono::system_clock::now();
182+
int delta = std::chrono::duration_cast<std::chrono::seconds>(first - now).count();
183183
// should be between 2 & 3 minutes from now
184184
BOOST_CHECK(delta > 2*60 && delta < 3*60);
185185
}

‎test/lint/lint-includes.sh‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,6 @@ EXPECTED_BOOST_INCLUDES=(
5353
boost/algorithm/string/classification.hpp
5454
boost/algorithm/string/replace.hpp
5555
boost/algorithm/string/split.hpp
56-
boost/chrono/chrono.hpp
5756
boost/date_time/posix_time/posix_time.hpp
5857
boost/filesystem.hpp
5958
boost/filesystem/fstream.hpp

0 commit comments

Comments
 (0)