Branch data Line data Source code
1 : : // Copyright (c) 2012-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_CHECKQUEUE_H
6 : : #define BITCOIN_CHECKQUEUE_H
7 : :
8 : : #include <sync.h>
9 : : #include <tinyformat.h>
10 : : #include <util/log.h>
11 : : #include <util/threadnames.h>
12 : :
13 : : #include <algorithm>
14 : : #include <iterator>
15 : : #include <optional>
16 : : #include <vector>
17 : :
18 : : /**
19 : : * Queue for verifications that have to be performed.
20 : : * The verifications are represented by a type T, which must provide an
21 : : * operator(), returning an std::optional<R>.
22 : : *
23 : : * The overall result of the computation is std::nullopt if all invocations
24 : : * return std::nullopt, or one of the other results otherwise.
25 : : *
26 : : * One thread (the master) is assumed to push batches of verifications
27 : : * onto the queue, where they are processed by N-1 worker threads. When
28 : : * the master is done adding work, it temporarily joins the worker pool
29 : : * as an N'th worker, until all jobs are done.
30 : : *
31 : : */
32 : : template <typename T, typename R = std::remove_cvref_t<decltype(std::declval<T>()().value())>>
33 : : class CCheckQueue
34 : : {
35 : : private:
36 : : //! Mutex to protect the inner state
37 : : Mutex m_mutex;
38 : :
39 : : //! Worker threads block on this when out of work
40 : : std::condition_variable m_worker_cv;
41 : :
42 : : //! Master thread blocks on this when out of work
43 : : std::condition_variable m_master_cv;
44 : :
45 : : //! The queue of elements to be processed.
46 : : //! As the order of booleans doesn't matter, it is used as a LIFO (stack)
47 : : std::vector<T> queue GUARDED_BY(m_mutex);
48 : :
49 : : //! The number of workers (including the master) that are idle.
50 : : int nIdle GUARDED_BY(m_mutex){0};
51 : :
52 : : //! The total number of workers (including the master).
53 : : int nTotal GUARDED_BY(m_mutex){0};
54 : :
55 : : //! The temporary evaluation result.
56 : : std::optional<R> m_result GUARDED_BY(m_mutex);
57 : :
58 : : /**
59 : : * Number of verifications that haven't completed yet.
60 : : * This includes elements that are no longer queued, but still in the
61 : : * worker's own batches.
62 : : */
63 : : unsigned int nTodo GUARDED_BY(m_mutex){0};
64 : :
65 : : //! The maximum number of elements to be processed in one batch
66 : : const unsigned int nBatchSize;
67 : :
68 : : std::vector<std::thread> m_worker_threads;
69 : : bool m_request_stop GUARDED_BY(m_mutex){false};
70 : :
71 : : /// \anchor checkqueue
72 : : /** Internal function that does bulk of the verification work. If fMaster, return the final result. */
73 : 18481 : std::optional<R> Loop(bool fMaster) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
74 : : {
75 [ + + ]: 18481 : std::condition_variable& cond = fMaster ? m_master_cv : m_worker_cv;
76 : 18481 : std::vector<T> vChecks;
77 [ + - ]: 18481 : vChecks.reserve(nBatchSize);
78 : 18481 : unsigned int nNow = 0;
79 : 18481 : std::optional<R> local_result;
80 : : bool do_work;
81 : 86829 : do {
82 : : {
83 [ + - ]: 1438410 : WAIT_LOCK(m_mutex, lock);
84 : : // first do the clean-up of the previous loop run (allowing us to do it in the same critsect)
85 [ + + ]: 1438410 : if (nNow) {
86 [ + + + - ]: 1419929 : if (local_result.has_value() && !m_result.has_value()) {
87 : 1011 : std::swap(local_result, m_result);
88 : : }
89 : 1419929 : nTodo -= nNow;
90 [ + + + + ]: 1419929 : if (nTodo == 0 && !fMaster) {
91 : : // We processed the last element; inform the master it can exit and return the result
92 : 67836 : m_master_cv.notify_one();
93 : : }
94 : : } else {
95 : : // first iteration
96 : 18481 : nTotal++;
97 : : }
98 : : // logically, the do loop starts here
99 [ + + + + ]: 1590057 : while (queue.empty() && !m_request_stop) {
100 [ + + + + ]: 169755 : if (fMaster && nTodo == 0) {
101 : 18108 : nTotal--;
102 : 18108 : std::optional<R> to_return = std::move(m_result);
103 : : // reset the status for new work later
104 [ + + ]: 18108 : m_result = std::nullopt;
105 : : // return the current status
106 : 2235 : return to_return;
107 : : }
108 : 151647 : nIdle++;
109 [ + - ]: 151647 : cond.wait(lock); // wait
110 : 151647 : nIdle--;
111 : : }
112 [ + + ]: 1420302 : if (m_request_stop) {
113 : : // return value does not matter, because m_request_stop is only set in the destructor.
114 : 373 : return std::nullopt;
115 : : }
116 : :
117 : : // Decide how many work units to process now.
118 : : // * Do not try to do everything at once, but aim for increasingly smaller batches so
119 : : // all workers finish approximately simultaneously.
120 : : // * Try to account for idle jobs which will instantly start helping.
121 : : // * Don't do batches smaller than 1 (duh), or larger than nBatchSize.
122 [ - + + + : 2784258 : nNow = std::max(1U, std::min(nBatchSize, (unsigned int)queue.size() / (nTotal + nIdle + 1)));
+ + ]
123 [ + - ]: 1419929 : auto start_it = queue.end() - nNow;
124 [ + - ]: 1419929 : vChecks.assign(std::make_move_iterator(start_it), std::make_move_iterator(queue.end()));
125 : 1419929 : queue.erase(start_it, queue.end());
126 : : // Check whether we need to do work at all
127 [ + - ]: 1419929 : do_work = !m_result.has_value();
128 : 17474 : }
129 : : // execute work
130 [ + + ]: 1419929 : if (do_work) {
131 [ + + ]: 12896440 : for (T& check : vChecks) {
[ + + + + ]
132 [ + - + + ]: 11519670 : local_result = check();
[ + + ]
133 [ + + ]: 11515008 : if (local_result.has_value()) break;
134 : : }
135 : : }
136 [ + - ]: 1421187 : vChecks.clear();
137 : 86422 : } while (true);
138 : 18481 : }
139 : :
140 : : public:
141 : : //! Mutex to ensure only one concurrent CCheckQueueControl
142 : : Mutex m_control_mutex;
143 : :
144 : : //! Create a new check queue
145 : 184 : explicit CCheckQueue(unsigned int batch_size, int worker_threads_num)
146 [ + - ]: 184 : : nBatchSize(batch_size)
147 : : {
148 [ + - ]: 184 : LogInfo("Script verification uses %d additional threads", worker_threads_num);
149 [ + - ]: 184 : m_worker_threads.reserve(worker_threads_num);
150 [ + + ]: 557 : for (int n = 0; n < worker_threads_num; ++n) {
151 [ + - ]: 403 : m_worker_threads.emplace_back([this, n]() {
152 [ + - ][ + - : 373 : util::ThreadRename(strprintf("scriptch.%i", n));
+ - + - +
- + - +
- ]
153 : 373 : Loop(false /* worker thread */);
154 : : });
155 : : }
156 : 184 : }
157 : :
158 : : // Since this class manages its own resources, which is a thread
159 : : // pool `m_worker_threads`, copy and move operations are not appropriate.
160 : : CCheckQueue(const CCheckQueue&) = delete;
161 : : CCheckQueue& operator=(const CCheckQueue&) = delete;
162 : : CCheckQueue(CCheckQueue&&) = delete;
163 : : CCheckQueue& operator=(CCheckQueue&&) = delete;
164 : :
165 : : //! Join the execution until completion. If at least one evaluation wasn't successful, return
166 : : //! its error.
167 : 18108 : std::optional<R> Complete() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
168 : : {
169 : 18108 : return Loop(true /* master thread */);
170 : : }
171 : :
172 : : //! Add a batch of checks to the queue
173 : 2654105 : void Add(std::vector<T>&& vChecks) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
174 : : {
175 [ + + ]: 2654105 : if (vChecks.empty()) {
176 : : return;
177 : : }
178 : :
179 : : {
180 : 2390008 : LOCK(m_mutex);
181 [ + - ]: 2390008 : queue.insert(queue.end(), std::make_move_iterator(vChecks.begin()), std::make_move_iterator(vChecks.end()));
182 [ + - ]: 2390008 : nTodo += vChecks.size();
183 [ - + ]: 2390008 : }
184 : :
185 [ + + ]: 2390008 : if (vChecks.size() == 1) {
186 : 270296 : m_worker_cv.notify_one();
187 : : } else {
188 : 2119712 : m_worker_cv.notify_all();
189 : : }
190 : : }
191 : :
192 : 184 : ~CCheckQueue()
193 : : {
194 [ + - ]: 368 : WITH_LOCK(m_mutex, m_request_stop = true);
195 : 184 : m_worker_cv.notify_all();
196 [ + + ]: 557 : for (std::thread& t : m_worker_threads) {
197 : 373 : t.join();
198 : : }
199 : 184 : }
200 : :
201 [ + + ]: 16451 : bool HasThreads() const { return !m_worker_threads.empty(); }
202 : : };
203 : :
204 : : /**
205 : : * RAII-style controller object for a CCheckQueue that guarantees the passed
206 : : * queue is finished before continuing.
207 : : */
208 : : template <typename T, typename R = std::remove_cvref_t<decltype(std::declval<T>()().value())>>
209 : : class SCOPED_LOCKABLE CCheckQueueControl
210 : : {
211 : : private:
212 : : CCheckQueue<T, R>& m_queue;
213 : : UniqueLock<Mutex> m_lock;
214 : : bool fDone;
215 : :
216 : : public:
217 : : CCheckQueueControl() = delete;
218 : : CCheckQueueControl(const CCheckQueueControl&) = delete;
219 : : CCheckQueueControl& operator=(const CCheckQueueControl&) = delete;
220 : 18108 : explicit CCheckQueueControl(CCheckQueue<T>& queueIn) EXCLUSIVE_LOCK_FUNCTION(queueIn.m_control_mutex) : m_queue(queueIn), m_lock(LOCK_ARGS(queueIn.m_control_mutex)), fDone(false) {}
221 : :
222 : 18108 : std::optional<R> Complete()
223 : : {
224 [ + - ]: 18108 : auto ret = m_queue.Complete();
225 [ + + ]: 18108 : fDone = true;
226 : 2235 : return ret;
227 : : }
228 : :
229 : 2654105 : void Add(std::vector<T>&& vChecks)
230 : : {
231 [ + - ][ + - : 2654105 : m_queue.Add(std::move(vChecks));
+ - + - +
- + - +
- ]
232 : 2654105 : }
233 : :
234 : 18108 : ~CCheckQueueControl() UNLOCK_FUNCTION()
235 : : {
236 [ + + ]: 18108 : if (!fDone)
237 : 1005 : Complete();
238 [ + - ]: 18108 : }
239 : : };
240 : :
241 : : #endif // BITCOIN_CHECKQUEUE_H
|
