Skip to content

Commit

Permalink
scheduler: Make schedule* methods type safe
Browse files Browse the repository at this point in the history
  • Loading branch information
MarcoFalke committed Mar 10, 2020
1 parent fa70ccc commit fadafb8
Show file tree
Hide file tree
Showing 7 changed files with 34 additions and 35 deletions.
8 changes: 4 additions & 4 deletions src/init.cpp
Expand Up @@ -86,8 +86,8 @@ static const bool DEFAULT_PROXYRANDOMIZE = true;
static const bool DEFAULT_REST_ENABLE = false;
static const bool DEFAULT_STOPAFTERBLOCKIMPORT = false;

// Dump addresses to banlist.dat every 15 minutes (900s)
static constexpr int DUMP_BANS_INTERVAL = 60 * 15;
// How often to dump addresses to banlist.dat
static constexpr std::chrono::minutes DUMP_BANS_INTERVAL{15};


#ifdef WIN32
Expand Down Expand Up @@ -1278,7 +1278,7 @@ bool AppInitMain(NodeContext& node)
// Gather some entropy once per minute.
node.scheduler->scheduleEvery([]{
RandAddPeriodic();
}, 60000);
}, std::chrono::minutes{1});

GetMainSignals().RegisterBackgroundSignalScheduler(*node.scheduler);

Expand Down Expand Up @@ -1863,7 +1863,7 @@ bool AppInitMain(NodeContext& node)
BanMan* banman = node.banman.get();
node.scheduler->scheduleEvery([banman]{
banman->DumpBanlist();
}, DUMP_BANS_INTERVAL * 1000);
}, DUMP_BANS_INTERVAL);

return true;
}
6 changes: 3 additions & 3 deletions src/net.cpp
Expand Up @@ -45,8 +45,8 @@ static_assert(MINIUPNPC_API_VERSION >= 10, "miniUPnPc API version >= 10 assumed"

#include <math.h>

// Dump addresses to peers.dat every 15 minutes (900s)
static constexpr int DUMP_PEERS_INTERVAL = 15 * 60;
// How often to dump addresses to peers.dat
static constexpr std::chrono::minutes DUMP_PEERS_INTERVAL{15};

/** Number of DNS seeds to query when the number of connections is low. */
static constexpr int DNSSEEDS_TO_QUERY_AT_ONCE = 3;
Expand Down Expand Up @@ -2343,7 +2343,7 @@ bool CConnman::Start(CScheduler& scheduler, const Options& connOptions)
threadMessageHandler = std::thread(&TraceThread<std::function<void()> >, "msghand", std::function<void()>(std::bind(&CConnman::ThreadMessageHandler, this)));

// Dump network addresses
scheduler.scheduleEvery(std::bind(&CConnman::DumpAddresses, this), DUMP_PEERS_INTERVAL * 1000);
scheduler.scheduleEvery([this] { DumpAddresses(); }, DUMP_PEERS_INTERVAL);

return true;
}
Expand Down
2 changes: 1 addition & 1 deletion src/net_processing.cpp
Expand Up @@ -1124,7 +1124,7 @@ PeerLogicValidation::PeerLogicValidation(CConnman* connmanIn, BanMan* banman, CS
// combine them in one function and schedule at the quicker (peer-eviction)
// timer.
static_assert(EXTRA_PEER_CHECK_INTERVAL < STALE_CHECK_INTERVAL, "peer eviction timer should be less than stale tip check timer");
scheduler.scheduleEvery(std::bind(&PeerLogicValidation::CheckForStaleTipAndEvictPeers, this, consensusParams), EXTRA_PEER_CHECK_INTERVAL * 1000);
scheduler.scheduleEvery([&] { this->CheckForStaleTipAndEvictPeers(consensusParams); }, std::chrono::seconds{EXTRA_PEER_CHECK_INTERVAL});
}

