/bitcoin/src/checkqueue.h

Source
// Copyright (c) 2012-present The Bitcoin Core developers
// Distributed under the MIT software license, see the accompanying
// file COPYING or http://www.opensource.org/licenses/mit-license.php.

#ifndef BITCOIN_CHECKQUEUE_H
#define BITCOIN_CHECKQUEUE_H

#include <sync.h>
#include <tinyformat.h>
#include <util/log.h>
#include <util/threadnames.h>

#include <algorithm>
#include <iterator>
#include <optional>
#include <vector>

/**
 * Queue for verifications that have to be performed.
  * The verifications are represented by a type T, which must provide an
  * operator(), returning an std::optional<R>.
  *
  * The overall result of the computation is std::nullopt if all invocations
  * return std::nullopt, or one of the other results otherwise.
  *
  * One thread (the master) is assumed to push batches of verifications
  * onto the queue, where they are processed by N-1 worker threads. When
  * the master is done adding work, it temporarily joins the worker pool
  * as an N'th worker, until all jobs are done.
  *
  */
template <typename T, typename R = std::remove_cvref_t<decltype(std::declval<T>()().value())>>
class CCheckQueue
{
private:
    //! Mutex to protect the inner state
    Mutex m_mutex;

    //! Worker threads block on this when out of work
    std::condition_variable m_worker_cv;

    //! Master thread blocks on this when out of work
    std::condition_variable m_master_cv;

    //! The queue of elements to be processed.
    //! As the order of booleans doesn't matter, it is used as a LIFO (stack)
    std::vector<T> queue GUARDED_BY(m_mutex);

    //! The number of workers (including the master) that are idle.
    int nIdle GUARDED_BY(m_mutex){0};

    //! The total number of workers (including the master).
    int nTotal GUARDED_BY(m_mutex){0};

    //! The temporary evaluation result.
    std::optional<R> m_result GUARDED_BY(m_mutex);

    /**
     * Number of verifications that haven't completed yet.
     * This includes elements that are no longer queued, but still in the
     * worker's own batches.
     */
    unsigned int nTodo GUARDED_BY(m_mutex){0};

    //! The maximum number of elements to be processed in one batch
    const unsigned int nBatchSize;

    std::vector<std::thread> m_worker_threads;
    bool m_request_stop GUARDED_BY(m_mutex){false};

    /// \anchor checkqueue
    /** Internal function that does bulk of the verification work. If fMaster, return the final result. */
    std::optional<R> Loop(bool fMaster) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
    {
        std::condition_variable& cond = fMaster ? m_master_cv : m_worker_cv;
        std::vector<T> vChecks;
        vChecks.reserve(nBatchSize);
        unsigned int nNow = 0;
        std::optional<R> local_result;
        bool do_work;
        do {
            {
                WAIT_LOCK(m_mutex, lock);
                // first do the clean-up of the previous loop run (allowing us to do it in the same critsect)
                if (nNow) {
                    if (local_result.has_value() && !m_result.has_value()) {
                        std::swap(local_result, m_result);
                    }
                    nTodo -= nNow;
                    if (nTodo == 0 && !fMaster) {
                        // We processed the last element; inform the master it can exit and return the result
                        m_master_cv.notify_one();
                    }
                } else {
                    // first iteration
                    nTotal++;
                }
                // logically, the do loop starts here
                while (queue.empty() && !m_request_stop) {
                    if (fMaster && nTodo == 0) {
                        nTotal--;
                        std::optional<R> to_return = std::move(m_result);
                        // reset the status for new work later
                        m_result = std::nullopt;
                        // return the current status
                        return to_return;
                    }
                    nIdle++;
                    cond.wait(lock); // wait
                    nIdle--;
                }
                if (m_request_stop) {
                    // return value does not matter, because m_request_stop is only set in the destructor.
                    return std::nullopt;
                }

                // Decide how many work units to process now.
                // * Do not try to do everything at once, but aim for increasingly smaller batches so
                //   all workers finish approximately simultaneously.
                // * Try to account for idle jobs which will instantly start helping.
                // * Don't do batches smaller than 1 (duh), or larger than nBatchSize.
                nNow = std::max(1U, std::min(nBatchSize, (unsigned int)queue.size() / (nTotal + nIdle + 1)));
                auto start_it = queue.end() - nNow;
                vChecks.assign(std::make_move_iterator(start_it), std::make_move_iterator(queue.end()));
                queue.erase(start_it, queue.end());
                // Check whether we need to do work at all
                do_work = !m_result.has_value();
            }
            // execute work
            if (do_work) {
                for (T& check : vChecks) {
                    local_result = check();
                    if (local_result.has_value()) break;
                }
            }
            vChecks.clear();
        } while (true);
    }

public:
    //! Mutex to ensure only one concurrent CCheckQueueControl
    Mutex m_control_mutex;

