Branch data Line data Source code
1 : : // Copyright (c) 2015-2022 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 : : #include <scheduler.h>
6 : :
7 : : #include <sync.h>
8 : : #include <util/time.h>
9 : :
10 : : #include <cassert>
11 : : #include <functional>
12 : : #include <utility>
13 : :
14 : 1158 : CScheduler::CScheduler() = default;
15 : :
16 : 1158 : CScheduler::~CScheduler()
17 : : {
18 [ - + ]: 1158 : assert(nThreadsServicingQueue == 0);
19 [ + + - + ]: 1158 : if (stopWhenEmpty) assert(taskQueue.empty());
20 : 1158 : }
21 : :
22 : :
23 : 1171 : void CScheduler::serviceQueue()
24 : : {
25 : 1171 : WAIT_LOCK(newTaskMutex, lock);
26 : 1171 : ++nThreadsServicingQueue;
27 : :
28 : : // newTaskMutex is locked throughout this loop EXCEPT
29 : : // when the thread is waiting or when the user's function
30 : : // is called.
31 [ + + ]: 513275 : while (!shouldStop()) {
32 : : try {
33 [ + + + + ]: 1043079 : while (!shouldStop() && taskQueue.empty()) {
34 : : // Wait until there is something to do.
35 [ + - ]: 9511 : newTaskScheduled.wait(lock);
36 : : }
37 : :
38 : : // Wait until either there is a new task, or until
39 : : // the time of the first item on the queue:
40 : :
41 [ + + + - ]: 1275362 : while (!shouldStop() && !taskQueue.empty()) {
42 [ + - ]: 637114 : std::chrono::steady_clock::time_point timeToWaitFor = taskQueue.begin()->first;
43 [ + + + - ]: 637114 : if (newTaskScheduled.wait_until(lock, timeToWaitFor) == std::cv_status::timeout) {
44 : : break; // Exit loop after timeout, it means we reached the time of the event
45 : : }
46 : : }
47 : :
48 : : // If there are multiple threads, the queue can empty while we're waiting (another
49 : : // thread may service the task we were waiting on).
50 [ + + + - ]: 1023055 : if (shouldStop() || taskQueue.empty())
51 : 1153 : continue;
52 : :
53 [ + - ]: 510951 : Function f = taskQueue.begin()->second;
54 : 510951 : taskQueue.erase(taskQueue.begin());
55 : :
56 : 510951 : {
57 : : // Unlock before calling f, so it can reschedule itself or another task
58 : : // without deadlocking:
59 [ + - ]: 510951 : REVERSE_LOCK(lock);
60 [ + - ]: 510951 : f();
61 : 510951 : }
62 : 510951 : } catch (...) {
63 : 0 : --nThreadsServicingQueue;
64 : 0 : throw;
65 : 0 : }
66 : : }
67 : 1171 : --nThreadsServicingQueue;
68 [ + - ]: 1171 : newTaskScheduled.notify_one();
69 : 1171 : }
70 : :
71 : 521993 : void CScheduler::schedule(CScheduler::Function f, std::chrono::steady_clock::time_point t)
72 : : {
73 : 521993 : {
74 : 521993 : LOCK(newTaskMutex);
75 [ + - + - : 1043986 : taskQueue.insert(std::make_pair(t, f));
+ - ]
76 : 521993 : }
77 : 521993 : newTaskScheduled.notify_one();
78 : 521993 : }
79 : :
80 : 17 : void CScheduler::MockForward(std::chrono::seconds delta_seconds)
81 : : {
82 [ + - - + ]: 17 : assert(delta_seconds > 0s && delta_seconds <= 1h);
83 : :
84 : 17 : {
85 : 17 : LOCK(newTaskMutex);
86 : :
87 : : // use temp_queue to maintain updated schedule
88 : 17 : std::multimap<std::chrono::steady_clock::time_point, Function> temp_queue;
89 : :
90 [ + + ]: 162 : for (const auto& element : taskQueue) {
91 [ + - ]: 145 : temp_queue.emplace_hint(temp_queue.cend(), element.first - delta_seconds, element.second);
92 : : }
93 : :
94 : : // point taskQueue to temp_queue
95 : 17 : taskQueue = std::move(temp_queue);
96 [ + - ]: 17 : }
97 : :
98 : : // notify that the taskQueue needs to be processed
99 : 17 : newTaskScheduled.notify_one();
100 : 17 : }
101 : :
102 : 14216 : static void Repeat(CScheduler& s, CScheduler::Function f, std::chrono::milliseconds delta)
103 : : {
104 : 14216 : f();
105 [ + - + - : 140646 : s.scheduleFromNow([=, &s] { Repeat(s, f, delta); }, delta);
+ - + - ]
106 : 14216 : }
107 : :
108 : 6214 : void CScheduler::scheduleEvery(CScheduler::Function f, std::chrono::milliseconds delta)
109 : : {
110 [ + - + - : 39034 : scheduleFromNow([this, f, delta] { Repeat(*this, f, delta); }, delta);
+ - + - ]
111 : 6214 : }
112 : :
113 : 4 : size_t CScheduler::getQueueInfo(std::chrono::steady_clock::time_point& first,
114 : : std::chrono::steady_clock::time_point& last) const
115 : : {
116 : 4 : LOCK(newTaskMutex);
117 [ + + ]: 4 : size_t result = taskQueue.size();
118 [ + + ]: 4 : if (!taskQueue.empty()) {
119 : 3 : first = taskQueue.begin()->first;
120 : 3 : last = taskQueue.rbegin()->first;
121 : : }
122 [ + - ]: 4 : return result;
123 : 4 : }
124 : :
125 : 1155 : bool CScheduler::AreThreadsServicingQueue() const
126 : : {
127 : 1155 : LOCK(newTaskMutex);
128 [ + - ]: 1155 : return nThreadsServicingQueue;
129 : 1155 : }
130 : :
131 : :
132 : 885403 : void SerialTaskRunner::MaybeScheduleProcessQueue()
133 : : {
134 : 885403 : {
135 : 885403 : LOCK(m_callbacks_mutex);
136 : : // Try to avoid scheduling too many copies here, but if we
137 : : // accidentally have two ProcessQueue's scheduled at once its
138 : : // not a big deal.
139 [ + + ]: 885403 : if (m_are_callbacks_running) return;
140 [ + + ]: 634013 : if (m_callbacks_pending.empty()) return;
141 : 385138 : }
142 [ + - ]: 1496851 : m_scheduler.schedule([this] { this->ProcessQueue(); }, std::chrono::steady_clock::now());
143 : : }
144 : :
145 : 497604 : void SerialTaskRunner::ProcessQueue()
146 : : {
147 [ + - ]: 497604 : std::function<void()> callback;
148 : 497604 : {
149 [ + - ]: 497604 : LOCK(m_callbacks_mutex);
150 [ + - ]: 497604 : if (m_are_callbacks_running) return;
151 [ + + ]: 497604 : if (m_callbacks_pending.empty()) return;
152 : 442250 : m_are_callbacks_running = true;
153 : :
154 : 442250 : callback = std::move(m_callbacks_pending.front());
155 [ + - ]: 442250 : m_callbacks_pending.pop_front();
156 : 55354 : }
157 : :
158 : : // RAII the setting of fCallbacksRunning and calling MaybeScheduleProcessQueue
159 : : // to ensure both happen safely even if callback() throws.
160 : 442250 : struct RAIICallbacksRunning {
161 : : SerialTaskRunner* instance;
162 : 442250 : explicit RAIICallbacksRunning(SerialTaskRunner* _instance) : instance(_instance) {}
163 : 442250 : ~RAIICallbacksRunning()
164 : : {
165 : 442250 : {
166 : 442250 : LOCK(instance->m_callbacks_mutex);
167 [ + - ]: 442250 : instance->m_are_callbacks_running = false;
168 : 442250 : }
169 : 442250 : instance->MaybeScheduleProcessQueue();
170 : 442250 : }
171 : 442250 : } raiicallbacksrunning(this);
172 : :
173 [ + - ]: 442250 : callback();
174 : 497604 : }
175 : :
176 : 443153 : void SerialTaskRunner::insert(std::function<void()> func)
177 : : {
178 : 443153 : {
179 : 443153 : LOCK(m_callbacks_mutex);
180 [ + - ]: 443153 : m_callbacks_pending.emplace_back(std::move(func));
181 : 443153 : }
182 : 443153 : MaybeScheduleProcessQueue();
183 : 443153 : }
184 : :
185 : 1155 : void SerialTaskRunner::flush()
186 : : {
187 [ + - ]: 1155 : assert(!m_scheduler.AreThreadsServicingQueue());
188 : : bool should_continue = true;
189 [ + + ]: 2438 : while (should_continue) {
190 : 1283 : ProcessQueue();
191 : 1283 : LOCK(m_callbacks_mutex);
192 [ + - ]: 1283 : should_continue = !m_callbacks_pending.empty();
193 : 1283 : }
194 : 1155 : }
195 : :
196 : 147099 : size_t SerialTaskRunner::size()
197 : : {
198 : 147099 : LOCK(m_callbacks_mutex);
199 [ + - ]: 147099 : return m_callbacks_pending.size();
200 : : }
|
