partial Merge #18234: refactor: Replace boost::mutex,condition_var,chrono with std equivalents in scheduler

This backport does not include changes that depend on bitcoin pr 18037

70a6b529f306ff72ea1badf25e970a92b2b17ab3 lint-cppcheck: Remove -DHAVE_WORKING_BOOST_SLEEP_FOR (Anthony Towns)
294937b39de5924e772f8ed90d35c53290c8acab scheduler_tests: re-enable mockforward test (Anthony Towns)
cea19f685915be8affb2203184a549576194413f Drop unused reverselock.h (Anthony Towns)
d0ebd93270758ea97ea956b8821e17a2d001ea94 scheduler: switch from boost to std (Anthony Towns)
b9c426012770d166e6ebfab27689be44e6e89aa5 sync.h: add REVERSE_LOCK (Anthony Towns)
306f71b4eb4a0fd8e64f47dc008bc235b80b13d9 scheduler: don't rely on boost interrupt on shutdown (Anthony Towns)

Pull request description:

  Replacing boost functionality with C++11 stuff.

  Motivated by #18227, but should stand alone. Changing from `boost::condition_var` to `std::condition_var` means `threadGroup.interrupt_all` isn't enough to interrupt `serviceQueue` anymore, so that means calling `stop()` before `join_all()` is needed. And the existing reverselock.h code doesn't work with sync.h's DebugLock code (because the reversed lock won't be removed from `g_lockstack` which then leads to incorrect potential deadlock warnings), so I've replaced that with a dedicated class and macro that's aware of our debug lock behaviour.

  Fixes #16027, Fixes #14200, Fixes #18227

ACKs for top commit:
  laanwj:
    ACK 70a6b529f306ff72ea1badf25e970a92b2b17ab3

Tree-SHA512: d1da13adeabcf9186d114e2dad9a4fdbe2e440f7afbccde0c13dfbaf464efcd850b69d3371c5bf8b179d7ceb9d81f4af3cc22960b90834e41eaaf6d52ef7d331

# Conflicts:
#	src/reverselock.h
#	src/rpc/misc.cpp
#	src/scheduler.cpp
#	src/scheduler.h
#	src/sync.cpp
#	src/sync.h
#	src/test/reverselock_tests.cpp
#	src/test/scheduler_tests.cpp
#	src/test/test_dash.cpp
#	test/lint/extended-lint-cppcheck.sh
This commit is contained in:
Wladimir J. van der Laan 2020-03-06 20:47:49 +01:00 committed by Pasta
parent 97709a2bfa
commit 9b8884b430
No known key found for this signature in database
GPG Key ID: 52527BEDABE87984
13 changed files with 141 additions and 104 deletions

View File

@ -233,7 +233,6 @@ BITCOIN_CORE_H = \
psbt.h \
random.h \
reverse_iterator.h \
reverselock.h \
rpc/blockchain.h \
rpc/client.h \
rpc/mining.h \

View File

@ -268,6 +268,7 @@ void PrepareShutdown()
// After everything has been shut down, but before things get flushed, stop the
// CScheduler/checkqueue threadGroup
scheduler.stop();
threadGroup.interrupt_all();
threadGroup.join_all();
StopScriptCheckWorkerThreads();

View File

