diff --git a/src/bls/bls_worker.cpp b/src/bls/bls_worker.cpp index eade0642bc44..554f060a0767 100644 --- a/src/bls/bls_worker.cpp +++ b/src/bls/bls_worker.cpp @@ -69,6 +69,11 @@ void CBLSWorker::Stop() workerPool.stop(true); } +void CBLSWorker::PushJob(std::function job) +{ + workerPool.push([job = std::move(job)](int) { job(); }); +} + #ifndef BUILD_BITCOIN_INTERNAL bool CBLSWorker::GenerateContributions(int quorumThreshold, Span ids, BLSVerificationVectorPtr& vvecRet, std::vector& skSharesRet) { diff --git a/src/bls/bls_worker.h b/src/bls/bls_worker.h index 369058a9591a..7e57df6e964c 100644 --- a/src/bls/bls_worker.h +++ b/src/bls/bls_worker.h @@ -56,6 +56,7 @@ class CBLSWorker void Start(int16_t worker_count); void Stop(); + void PushJob(std::function job); #ifndef BUILD_BITCOIN_INTERNAL bool GenerateContributions(int threshold, Span ids, BLSVerificationVectorPtr& vvecRet, std::vector& skSharesRet); diff --git a/src/init.cpp b/src/init.cpp index 63318d925702..d78045f653cc 100644 --- a/src/init.cpp +++ b/src/init.cpp @@ -2562,6 +2562,9 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info) connOptions.m_peer_connect_timeout = peer_connect_timeout; connOptions.socketEventsMode = ::g_socket_events_mode; connOptions.m_active_masternode = node.active_ctx != nullptr; + // wallets may mix with CoinJoin, which connects to the mixing masternode + connOptions.m_masternode_connections = node.active_ctx != nullptr || quorums_watch || + node.wallet_loader != nullptr; // Port to bind to if `-bind=addr` is provided without a `:port` suffix. const uint16_t default_bind_port = diff --git a/src/llmq/context.cpp b/src/llmq/context.cpp index b961a0a7ec89..c94cf1cfa106 100644 --- a/src/llmq/context.cpp +++ b/src/llmq/context.cpp @@ -28,5 +28,6 @@ LLMQContext::LLMQContext(CDeterministicMNManager& dmnman, CEvoDB& evo_db, Chains LLMQContext::~LLMQContext() { + qman->InterruptWarming(); bls_worker->Stop(); } diff --git a/src/llmq/quorumsman.cpp b/src/llmq/quorumsman.cpp index 6dd0fc53a733..7ba6eb2479f7 100644 --- a/src/llmq/quorumsman.cpp +++ b/src/llmq/quorumsman.cpp @@ -22,7 +22,6 @@ #include #include #include -#include #include #include @@ -40,18 +39,10 @@ CQuorumManager::CQuorumManager(CBLSWorker& _blsWorker, CDeterministicMNManager& db{util::MakeDbWrapper({db_params.path / "llmq" / "quorumdb", db_params.memory, db_params.wipe, /*cache_size=*/1 << 20})} { mapQuorumsCache.Init(m_chainman.GetConsensus(), /*limit_by_connections=*/false); - m_cache_interrupt.reset(); - m_cache_thread = std::thread(&util::TraceThread, "q-cache", [this] { CacheWarmingThreadMain(); }); MigrateOldQuorumDB(_evoDb); } -CQuorumManager::~CQuorumManager() -{ - if (m_cache_thread.joinable()) { - m_cache_interrupt(); - m_cache_thread.join(); - } -} +CQuorumManager::~CQuorumManager() = default; bool CQuorumManager::GetEncryptedContributions(Consensus::LLMQType llmq_type, const CBlockIndex* block_index, const std::vector& valid_members, const uint256& protx_hash, @@ -505,50 +496,29 @@ void CQuorumManager::WriteContributions(const CQuorumPtr& quorum) const quorum->WriteContributions(*db); } -void CQuorumManager::CacheWarmingThreadMain() const +void CQuorumManager::QueueQuorumForWarming(CQuorumCPtr pQuorum) const { - while (!m_cache_interrupt) { - CQuorumCPtr pQuorum; - { - LOCK(m_cache_cs); - if (!m_cache_queue.empty()) { - pQuorum = std::move(m_cache_queue.front()); - m_cache_queue.pop_front(); - }; - } - - if (!pQuorum) { - m_cache_interrupt.sleep_for(std::chrono::milliseconds(100)); - continue; - } - + if (!pQuorum->HasVerificationVector()) { + return; + } + // The job may reference this manager: ~LLMQContext stops the worker pool before destroying it. + blsWorker.PushJob([this, pQuorum = std::move(pQuorum)]() { cxxtimer::Timer t(true); - LogPrint(BCLog::LLMQ, "CQuorumManager::%s -- type=%d height=%d hash=%s start\n", __func__, + LogPrint(BCLog::LLMQ, "CQuorumManager::QueueQuorumForWarming -- type=%d height=%d hash=%s start\n", std23::to_underlying(pQuorum->params.type), pQuorum->m_quorum_base_block_index->nHeight, pQuorum->m_quorum_base_block_index->GetBlockHash().ToString()); - - // when then later some other thread tries to get keys, it will be much faster for (const auto i : util::irange(pQuorum->members.size())) { - if (m_cache_interrupt) { - break; + if (m_warming_interrupted) { + return; } if (pQuorum->qc->validMembers[i]) { pQuorum->GetPubKeyShare(i); } } - - LogPrint(BCLog::LLMQ, "CQuorumManager::%s -- type=%d height=%d hash=%s done. time=%d\n", __func__, + LogPrint(BCLog::LLMQ, "CQuorumManager::QueueQuorumForWarming -- type=%d height=%d hash=%s done. time=%d\n", std23::to_underlying(pQuorum->params.type), pQuorum->m_quorum_base_block_index->nHeight, pQuorum->m_quorum_base_block_index->GetBlockHash().ToString(), t.count()); - } -} - -void CQuorumManager::QueueQuorumForWarming(CQuorumCPtr pQuorum) const -{ - if (pQuorum->HasVerificationVector()) { - LOCK(m_cache_cs); - m_cache_queue.push_back(std::move(pQuorum)); - } + }); } // TODO: remove in v23 diff --git a/src/llmq/quorumsman.h b/src/llmq/quorumsman.h index f61f30b4c6b1..ac17caf45893 100644 --- a/src/llmq/quorumsman.h +++ b/src/llmq/quorumsman.h @@ -15,14 +15,12 @@ #include #include -#include #include -#include +#include #include #include -#include class CBLSSignature; class CBLSWorker; @@ -96,10 +94,7 @@ class CQuorumManager final mutable Uint256LruHashMap quorumBaseBlockIndexCache GUARDED_BY(cs_quorumBaseBlockIndexCache); - mutable Mutex m_cache_cs; - mutable std::deque m_cache_queue GUARDED_BY(m_cache_cs); - mutable CThreadInterrupt m_cache_interrupt; - mutable std::thread m_cache_thread; + std::atomic m_warming_interrupted{false}; public: CQuorumManager() = delete; @@ -135,20 +130,20 @@ class CQuorumManager final // all these methods will lock cs_main for a short period of time CQuorumCPtr GetQuorum(Consensus::LLMQType llmqType, const uint256& quorumHash) const - EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps, !m_cache_cs); + EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps); CQuorumCPtr GetQuorum(Consensus::LLMQType llmqType, const uint256& quorumHash, const CChain& chain) const - EXCLUSIVE_LOCKS_REQUIRED(::cs_main, !cs_db, !m_cs_maps, !m_cache_cs); + EXCLUSIVE_LOCKS_REQUIRED(::cs_main, !cs_db, !m_cs_maps); std::vector ScanQuorums(Consensus::LLMQType llmqType, size_t nCountRequested) const - EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps, !m_cache_cs); + EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps); // this one is cs_main-free std::vector ScanQuorums(Consensus::LLMQType llmqType, gsl::not_null pindexStart, size_t nCountRequested) const - EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps, !m_cache_cs); + EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps); std::vector ScanQuorums(Consensus::LLMQType llmqType, gsl::not_null pindexStart, size_t nCountRequested, const CChain& chain) const - EXCLUSIVE_LOCKS_REQUIRED(::cs_main, !cs_db, !m_cs_maps, !m_cache_cs); + EXCLUSIVE_LOCKS_REQUIRED(::cs_main, !cs_db, !m_cs_maps); bool IsMasternode() const; bool IsWatching() const; @@ -175,29 +170,30 @@ class CQuorumManager final CQuorumPtr GetCachedMutableQuorum(Consensus::LLMQType llmqType, const uint256& quorumHash) const EXCLUSIVE_LOCKS_REQUIRED(!m_cs_maps); void WriteContributions(const CQuorumPtr& quorum) const EXCLUSIVE_LOCKS_REQUIRED(!cs_db); - void QueueQuorumForWarming(CQuorumCPtr pQuorum) const EXCLUSIVE_LOCKS_REQUIRED(!m_cache_cs); + void QueueQuorumForWarming(CQuorumCPtr pQuorum) const; + /** Make queued and running warming jobs return; call before stopping the BLS worker pool */ + void InterruptWarming() { m_warming_interrupted = true; } private: // all private methods here are cs_main-free std::vector ScanQuorums(Consensus::LLMQType llmqType, gsl::not_null pindexStart, size_t nCountRequested, const CChain* chain) const - EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps, !m_cache_cs); + EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps); bool BuildQuorumContributions(const CFinalCommitmentPtr& fqc, const std::shared_ptr& quorum) const; CQuorumPtr BuildQuorumFromCommitment(Consensus::LLMQType llmqType, gsl::not_null pQuorumBaseBlockIndex, bool populate_cache) const - EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps, !m_cache_cs); + EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps); CQuorumCPtr GetQuorum(Consensus::LLMQType llmqType, gsl::not_null pindex, bool populate_cache = true) const - EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps, !m_cache_cs); + EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps); CQuorumCPtr GetQuorum(Consensus::LLMQType llmqType, gsl::not_null pindex, const CChain& chain, bool populate_cache = true) const - EXCLUSIVE_LOCKS_REQUIRED(::cs_main, !cs_db, !m_cs_maps, !m_cache_cs); + EXCLUSIVE_LOCKS_REQUIRED(::cs_main, !cs_db, !m_cs_maps); - void CacheWarmingThreadMain() const EXCLUSIVE_LOCKS_REQUIRED(!m_cache_cs); void MigrateOldQuorumDB(CEvoDB& evoDb) const EXCLUSIVE_LOCKS_REQUIRED(!cs_db); }; diff --git a/src/net.cpp b/src/net.cpp index d96058c96db0..094d7549caf3 100644 --- a/src/net.cpp +++ b/src/net.cpp @@ -4093,10 +4093,11 @@ bool CConnman::Start(CDeterministicMNManager& dmnman, CMasternodeMetaMan& mn_met [this, connect = connOptions.m_specified_outgoing, &dmnman] { ThreadOpenConnections(connect, dmnman); }); } - // Initiate masternode connections - threadOpenMasternodeConnections = std::thread(&util::TraceThread, "mncon", [this, &dmnman, &mn_metaman, &mn_sync] { - ThreadOpenMasternodeConnections(dmnman, mn_metaman, mn_sync); - }); + if (m_masternode_connections) { + threadOpenMasternodeConnections = std::thread(&util::TraceThread, "mncon", [this, &dmnman, &mn_metaman, &mn_sync] { + ThreadOpenMasternodeConnections(dmnman, mn_metaman, mn_sync); + }); + } // Process messages threadMessageHandler = std::thread(&util::TraceThread, "msghand", [this] { ThreadMessageHandler(); }); diff --git a/src/net.h b/src/net.h index 33a43e579d90..7776cb1f72d3 100644 --- a/src/net.h +++ b/src/net.h @@ -1245,6 +1245,9 @@ friend class CNode; SocketEventsMode socketEventsMode = SocketEventsMode::Select; bool m_i2p_accept_incoming; bool m_active_masternode = false; + //! Run the thread that opens connections to masternodes; only masternodes, quorum watchers and + //! CoinJoin mixing ever request them. + bool m_masternode_connections = true; }; void Init(const Options& connOptions) EXCLUSIVE_LOCKS_REQUIRED(!m_added_nodes_mutex, !m_total_bytes_sent_mutex) @@ -1283,6 +1286,7 @@ friend class CNode; socketEventsMode = connOptions.socketEventsMode; m_onion_binds = connOptions.onion_binds; m_active_masternode = connOptions.m_active_masternode; + m_masternode_connections = connOptions.m_masternode_connections; } CConnman(uint64_t seed0, uint64_t seed1, AddrMan& addrman, const NetGroupManager& netgroupman, @@ -1928,6 +1932,7 @@ friend class CNode; /** Flag for activating masternode mode */ bool m_active_masternode{false}; + bool m_masternode_connections{true}; SocketEventsMode socketEventsMode; std::unique_ptr m_edge_trig_events{nullptr}; diff --git a/test/functional/test_framework/test_node.py b/test/functional/test_framework/test_node.py index 1e5351de6986..eb9be6c7eb2c 100755 --- a/test/functional/test_framework/test_node.py +++ b/test/functional/test_framework/test_node.py @@ -114,6 +114,13 @@ def __init__(self, i, datadir, extra_args_from_options, *, chain, rpchost, timew "-debugexclude=leveldb", "-debugexclude=rand", "-uacomment=testnode%d" % i, # required for subversion uniqueness across peers + # Two threads keep the check queues and the RPC server concurrent; more only adds + # threads, and every thread costs ~0.9 MB of thread-local storage (BLS context). + "-par=2", + "-rpcthreads=2", + # The signature and script execution caches are allocated and zero-filled at startup + # whether or not anything is ever cached; 1 MiB still leaves 16384 entries each. + "-maxsigcachesize=1", ] if self.mocktime != 0: self.args.append(f"-mocktime={mocktime}") @@ -138,6 +145,8 @@ def __init__(self, i, datadir, extra_args_from_options, *, chain, rpchost, timew self.args.append("-logsourcelocations") if self.version_is_at_least(22010000): self.args.append("-loglevel=trace") + if self.version_is_at_least(23000000): + self.args.append("-parbls=2") # Default behavior from global -v2transport flag is added to args to persist it over restarts. # May be overwritten in individual tests, using extra_args.