Scheduling background work

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