1 files changed, 82 insertions, 7 deletions

diff --git a/src/scheduler.cpp b/src/scheduler.cpp
index b01170074d..4edb2c6d9b 100644
--- a/src/scheduler.cpp
+++ b/src/scheduler.cpp
@@ -4,6 +4,7 @@
 
 #include "scheduler.h"
 
+#include "random.h"
 #include "reverselock.h"
 
 #include <assert.h>
@@ -23,7 +24,9 @@ CScheduler::~CScheduler()
 #if BOOST_VERSION < 105000
 static boost::system_time toPosixTime(const boost::chrono::system_clock::time_point& t)
 {
-    return boost::posix_time::from_time_t(boost::chrono::system_clock::to_time_t(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
 
@@ -37,6 +40,11 @@ void CScheduler::serviceQueue()
     // is called.
     while (!shouldStop()) {
         try {
+            if (!shouldStop() && taskQueue.empty()) {
+                reverse_lock<boost::unique_lock<boost::mutex> > rlock(lock);
+                // Use this chance to get a tiny bit more entropy
+                RandAddSeedSleep();
+            }
             while (!shouldStop() && taskQueue.empty()) {
                 // Wait until there is something to do.
                 newTaskScheduled.wait(lock);
@@ -104,20 +112,20 @@ void CScheduler::schedule(CScheduler::Function f, boost::chrono::system_clock::t
     newTaskScheduled.notify_one();
 }
 
-void CScheduler::scheduleFromNow(CScheduler::Function f, int64_t deltaSeconds)
+void CScheduler::scheduleFromNow(CScheduler::Function f, int64_t deltaMilliSeconds)
 {
-    schedule(f, boost::chrono::system_clock::now() + boost::chrono::seconds(deltaSeconds));
+    schedule(f, boost::chrono::system_clock::now() + boost::chrono::milliseconds(deltaMilliSeconds));
 }
 
-static void Repeat(CScheduler* s, CScheduler::Function f, int64_t deltaSeconds)
+static void Repeat(CScheduler* s, CScheduler::Function f, int64_t deltaMilliSeconds)
 {
     f();
-    s->scheduleFromNow(boost::bind(&Repeat, s, f, deltaSeconds), deltaSeconds);
+    s->scheduleFromNow(boost::bind(&Repeat, s, f, deltaMilliSeconds), deltaMilliSeconds);
 }
 
-void CScheduler::scheduleEvery(CScheduler::Function f, int64_t deltaSeconds)
+void CScheduler::scheduleEvery(CScheduler::Function f, int64_t deltaMilliSeconds)
 {
-    scheduleFromNow(boost::bind(&Repeat, this, f, deltaSeconds), deltaSeconds);
+    scheduleFromNow(boost::bind(&Repeat, this, f, deltaMilliSeconds), deltaMilliSeconds);
 }
 
 size_t CScheduler::getQueueInfo(boost::chrono::system_clock::time_point &first,
@@ -131,3 +139,70 @@ size_t CScheduler::getQueueInfo(boost::chrono::system_clock::time_point &first,
     }
     return result;
 }
+
+bool CScheduler::AreThreadsServicingQueue() const {
+    boost::unique_lock<boost::mutex> 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));
+}
+
+void SingleThreadedSchedulerClient::ProcessQueue() {
+    std::function<void (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 (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();
+    }
+}