@ -43,7 +43,7 @@ CChainLocksHandler::CChainLocksHandler()
CChainLocksHandler::~CChainLocksHandler()
{
scheduler_thread->interrupt();
scheduler->stop();
scheduler_thread->join();
delete scheduler_thread;
delete scheduler;

View File

@ -1,34 +0,0 @@
// Copyright (c) 2015 The Bitcoin Core developers
// Distributed under the MIT software license, see the accompanying
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
#ifndef BITCOIN_REVERSELOCK_H
#define BITCOIN_REVERSELOCK_H
/**
* An RAII-style reverse lock. Unlocks on construction and locks on destruction.
*/
template<typename Lock>
class reverse_lock
{
public:
explicit reverse_lock(Lock& _lock) : lock(_lock) {
_lock.unlock();
_lock.swap(templock);
}
~reverse_lock() {
templock.lock();
templock.swap(lock);
}
private:
reverse_lock(reverse_lock const&);
reverse_lock& operator=(reverse_lock const&);
Lock& lock;
Lock templock;
};
#endif // BITCOIN_REVERSELOCK_H

View File

@ -5,7 +5,6 @@
#include <scheduler.h>
#include <random.h>
#include <reverselock.h>
#include <assert.h>
#include <utility>
@ -20,18 +19,9 @@ CScheduler::~CScheduler()
}
#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 +30,7 @@ void CScheduler::serviceQueue()
while (!shouldStop()) {
try {
if (!shouldStop() && taskQueue.empty()) {
reverse_lock<boost::unique_lock<boost::mutex> > rlock(lock);
REVERSE_LOCK(lock);
// Use this chance to get more entropy
RandAddSeedSleep();
}
@ -52,21 +42,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())
@ -78,7 +60,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 (...) {
@ -93,7 +75,7 @@ void CScheduler::serviceQueue()
void CScheduler::stop(bool drain)
{
{
boost::unique_lock<boost::mutex> lock(newTaskMutex);
LOCK(newTaskMutex);
if (drain)
stopWhenEmpty = true;
else
@ -102,10 +84,10 @@ 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();
@ -113,7 +95,7 @@ void CScheduler::schedule(CScheduler::Function f, boost::chrono::system_clock::t
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)
@ -127,10 +109,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;
@ -140,7 +122,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;
}
@ -154,7 +136,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() {

View File

@ -7,11 +7,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>
@ -27,8 +28,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.
@ -43,7 +44,7 @@ public:
typedef std::function<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);
@ -68,20 +69,20 @@ public:
// 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()); }
};
/**

View File

@ -15,6 +15,7 @@
#include <stdio.h>
#include <system_error>
#include <map>
#include <memory>
#include <set>
@ -61,6 +62,11 @@ struct CLockLocation {
mutexName, sourceFile, itostr(sourceLine), (fTry ? " (TRY)" : ""), m_thread_name);
}
std::string Name() const
{
return mutexName;
}
private:
bool fTry;
std::string mutexName;
@ -161,6 +167,18 @@ void EnterCritical(const char* pszName, const char* pszFile, int nLine, void* cs
push_lock(cs, CLockLocation(pszName, pszFile, nLine, fTry, util::ThreadGetInternalName()));
}
void CheckLastCritical(void* cs, std::string& lockname, const char* guardname, const char* file, int line)
{
if (!g_lockstack.empty()) {
const auto& lastlock = g_lockstack.back();
if (lastlock.first == cs) {
lockname = lastlock.second.Name();
return;
}
}
throw std::system_error(EPERM, std::generic_category(), strprintf("%s:%s %s was not most recent critical section locked", file, line, guardname));
}
void LeaveCritical()
{
pop_lock();

View File

@ -50,6 +50,7 @@ LEAVE_CRITICAL_SECTION(mutex); // no RAII
#ifdef DEBUG_LOCKORDER
void EnterCritical(const char* pszName, const char* pszFile, int nLine, void* cs, bool fTry = false);
void LeaveCritical();
void CheckLastCritical(void* cs, std::string& lockname, const char* guardname, const char* file, int line);
std::string LocksHeld();
template <typename MutexType>
void AssertLockHeldInternal(const char* pszName, const char* pszFile, int nLine, MutexType* cs) ASSERT_EXCLUSIVE_LOCK(cs);
@ -65,6 +66,7 @@ extern bool g_debug_lockorder_abort;
#else
void static inline EnterCritical(const char* pszName, const char* pszFile, int nLine, void* cs, bool fTry = false) {}
void static inline LeaveCritical() {}
void static inline CheckLastCritical(void* cs, std::string& lockname, const char* guardname, const char* file, int line) {}
template <typename MutexType>
void static inline AssertLockHeldInternal(const char* pszName, const char* pszFile, int nLine, MutexType* cs) ASSERT_EXCLUSIVE_LOCK(cs) {}
void static inline AssertLockNotHeldInternal(const char* pszName, const char* pszFile, int nLine, void* cs) {}
@ -180,8 +182,45 @@ public:
{
return Base::owns_lock();
}
protected:
// needed for reverse_lock
UniqueLock() { }
public:
/**
* An RAII-style reverse lock. Unlocks on construction and locks on destruction.
*/
class reverse_lock {
public:
explicit reverse_lock(UniqueLock& _lock, const char* _guardname, const char* _file, int _line) : lock(_lock), file(_file), line(_line) {
CheckLastCritical((void*)lock.mutex(), lockname, _guardname, _file, _line);
lock.unlock();
LeaveCritical();
lock.swap(templock);
}
~reverse_lock() {
templock.swap(lock);
EnterCritical(lockname.c_str(), file.c_str(), line, (void*)lock.mutex());
lock.lock();
}
private:
reverse_lock(reverse_lock const&);
reverse_lock& operator=(reverse_lock const&);
UniqueLock& lock;
UniqueLock templock;
std::string lockname;
const std::string file;
const int line;
};
friend class reverse_lock;
};
#define REVERSE_LOCK(g) decltype(g)::reverse_lock PASTE2(revlock, __COUNTER__)(g, #g, __FILE__, __LINE__)
template<typename MutexArg>
using DebugLock = UniqueLock<typename std::remove_reference<typename std::remove_pointer<MutexArg>::type>::type>;

View File

@ -2,7 +2,7 @@
// Distributed under the MIT software license, see the accompanying
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
#include <reverselock.h>
#include <sync.h>
#include <test/test_dash.h>
#include <boost/test/unit_test.hpp>
@ -11,21 +11,50 @@ BOOST_FIXTURE_TEST_SUITE(reverselock_tests, BasicTestingSetup)
BOOST_AUTO_TEST_CASE(reverselock_basics)
{
boost::mutex mutex;
boost::unique_lock<boost::mutex> lock(mutex);
Mutex mutex;
WAIT_LOCK(mutex, lock);
BOOST_CHECK(lock.owns_lock());
{
reverse_lock<boost::unique_lock<boost::mutex> > rlock(lock);
REVERSE_LOCK(lock);
BOOST_CHECK(!lock.owns_lock());
}
BOOST_CHECK(lock.owns_lock());
}
BOOST_AUTO_TEST_CASE(reverselock_multiple)
{
Mutex mutex2;
Mutex mutex;
WAIT_LOCK(mutex2, lock2);
WAIT_LOCK(mutex, lock);
// Make sure undoing two locks succeeds
{
REVERSE_LOCK(lock);
BOOST_CHECK(!lock.owns_lock());
REVERSE_LOCK(lock2);
BOOST_CHECK(!lock2.owns_lock());
}
BOOST_CHECK(lock.owns_lock());
BOOST_CHECK(lock2.owns_lock());
}
BOOST_AUTO_TEST_CASE(reverselock_errors)
{
boost::mutex mutex;
boost::unique_lock<boost::mutex> lock(mutex);
Mutex mutex2;
Mutex mutex;
WAIT_LOCK(mutex2, lock2);
WAIT_LOCK(mutex, lock);
#ifdef DEBUG_LOCKORDER
// Make sure trying to reverse lock a previous lock fails
try {
REVERSE_LOCK(lock2);
BOOST_CHECK(false); // REVERSE_LOCK(lock2) succeeded
} catch(...) { }
BOOST_CHECK(lock2.owns_lock());
#endif
// Make sure trying to reverse lock an unlocked lock fails
lock.unlock();
@ -34,7 +63,7 @@ BOOST_AUTO_TEST_CASE(reverselock_errors)
bool failed = false;
try {
reverse_lock<boost::unique_lock<boost::mutex> > rlock(lock);
REVERSE_LOCK(lock);
} catch(...) {
failed = true;
}
@ -49,7 +78,7 @@ BOOST_AUTO_TEST_CASE(reverselock_errors)
lock.lock();
BOOST_CHECK(lock.owns_lock());
{
reverse_lock<boost::unique_lock<boost::mutex> > rlock(lock);
REVERSE_LOCK(lock);
BOOST_CHECK(!lock.owns_lock());
}

View File

@ -13,13 +13,13 @@
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);
@ -47,15 +47,15 @@ BOOST_AUTO_TEST_CASE(manythreads)
auto randomMsec = [](FastRandomContext& rc) -> int { return -11 + (int)rc.randrange(1012); }; // [-11, 1000]
auto randomDelta = [](FastRandomContext& rc) -> int { return -1000 + (int)rc.randrange(2001); }; // [-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]),
@ -73,14 +73,14 @@ BOOST_AUTO_TEST_CASE(manythreads)
microThreads.create_thread(std::bind(&CScheduler::serviceQueue, &microTasks));
UninterruptibleSleep(std::chrono::microseconds{600});
now = boost::chrono::system_clock::now();
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]),

View File

@ -116,6 +116,7 @@ TestingSetup::TestingSetup(const std::string& chainName) : BasicTestingSetup(cha
TestingSetup::~TestingSetup()
{
scheduler.stop();
llmq::InterruptLLMQSystem();
llmq::StopLLMQSystem();
g_txindex->Interrupt();

View File

@ -72,6 +72,8 @@ BOOST_FIXTURE_TEST_CASE(txindex_initial_sync, TestChain100Setup)
// shutdown sequence (c.f. Shutdown() in init.cpp)
txindex.Stop();
// txindex job may be scheduled, so stop scheduler before destructing
scheduler.stop();
threadGroup.interrupt_all();
threadGroup.join_all();

View File

@ -53,7 +53,6 @@ EXPECTED_BOOST_INCLUDES=(
boost/algorithm/string/classification.hpp
boost/algorithm/string/replace.hpp
boost/algorithm/string/split.hpp
boost/chrono/chrono.hpp
boost/date_time/posix_time/posix_time.hpp
boost/filesystem.hpp
boost/filesystem/fstream.hpp