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