1 // Copyright (c) 2015-2018 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 //
9 // NOTE:
10 // boost::thread / boost::chrono should be ported to std::thread / std::chrono
11 // when we support C++11.
12 //
13 #include <boost/chrono/chrono.hpp>
14 #include <boost/thread.hpp>
15 #include <map>
16 
17 #include <sync.h>
18 
19 //
20 // Simple class for background tasks that should be run
21 // periodically or once "after a while"
22 //
23 // Usage:
24 //
25 // CScheduler* s = new CScheduler();
26 // s->scheduleFromNow(doSomething, 11); // Assuming a: void doSomething() { }
27 // s->scheduleFromNow(std::bind(Class::func, this, argument), 3);
28 // boost::thread* t = new boost::thread(std::bind(CScheduler::serviceQueue, s));
29 //
30 // ... then at program shutdown, clean up the thread running serviceQueue:
31 // t->interrupt();
32 // t->join();
33 // delete t;
34 // delete s; // Must be done after thread is interrupted/joined.
35 //
36 
37 class CScheduler
38 {
39 public:
40     CScheduler();
41     ~CScheduler();
42 
43     typedef std::function<void()> Function;
44 
45     // Call func at/after time t
46     void schedule(Function f, boost::chrono::system_clock::time_point t=boost::chrono::system_clock::now());
47 
48     // Convenience method: call f once deltaMilliSeconds from now
49     void scheduleFromNow(Function f, int64_t deltaMilliSeconds);
50 
51     // Another convenience method: call f approximately
52     // every deltaMilliSeconds forever, starting deltaMilliSeconds from now.
53     // To be more precise: every time f is finished, it
54     // is rescheduled to run deltaMilliSeconds later. If you
55     // need more accurate scheduling, don't use this method.
56     void scheduleEvery(Function f, int64_t deltaMilliSeconds);
57 
58     // To keep things as simple as possible, there is no unschedule.
59 
60     // Services the queue 'forever'. Should be run in a thread,
61     // and interrupted using boost::interrupt_thread
62     void serviceQueue();
63 
64     // Tell any threads running serviceQueue to stop as soon as they're
65     // done servicing whatever task they're currently servicing (drain=false)
66     // or when there is no work left to be done (drain=true)
67     void stop(bool drain=false);
68 
69     // Returns number of tasks waiting to be serviced,
70     // and first and last task times
71     size_t getQueueInfo(boost::chrono::system_clock::time_point &first,
72                         boost::chrono::system_clock::time_point &last) const;
73 
74     // Returns true if there are threads actively running in serviceQueue()
75     bool AreThreadsServicingQueue() const;
76 
77 private:
78     std::multimap<boost::chrono::system_clock::time_point, Function> taskQueue;
79     boost::condition_variable newTaskScheduled;
80     mutable boost::mutex newTaskMutex;
81     int nThreadsServicingQueue;
82     bool stopRequested;
83     bool stopWhenEmpty;
shouldStop()84     bool shouldStop() const { return stopRequested || (stopWhenEmpty && taskQueue.empty()); }
85 };
86 
87 /**
88  * Class used by CScheduler clients which may schedule multiple jobs
89  * which are required to be run serially. Jobs may not be run on the
90  * same thread, but no two jobs will be executed
91  * at the same time and memory will be release-acquire consistent
92  * (the scheduler will internally do an acquire before invoking a callback
93  * as well as a release at the end). In practice this means that a callback
94  * B() will be able to observe all of the effects of callback A() which executed
95  * before it.
96  */
97 class SingleThreadedSchedulerClient {
98 private:
99     CScheduler *m_pscheduler;
100 
101     CCriticalSection m_cs_callbacks_pending;
102     std::list<std::function<void ()>> m_callbacks_pending GUARDED_BY(m_cs_callbacks_pending);
103     bool m_are_callbacks_running GUARDED_BY(m_cs_callbacks_pending) = false;
104 
105     void MaybeScheduleProcessQueue();
106     void ProcessQueue();
107 
108 public:
SingleThreadedSchedulerClient(CScheduler * pschedulerIn)109     explicit SingleThreadedSchedulerClient(CScheduler *pschedulerIn) : m_pscheduler(pschedulerIn) {}
110 
111     /**
112      * Add a callback to be executed. Callbacks are executed serially
113      * and memory is release-acquire consistent between callback executions.
114      * Practically, this means that callbacks can behave as if they are executed
115      * in order by a single thread.
116      */
117     void AddToProcessQueue(std::function<void ()> func);
118 
119     // Processes all remaining queue members on the calling thread, blocking until queue is empty
120     // Must be called after the CScheduler has no remaining processing threads!
121     void EmptyQueue();
122 
123     size_t CallbacksPending();
124 };
125 
126 #endif
127