    //! Create a new check queue
    explicit CCheckQueue(unsigned int batch_size, int worker_threads_num)
        : nBatchSize(batch_size)
    {
        LogInfo("Script verification uses %d additional threads", worker_threads_num);
        m_worker_threads.reserve(worker_threads_num);
        for (int n = 0; n < worker_threads_num; ++n) {
            m_worker_threads.emplace_back([this, n]() {
                util::ThreadRename(strprintf("scriptch.%i", n));
                Loop(false /* worker thread */);
            });
        }
    }

    // Since this class manages its own resources, which is a thread
    // pool `m_worker_threads`, copy and move operations are not appropriate.
    CCheckQueue(const CCheckQueue&) = delete;
    CCheckQueue& operator=(const CCheckQueue&) = delete;
    CCheckQueue(CCheckQueue&&) = delete;
    CCheckQueue& operator=(CCheckQueue&&) = delete;

    //! Join the execution until completion. If at least one evaluation wasn't successful, return
    //! its error.
    std::optional<R> Complete() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
    {
        return Loop(true /* master thread */);
    }

    //! Add a batch of checks to the queue
    void Add(std::vector<T>&& vChecks) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
    {
        if (vChecks.empty()) {
            return;
        }

        {
            LOCK(m_mutex);
            queue.insert(queue.end(), std::make_move_iterator(vChecks.begin()), std::make_move_iterator(vChecks.end()));
            nTodo += vChecks.size();
        }

        if (vChecks.size() == 1) {
            m_worker_cv.notify_one();
        } else {
            m_worker_cv.notify_all();
        }
    }

    ~CCheckQueue()
    {
        WITH_LOCK(m_mutex, m_request_stop = true);
        m_worker_cv.notify_all();
        for (std::thread& t : m_worker_threads) {
            t.join();
        }
    }

    bool HasThreads() const { return !m_worker_threads.empty(); }
};

/**
 * RAII-style controller object for a CCheckQueue that guarantees the passed
 * queue is finished before continuing.
 */
template <typename T, typename R = std::remove_cvref_t<decltype(std::declval<T>()().value())>>
class SCOPED_LOCKABLE CCheckQueueControl
{
private:
    CCheckQueue<T, R>& m_queue;
    UniqueLock<Mutex> m_lock;
    bool fDone;

public:
    CCheckQueueControl() = delete;
    CCheckQueueControl(const CCheckQueueControl&) = delete;
    CCheckQueueControl& operator=(const CCheckQueueControl&) = delete;
    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) {}

    std::optional<R> Complete()
    {
        auto ret = m_queue.Complete();
        fDone = true;
        return ret;
    }

    void Add(std::vector<T>&& vChecks)
    {
        m_queue.Add(std::move(vChecks));
    }

    ~CCheckQueueControl() UNLOCK_FUNCTION()
    {
        if (!fDone)
            Complete();
    }
};

#endif // BITCOIN_CHECKQUEUE_H

Coverage Report

Created: 2026-06-01 18:35

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