Sharing verification work across threads

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  */
32template <typename T, typename R = std::remove_cvref_t<decltype(std::declval<T>()().value())>>
33class CCheckQueue
34{
35private:
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    std::optional<R> Loop(bool fMaster) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
74    {
75        std::condition_variable& cond = fMaster ? m_master_cv : m_worker_cv;
76        std::vector<T> vChecks;
77        vChecks.reserve(nBatchSize);
78        unsigned int nNow = 0;
79        std::optional<R> local_result;
80        bool do_work;
81        do {
82            {
83                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                if (nNow) {
86                    if (local_result.has_value() && !m_result.has_value()) {
87                        std::swap(local_result, m_result);
88                    }
89                    nTodo -= nNow;
90                    if (nTodo == 0 && !fMaster) {
91                        // We processed the last element; inform the master it can exit and return the result
92                        m_master_cv.notify_one();
93                    }
94                } else {
95                    // first iteration
96                    nTotal++;
97                }
98                // logically, the do loop starts here
99                while (queue.empty() && !m_request_stop) {
100                    if (fMaster && nTodo == 0) {
101                        nTotal--;
102                        std::optional<R> to_return = std::move(m_result);
103                        // reset the status for new work later
104                        m_result = std::nullopt;
105                        // return the current status
106                        return to_return;
107                    }
108                    nIdle++;
109                    cond.wait(lock); // wait
110                    nIdle--;
111                }
112                if (m_request_stop) {
113                    // return value does not matter, because m_request_stop is only set in the destructor.
114                    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                nNow = std::max(1U, std::min(nBatchSize, (unsigned int)queue.size() / (nTotal + nIdle + 1)));
123                auto start_it = queue.end() - nNow;
124                vChecks.assign(std::make_move_iterator(start_it), std::make_move_iterator(queue.end()));
125                queue.erase(start_it, queue.end());
126                // Check whether we need to do work at all
127                do_work = !m_result.has_value();
128            }
129            // execute work
130            if (do_work) {
131                for (T& check : vChecks) {
132                    local_result = check();
133                    if (local_result.has_value()) break;
134                }
135            }
136            vChecks.clear();
137        } while (true);
138    }
139
140public:
141    //! Mutex to ensure only one concurrent CCheckQueueControl
142    Mutex m_control_mutex;
143
144    //! Create a new check queue
145    explicit CCheckQueue(unsigned int batch_size, int worker_threads_num)
146        : nBatchSize(batch_size)
147    {
148        LogInfo("Script verification uses %d additional threads", worker_threads_num);
149        m_worker_threads.reserve(worker_threads_num);
150        for (int n = 0; n < worker_threads_num; ++n) {
151            m_worker_threads.emplace_back([this, n]() {
152                util::ThreadRename(strprintf("scriptch.%02i", n));
153                Loop(false /* worker thread */);
154            });
155        }
156    }
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    std::optional<R> Complete() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
168    {
169        return Loop(true /* master thread */);
170    }
171
172    //! Add a batch of checks to the queue
173    void Add(std::vector<T>&& vChecks) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
174    {
175        if (vChecks.empty()) {
176            return;
177        }
178
179        {
180            LOCK(m_mutex);
181            queue.insert(queue.end(), std::make_move_iterator(vChecks.begin()), std::make_move_iterator(vChecks.end()));
182            nTodo += vChecks.size();
183        }
184
185        if (vChecks.size() == 1) {
186            m_worker_cv.notify_one();
187        } else {
188            m_worker_cv.notify_all();
189        }
190    }
191
192    ~CCheckQueue()
193    {
194        WITH_LOCK(m_mutex, m_request_stop = true);
195        m_worker_cv.notify_all();
196        for (std::thread& t : m_worker_threads) {
197            t.join();
198        }
199    }
200
201    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 */
208template <typename T, typename R = std::remove_cvref_t<decltype(std::declval<T>()().value())>>
209class SCOPED_LOCKABLE CCheckQueueControl
210{
211private:
212    CCheckQueue<T, R>& m_queue;
213    UniqueLock<Mutex> m_lock;
214    bool fDone;
215
216public:
217    CCheckQueueControl() = delete;
218    CCheckQueueControl(const CCheckQueueControl&) = delete;
219    CCheckQueueControl& operator=(const CCheckQueueControl&) = delete;
220    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    std::optional<R> Complete()
223    {
224        auto ret = m_queue.Complete();
225        fDone = true;
226        return ret;
227    }
228
229    void Add(std::vector<T>&& vChecks)
230    {
231        m_queue.Add(std::move(vChecks));
232    }
233
234    ~CCheckQueueControl() UNLOCK_FUNCTION()
235    {
236        if (!fDone)
237            Complete();
238    }
239};
240
241#endif // BITCOIN_CHECKQUEUE_H
242