Line data    Source code

       1             : // Copyright (c) 2015 The Bitcoin Core developers
       2             : // Copyright (c) 2017-2021 The PIVX Core developers
       3             : // Distributed under the MIT software license, see the accompanying
       4             : // file COPYING or http://www.opensource.org/licenses/mit-license.php.
       5             : 
       6             : #include "scheduler.h"
       7             : 
       8             : #include "random.h"
       9             : 
      10             : #include <assert.h>
      11             : #include <utility>
      12             : 
      13         448 : CScheduler::CScheduler() : nThreadsServicingQueue(0), stopRequested(false), stopWhenEmpty(false)
      14             : {
      15         448 : }
      16             : 
      17         448 : CScheduler::~CScheduler()
      18             : {
      19         448 :     assert(nThreadsServicingQueue == 0);
      20         448 : }
      21             : 
      22         448 : void CScheduler::serviceQueue()
      23             : {
      24         448 :     WAIT_LOCK(newTaskMutex, lock);
      25         448 :     ++nThreadsServicingQueue;
      26             : 
      27             :     // newTaskMutex is locked throughout this loop EXCEPT
      28             :     // when the thread is waiting or when the user's function
      29             :     // is called.
      30      163432 :     while (!shouldStop()) {
      31      162984 :         try {
      32      163437 :             if (!shouldStop() && taskQueue.empty()) {
      33        1972 :                 REVERSE_LOCK(lock);
      34             :             }
      35      329804 :             while (!shouldStop() && taskQueue.empty()) {
      36             :                 // Wait until there is something to do.
      37        1972 :                 newTaskScheduled.wait(lock);
      38             :             }
      39             : 
      40             :             // Wait until either there is a new task, or until
      41             :             // the time of the first item on the queue:
      42             : 
      43      376015 :             while (!shouldStop() && !taskQueue.empty()) {
      44      187796 :                 std::chrono::system_clock::time_point timeToWaitFor = taskQueue.begin()->first;
      45      187796 :                 if (newTaskScheduled.wait_until(lock, timeToWaitFor) == std::cv_status::timeout) {
      46             :                     break; // Exit loop after timeout, it means we reached the time of the event
      47             :                 }
      48             :             }
      49             : 
      50             :             // If there are multiple threads, the queue can empty while we're waiting (another
      51             :             // thread may service the task we were waiting on).
      52      325528 :             if (shouldStop() || taskQueue.empty())
      53         444 :                 continue;
      54             : 
      55      325080 :             Function f = taskQueue.begin()->second;
      56      162540 :             taskQueue.erase(taskQueue.begin());
      57             : 
      58      162540 :             {
      59             :                 // Unlock before calling f, so it can reschedule itself or another task
      60             :                 // without deadlocking:
      61      325080 :                 REVERSE_LOCK(lock);
      62      162540 :                 f();
      63             :             }
      64           0 :         } catch (...) {
      65           0 :             --nThreadsServicingQueue;
      66           0 :             throw;
      67             :         }
      68             :     }
      69         448 :     --nThreadsServicingQueue;
      70         448 :     newTaskScheduled.notify_one();
      71         448 : }
      72             : 
      73         435 : void CScheduler::stop(bool drain)
      74             : {
      75         435 :     {
      76         435 :         LOCK(newTaskMutex);
      77         435 :         if (drain)
      78           2 :             stopWhenEmpty = true;
      79             :         else
      80         433 :             stopRequested = true;
      81             :     }
      82         435 :     newTaskScheduled.notify_all();
      83         435 : }
      84             : 
      85      164356 : void CScheduler::schedule(CScheduler::Function f, std::chrono::system_clock::time_point t)
      86             : {
      87      164356 :     {
      88      164356 :         LOCK(newTaskMutex);
      89      164356 :         taskQueue.emplace(t, f);
      90             :     }
      91      164356 :     newTaskScheduled.notify_one();
      92      164356 : }
      93             : 
      94       43147 : void CScheduler::scheduleFromNow(CScheduler::Function f, int64_t deltaMilliSeconds)
      95             : {
      96       43147 :     schedule(f, std::chrono::system_clock::now() + std::chrono::milliseconds(deltaMilliSeconds));
      97       43147 : }
      98             : 
      99       18469 : static void Repeat(CScheduler* s, CScheduler::Function f, int64_t deltaMilliSeconds)
     100             : {
     101       18469 :     f();
     102       36938 :     s->scheduleFromNow(std::bind(&Repeat, s, f, deltaMilliSeconds), deltaMilliSeconds);
     103       18469 : }
     104             : 
     105        1760 : void CScheduler::scheduleEvery(CScheduler::Function f, int64_t deltaMilliSeconds)
     106             : {
     107        3520 :     scheduleFromNow(std::bind(&Repeat, this, f, deltaMilliSeconds), deltaMilliSeconds);
     108        1760 : }
     109             : 
     110           2 : size_t CScheduler::getQueueInfo(std::chrono::system_clock::time_point &first,
     111             :                              std::chrono::system_clock::time_point &last) const
     112             : {
     113           2 :     LOCK(newTaskMutex);
     114           2 :     size_t result = taskQueue.size();
     115           2 :     if (!taskQueue.empty()) {
     116           1 :         first = taskQueue.begin()->first;
     117           1 :         last = taskQueue.rbegin()->first;
     118             :     }
     119           2 :     return result;
     120             : }
     121             : 
     122         433 : bool CScheduler::AreThreadsServicingQueue() const {
     123         433 :     LOCK(newTaskMutex);
     124         433 :     return nThreadsServicingQueue;
     125             : }
     126             : 
     127      206624 : void SingleThreadedSchedulerClient::MaybeScheduleProcessQueue() {
     128      206624 :     {
     129      206624 :         LOCK(m_cs_callbacks_pending);
     130             :         // Try to avoid scheduling too many copies here, but if we
     131             :         // accidentally have two ProcessQueue's scheduled at once its
     132             :         // not a big deal.
     133      292439 :         if (m_are_callbacks_running) return;
     134      148058 :         if (m_callbacks_pending.empty()) return;
     135             :     }
     136      241618 :     m_pscheduler->schedule(std::bind(&SingleThreadedSchedulerClient::ProcessQueue, this), std::chrono::system_clock::now());
     137             : }
     138             : 
     139      121211 : void SingleThreadedSchedulerClient::ProcessQueue() {
     140      224523 :     std::function<void (void)> callback;
     141      121211 :     {
     142      121211 :         LOCK(m_cs_callbacks_pending);
     143      139110 :         if (m_are_callbacks_running) return;
     144      121209 :         if (m_callbacks_pending.empty()) return;
     145      103312 :         m_are_callbacks_running = true;
     146             : 
     147      103312 :         callback = std::move(m_callbacks_pending.front());
     148      103312 :         m_callbacks_pending.pop_front();
     149             :     }
     150             : 
     151             :     // RAII the setting of fCallbacksRunning and calling MaybeScheduleProcessQueue
     152             :     // to ensure both happen safely even if callback() throws.
     153      103312 :     struct RAIICallbacksRunning {
     154             :         SingleThreadedSchedulerClient* instance;
     155      103312 :         explicit RAIICallbacksRunning(SingleThreadedSchedulerClient* _instance) : instance(_instance) {}
     156      206624 :         ~RAIICallbacksRunning() {
     157      103312 :             {
     158      103312 :                 LOCK(instance->m_cs_callbacks_pending);
     159      103312 :                 instance->m_are_callbacks_running = false;
     160             :             }
     161      103312 :             instance->MaybeScheduleProcessQueue();
     162      103312 :         }
     163      103312 :     } raiicallbacksrunning(this);
     164             : 
     165      103312 :     callback();
     166             : }
     167             : 
     168      103312 : void SingleThreadedSchedulerClient::AddToProcessQueue(std::function<void (void)> func) {
     169      103312 :     assert(m_pscheduler);
     170             : 
     171      103312 :     {
     172      103312 :         LOCK(m_cs_callbacks_pending);
     173      206624 :         m_callbacks_pending.emplace_back(std::move(func));
     174             :     }
     175      103312 :     MaybeScheduleProcessQueue();
     176      103312 : }
     177             : 
     178         433 : void SingleThreadedSchedulerClient::EmptyQueue() {
     179         433 :     assert(!m_pscheduler->AreThreadsServicingQueue());
     180             :     bool should_continue = true;
     181         885 :     while (should_continue) {
     182         452 :         ProcessQueue();
     183         904 :         LOCK(m_cs_callbacks_pending);
     184         452 :         should_continue = !m_callbacks_pending.empty();
     185             :     }
     186         433 : }
     187             : 
     188       28782 : size_t SingleThreadedSchedulerClient::CallbacksPending() {
     189       28782 :     LOCK(m_cs_callbacks_pending);
     190       28782 :     return m_callbacks_pending.size();
     191             : }