dash/src/scheduler.cpp
Wladimir J. van der Laan 9b8884b430
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
2021-09-28 19:42:22 -04:00

195 lines
5.8 KiB
C++

// 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.
#include <scheduler.h>
#include <random.h>
#include <assert.h>
#include <utility>
CScheduler::CScheduler() : nThreadsServicingQueue(0), stopRequested(false), stopWhenEmpty(false)
{
}
CScheduler::~CScheduler()
{
assert(nThreadsServicingQueue == 0);
}
void CScheduler::serviceQueue()
{
WAIT_LOCK(newTaskMutex, lock);
++nThreadsServicingQueue;
// newTaskMutex is locked throughout this loop EXCEPT
// when the thread is waiting or when the user's function
// is called.
while (!shouldStop()) {
try {
if (!shouldStop() && taskQueue.empty()) {
REVERSE_LOCK(lock);
// Use this chance to get more entropy
RandAddSeedSleep();
}
while (!shouldStop() && taskQueue.empty()) {
// Wait until there is something to do.
newTaskScheduled.wait(lock);
}
// Wait until either there is a new task, or until
// the time of the first item on the queue:
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())
continue;
Function f = taskQueue.begin()->second;
taskQueue.erase(taskQueue.begin());
{
// Unlock before calling f, so it can reschedule itself or another task
// without deadlocking:
REVERSE_LOCK(lock);
f();
}
} catch (...) {
--nThreadsServicingQueue;
throw;
}
}
--nThreadsServicingQueue;
newTaskScheduled.notify_one();
}
void CScheduler::stop(bool drain)
{
{
LOCK(newTaskMutex);
if (drain)
stopWhenEmpty = true;
else
stopRequested = true;
}
newTaskScheduled.notify_all();
}
void CScheduler::schedule(CScheduler::Function f, std::chrono::system_clock::time_point t)
{
{
LOCK(newTaskMutex);
taskQueue.insert(std::make_pair(t, f));
}
newTaskScheduled.notify_one();
}
void CScheduler::scheduleFromNow(CScheduler::Function f, int64_t deltaMilliSeconds)
{
schedule(f, std::chrono::system_clock::now() + std::chrono::milliseconds(deltaMilliSeconds));
}
static void Repeat(CScheduler* s, CScheduler::Function f, int64_t deltaMilliSeconds)
{
f();
s->scheduleFromNow(std::bind(&Repeat, s, f, deltaMilliSeconds), deltaMilliSeconds);
}
void CScheduler::scheduleEvery(CScheduler::Function f, int64_t deltaMilliSeconds)
{
scheduleFromNow(std::bind(&Repeat, this, f, deltaMilliSeconds), deltaMilliSeconds);
}
size_t CScheduler::getQueueInfo(std::chrono::system_clock::time_point &first,
std::chrono::system_clock::time_point &last) const
{
LOCK(newTaskMutex);
size_t result = taskQueue.size();
if (!taskQueue.empty()) {
first = taskQueue.begin()->first;
last = taskQueue.rbegin()->first;
}
return result;
}
bool CScheduler::AreThreadsServicingQueue() const {
LOCK(newTaskMutex);
return nThreadsServicingQueue;
}
void SingleThreadedSchedulerClient::MaybeScheduleProcessQueue() {
{
LOCK(m_cs_callbacks_pending);
// Try to avoid scheduling too many copies here, but if we
// accidentally have two ProcessQueue's scheduled at once its
// not a big deal.
if (m_are_callbacks_running) return;
if (m_callbacks_pending.empty()) return;
}
m_pscheduler->schedule(std::bind(&SingleThreadedSchedulerClient::ProcessQueue, this), std::chrono::system_clock::now());
}
void SingleThreadedSchedulerClient::ProcessQueue() {
std::function<void ()> callback;
{
LOCK(m_cs_callbacks_pending);
if (m_are_callbacks_running) return;
if (m_callbacks_pending.empty()) return;
m_are_callbacks_running = true;
callback = std::move(m_callbacks_pending.front());
m_callbacks_pending.pop_front();
}
// RAII the setting of fCallbacksRunning and calling MaybeScheduleProcessQueue
// to ensure both happen safely even if callback() throws.
struct RAIICallbacksRunning {
SingleThreadedSchedulerClient* instance;
explicit RAIICallbacksRunning(SingleThreadedSchedulerClient* _instance) : instance(_instance) {}
~RAIICallbacksRunning() {
{
LOCK(instance->m_cs_callbacks_pending);
instance->m_are_callbacks_running = false;
}
instance->MaybeScheduleProcessQueue();
}
} raiicallbacksrunning(this);
callback();
}
void SingleThreadedSchedulerClient::AddToProcessQueue(std::function<void ()> func) {
assert(m_pscheduler);
{
LOCK(m_cs_callbacks_pending);
m_callbacks_pending.emplace_back(std::move(func));
}
MaybeScheduleProcessQueue();
}
void SingleThreadedSchedulerClient::EmptyQueue() {
assert(!m_pscheduler->AreThreadsServicingQueue());
bool should_continue = true;
while (should_continue) {
ProcessQueue();
LOCK(m_cs_callbacks_pending);
should_continue = !m_callbacks_pending.empty();
}
}
size_t SingleThreadedSchedulerClient::CallbacksPending() {
LOCK(m_cs_callbacks_pending);
return m_callbacks_pending.size();
}