checkqueue.h

Go to the documentation of this file.

// Copyright (c) 2012-2018 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/threadnames.h>
 
#include <algorithm>
#include <iterator>
#include <optional>
#include <vector>
 
template <typename T,
          typename R =
              std::remove_cvref_t<decltype(std::declval<T>()().value())>>
class CCheckQueue {
private:
    Mutex m_mutex;
 
    std::condition_variable m_worker_cv;
 
    std::condition_variable m_master_cv;
 
    std::vector<T> queue GUARDED_BY(m_mutex);
 
    int nIdle GUARDED_BY(m_mutex){0};
 
    int nTotal GUARDED_BY(m_mutex){0};
 
    std::optional<R> m_result GUARDED_BY(m_mutex);
 
    unsigned int nTodo GUARDED_BY(m_mutex){0};
 
    const unsigned int nBatchSize;
 
    std::vector<std::thread> m_worker_threads;
    bool m_request_stop GUARDED_BY(m_mutex){false};
 
    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 m_control_mutex;
 
    explicit CCheckQueue(unsigned int nBatchSizeIn)
        : nBatchSize(nBatchSizeIn) {}
 
    void StartWorkerThreads(const int threads_num)
        EXCLUSIVE_LOCKS_REQUIRED(!m_mutex) {
        {
            LOCK(m_mutex);
            nIdle = 0;
            nTotal = 0;
            m_result = std::nullopt;
        }
        assert(m_worker_threads.empty());
        for (int n = 0; n < threads_num; ++n) {
            m_worker_threads.emplace_back([this, n]() {
                util::ThreadRename(strprintf("scriptch.%i", n));
                Loop(false /* worker thread */);
            });
        }
    }
 
    std::optional<R> Complete() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex) {
        return Loop(true /* master thread */);
    }
 
    void Add(std::vector<T> &&vChecks) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex) {
        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 if (vChecks.size() > 1) {
            m_worker_cv.notify_all();
        }
    }
 
    void StopWorkerThreads() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex) {
        WITH_LOCK(m_mutex, m_request_stop = true);
        m_worker_cv.notify_all();
        for (std::thread &t : m_worker_threads) {
            t.join();
        }
        m_worker_threads.clear();
        WITH_LOCK(m_mutex, m_request_stop = false);
    }
 
    ~CCheckQueue() { assert(m_worker_threads.empty()); }
};
 
template <typename T,
          typename R =
              std::remove_cvref_t<decltype(std::declval<T>()().value())>>
class CCheckQueueControl {
private:
    CCheckQueue<T, R> *const pqueue;
    bool fDone;
 
public:
    CCheckQueueControl() = delete;
    CCheckQueueControl(const CCheckQueueControl &) = delete;
    CCheckQueueControl &operator=(const CCheckQueueControl &) = delete;
    explicit CCheckQueueControl(CCheckQueue<T> *const pqueueIn)
        : pqueue(pqueueIn), fDone(false) {
        // passed queue is supposed to be unused, or nullptr
        if (pqueue != nullptr) {
            ENTER_CRITICAL_SECTION(pqueue->m_control_mutex);
        }
    }
 
    std::optional<R> Complete() {
        if (pqueue == nullptr) {
            return std::nullopt;
        }
        auto ret = pqueue->Complete();
        fDone = true;
        return ret;
    }
 
    void Add(std::vector<T> &&vChecks) {
        if (pqueue != nullptr) {
            pqueue->Add(std::move(vChecks));
        }
    }
 
    ~CCheckQueueControl() {
        if (!fDone) {
            Complete();
        }
        if (pqueue != nullptr) {
            LEAVE_CRITICAL_SECTION(pqueue->m_control_mutex);
        }
    }
};
 
#endif // BITCOIN_CHECKQUEUE_H