1 // Copyright (c) 2015-2020 The Bitcoin Core developers 2 // Distributed under the MIT software license, see the accompanying 3 // file COPYING or http://www.opensource.org/licenses/mit-license.php. 4 5 #ifndef BITCOIN_SCHEDULER_H 6 #define BITCOIN_SCHEDULER_H 7 8 #include <condition_variable> 9 #include <functional> 10 #include <list> 11 #include <map> 12 #include <thread> 13 14 #include <sync.h> 15 16 /** 17 * Simple class for background tasks that should be run 18 * periodically or once "after a while" 19 * 20 * Usage: 21 * 22 * CScheduler* s = new CScheduler(); 23 * s->scheduleFromNow(doSomething, std::chrono::milliseconds{11}); // Assuming a: void doSomething() { } 24 * s->scheduleFromNow([=] { this->func(argument); }, std::chrono::milliseconds{3}); 25 * std::thread* t = new std::thread([&] { s->serviceQueue(); }); 26 * 27 * ... then at program shutdown, make sure to call stop() to clean up the thread(s) running serviceQueue: 28 * s->stop(); 29 * t->join(); 30 * delete t; 31 * delete s; // Must be done after thread is interrupted/joined. 32 */ 33 class CScheduler 34 { 35 public: 36 CScheduler(); 37 ~CScheduler(); 38 39 std::thread m_service_thread; 40 41 typedef std::function<void()> Function; 42 43 /** Call func at/after time t */ 44 void schedule(Function f, std::chrono::system_clock::time_point t); 45 46 /** Call f once after the delta has passed */ scheduleFromNow(Function f,std::chrono::milliseconds delta)47 void scheduleFromNow(Function f, std::chrono::milliseconds delta) 48 { 49 schedule(std::move(f), std::chrono::system_clock::now() + delta); 50 } 51 52 /** 53 * Repeat f until the scheduler is stopped. First run is after delta has passed once. 54 * 55 * The timing is not exact: Every time f is finished, it is rescheduled to run again after delta. If you need more 56 * accurate scheduling, don't use this method. 57 */ 58 void scheduleEvery(Function f, std::chrono::milliseconds delta); 59 60 /** 61 * Mock the scheduler to fast forward in time. 62 * Iterates through items on taskQueue and reschedules them 63 * to be delta_seconds sooner. 64 */ 65 void MockForward(std::chrono::seconds delta_seconds); 66 67 /** 68 * Services the queue 'forever'. Should be run in a thread. 69 */ 70 void serviceQueue(); 71 72 /** Tell any threads running serviceQueue to stop as soon as the current task is done */ stop()73 void stop() 74 { 75 WITH_LOCK(newTaskMutex, stopRequested = true); 76 newTaskScheduled.notify_all(); 77 if (m_service_thread.joinable()) m_service_thread.join(); 78 } 79 /** Tell any threads running serviceQueue to stop when there is no work left to be done */ StopWhenDrained()80 void StopWhenDrained() 81 { 82 WITH_LOCK(newTaskMutex, stopWhenEmpty = true); 83 newTaskScheduled.notify_all(); 84 if (m_service_thread.joinable()) m_service_thread.join(); 85 } 86 87 /** 88 * Returns number of tasks waiting to be serviced, 89 * and first and last task times 90 */ 91 size_t getQueueInfo(std::chrono::system_clock::time_point& first, 92 std::chrono::system_clock::time_point& last) const; 93 94 /** Returns true if there are threads actively running in serviceQueue() */ 95 bool AreThreadsServicingQueue() const; 96 97 private: 98 mutable Mutex newTaskMutex; 99 std::condition_variable newTaskScheduled; 100 std::multimap<std::chrono::system_clock::time_point, Function> taskQueue GUARDED_BY(newTaskMutex); GUARDED_BY(newTaskMutex)101 int nThreadsServicingQueue GUARDED_BY(newTaskMutex){0}; GUARDED_BY(newTaskMutex)102 bool stopRequested GUARDED_BY(newTaskMutex){false}; GUARDED_BY(newTaskMutex)103 bool stopWhenEmpty GUARDED_BY(newTaskMutex){false}; shouldStop()104 bool shouldStop() const EXCLUSIVE_LOCKS_REQUIRED(newTaskMutex) { return stopRequested || (stopWhenEmpty && taskQueue.empty()); } 105 }; 106 107 /** 108 * Class used by CScheduler clients which may schedule multiple jobs 109 * which are required to be run serially. Jobs may not be run on the 110 * same thread, but no two jobs will be executed 111 * at the same time and memory will be release-acquire consistent 112 * (the scheduler will internally do an acquire before invoking a callback 113 * as well as a release at the end). In practice this means that a callback 114 * B() will be able to observe all of the effects of callback A() which executed 115 * before it. 116 */ 117 class SingleThreadedSchedulerClient 118 { 119 private: 120 CScheduler* m_pscheduler; 121 122 RecursiveMutex m_cs_callbacks_pending; 123 std::list<std::function<void()>> m_callbacks_pending GUARDED_BY(m_cs_callbacks_pending); 124 bool m_are_callbacks_running GUARDED_BY(m_cs_callbacks_pending) = false; 125 126 void MaybeScheduleProcessQueue(); 127 void ProcessQueue(); 128 129 public: SingleThreadedSchedulerClient(CScheduler * pschedulerIn)130 explicit SingleThreadedSchedulerClient(CScheduler* pschedulerIn) : m_pscheduler(pschedulerIn) {} 131 132 /** 133 * Add a callback to be executed. Callbacks are executed serially 134 * and memory is release-acquire consistent between callback executions. 135 * Practically, this means that callbacks can behave as if they are executed 136 * in order by a single thread. 137 */ 138 void AddToProcessQueue(std::function<void()> func); 139 140 /** 141 * Processes all remaining queue members on the calling thread, blocking until queue is empty 142 * Must be called after the CScheduler has no remaining processing threads! 143 */ 144 void EmptyQueue(); 145 146 size_t CallbacksPending(); 147 }; 148 149 #endif 150