/**
Expand Down
13 changes: 4 additions & 9 deletions src/scheduler.cpp
Expand Up @@ -92,11 +92,6 @@ void CScheduler::schedule(CScheduler::Function f, std::chrono::system_clock::tim
newTaskScheduled.notify_one();
}

void CScheduler::scheduleFromNow(CScheduler::Function f, int64_t deltaMilliSeconds)
{
schedule(f, std::chrono::system_clock::now() + std::chrono::milliseconds(deltaMilliSeconds));
}

void CScheduler::MockForward(std::chrono::seconds delta_seconds)
{
assert(delta_seconds.count() > 0 && delta_seconds < std::chrono::hours{1});
Expand All @@ -119,15 +114,15 @@ void CScheduler::MockForward(std::chrono::seconds delta_seconds)
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(std::chrono::system_clock::time_point &first,
Expand Down
24 changes: 14 additions & 10 deletions src/scheduler.h
Expand Up @@ -24,8 +24,8 @@
// Usage:
//
// CScheduler* s = new CScheduler();
// s->scheduleFromNow(doSomething, 11); // Assuming a: void doSomething() { }
// s->scheduleFromNow(std::bind(Class::func, this, argument), 3);
// s->scheduleFromNow(doSomething, std::chrono::milliseconds{11}); // Assuming a: void doSomething() { }
// s->scheduleFromNow([=] { this->func(argument); }, std::chrono::milliseconds{3});
// boost::thread* t = new boost::thread(std::bind(CScheduler::serviceQueue, s));
//
// ... then at program shutdown, make sure to call stop() to clean up the thread(s) running serviceQueue:
Expand All @@ -46,15 +46,19 @@ class CScheduler
// Call func at/after time t
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);
/** Call f once after the delta has passed */
void scheduleFromNow(Function f, std::chrono::milliseconds delta)
{
schedule(std::move(f), std::chrono::system_clock::now() + delta);
}

// Another convenience method: call f approximately
// every deltaMilliSeconds forever, starting deltaMilliSeconds from now.
// To be more precise: every time f is finished, it
// is rescheduled to run deltaMilliSeconds later. If you
// need more accurate scheduling, don't use this method.
void scheduleEvery(Function f, int64_t deltaMilliSeconds);
/**
* Repeat f until the scheduler is stopped. First run is after delta has passed once.
*
* The timing is not exact: Every time f is finished, it is rescheduled to run again after delta. If you need more
* accurate scheduling, don't use this method.
*/
void scheduleEvery(Function f, std::chrono::milliseconds delta);

/**
* Mock the scheduler to fast forward in time.
Expand Down
12 changes: 6 additions & 6 deletions src/test/scheduler_tests.cpp
Expand Up @@ -150,10 +150,10 @@ BOOST_AUTO_TEST_CASE(mockforward)
CScheduler::Function dummy = [&counter]{counter++;};

// schedule jobs for 2, 5 & 8 minutes into the future
int64_t min_in_milli = 60*1000;
scheduler.scheduleFromNow(dummy, 2*min_in_milli);
scheduler.scheduleFromNow(dummy, 5*min_in_milli);
scheduler.scheduleFromNow(dummy, 8*min_in_milli);

scheduler.scheduleFromNow(dummy, std::chrono::minutes{2});
scheduler.scheduleFromNow(dummy, std::chrono::minutes{5});
scheduler.scheduleFromNow(dummy, std::chrono::minutes{8});

// check taskQueue
std::chrono::system_clock::time_point first, last;
Expand All @@ -163,10 +163,10 @@ BOOST_AUTO_TEST_CASE(mockforward)
std::thread scheduler_thread([&]() { scheduler.serviceQueue(); });

// bump the scheduler forward 5 minutes
scheduler.MockForward(std::chrono::seconds(5*60));
scheduler.MockForward(std::chrono::minutes{5});

// ensure scheduler has chance to process all tasks queued for before 1 ms from now.
scheduler.scheduleFromNow([&scheduler]{ scheduler.stop(false); }, 1);
scheduler.scheduleFromNow([&scheduler] { scheduler.stop(false); }, std::chrono::milliseconds{1});
scheduler_thread.join();

// check that the queue only has one job remaining
Expand Down
4 changes: 2 additions & 2 deletions src/wallet/load.cpp
Expand Up @@ -88,8 +88,8 @@ void StartWallets(CScheduler& scheduler)
}

// Schedule periodic wallet flushes and tx rebroadcasts
scheduler.scheduleEvery(MaybeCompactWalletDB, 500);
scheduler.scheduleEvery(MaybeResendWalletTxs, 1000);
scheduler.scheduleEvery(MaybeCompactWalletDB, std::chrono::milliseconds{500});
scheduler.scheduleEvery(MaybeResendWalletTxs, std::chrono::milliseconds{1000});
}

void FlushWallets()
Expand Down

0 comments on commit fadafb8

Please sign in to comment.