diff --git a/src/coins.cpp b/src/coins.cpp index 69ad264898e..7bb05f68c1a 100644 --- a/src/coins.cpp +++ b/src/coins.cpp @@ -368,6 +368,7 @@ void CCoinsViewCache::SanityCheck() const CCoinsViewCache::ResetGuard CoinsViewOverlay::StartFetching(const CBlock& block LIFETIMEBOUND) noexcept { + Assert(m_futures.empty()); Assert(m_inputs.empty()); Assert(m_input_head.load(std::memory_order_relaxed) == 0); Assert(m_input_tail == 0); @@ -383,9 +384,21 @@ CCoinsViewCache::ResetGuard CoinsViewOverlay::StartFetching(const CBlock& block } earlier_txids.emplace(tx->GetHash()); } - // Only process inputs if we have something to fetch. + // Only submit tasks if we have something to fetch. if (m_inputs.size()) { - while (ProcessInput()) {} + std::vector> tasks(workers_count, [this] { + while (ProcessInput()) {} + }); + if (auto futures{m_thread_pool->Submit(std::move(tasks))}) { + m_futures = std::move(*futures); + } else { + // Submit can fail if a shared owner of the thread pool outside of this class calls Stop() or + // Interrupt() on a different thread after we call WorkersCount() above. In that case parallel + // fetching will not make progress, so we clear the inputs to fall back to single threaded fetching. + LogWarning("Failed to submit prevout fetch tasks; falling back to single-threaded fetching for this block."); + m_inputs.clear(); + StopFetching(); // Assert nothing changed if we failed to start tasks. + } } } return CreateResetGuard(); diff --git a/src/coins.h b/src/coins.h index df519f76de4..5528bde0581 100644 --- a/src/coins.h +++ b/src/coins.h @@ -16,6 +16,7 @@ #include #include #include +#include #include #include @@ -24,6 +25,7 @@ #include #include +#include #include #include #include @@ -496,7 +498,7 @@ public: * If reallocate_cache is false, the cache will retain the same memory footprint * after flushing and should be destroyed to deallocate. */ - void Flush(bool reallocate_cache = true); + virtual void Flush(bool reallocate_cache = true); /** * Push the modifications applied to this cache to its base while retaining @@ -598,6 +600,7 @@ private: Assert(!other.ready.test(std::memory_order_relaxed)); } }; + //! Must only be mutated when m_futures is empty. Elements may be mutated when m_futures is not empty. std::vector m_inputs{}; /** @@ -619,9 +622,21 @@ private: return true; } - //! Clear fetching data. + //! Stop all worker threads and clear fetching data. + //! Calling this is idempotent, and may safely be called if not fetching. void StopFetching() noexcept { + if (m_futures.empty()) { + Assert(m_inputs.empty()); + Assert(m_input_head.load(std::memory_order_relaxed) == 0); + Assert(m_input_tail == 0); + return; + } + // Skip fetching the rest of the inputs by moving the head to the end. + m_input_head.store(m_inputs.size(), std::memory_order_relaxed); + // Wait for all threads to stop. + for (auto& future : m_futures) future.wait(); + m_futures.clear(); m_inputs.clear(); m_input_head.store(0, std::memory_order_relaxed); m_input_tail = 0; @@ -644,8 +659,9 @@ private: return base->PeekCoin(outpoint); } - //! Non-null. + //! Non-null. May have zero workers when input fetching is disabled. std::shared_ptr m_thread_pool; + std::vector> m_futures{}; protected: void Reset() noexcept override @@ -662,8 +678,22 @@ public: Assert(m_thread_pool); } + ~CoinsViewOverlay() noexcept override { StopFetching(); } + //! Start fetching inputs from block. [[nodiscard]] ResetGuard StartFetching(const CBlock& block LIFETIMEBOUND) noexcept; + + void Flush(bool reallocate_cache = true) override + { + if (!Assume(AllInputsConsumed())) { + LogWarning("Block %s input prevout prefetch queue was not fully consumed; inputs were accessed out of order, so prefetching degraded to serial lookups for this block.", GetBestBlock().ToString()); + } + StopFetching(); + CCoinsViewCache::Flush(reallocate_cache); + } + + //! Verify that all parallel fetched input prevouts have been consumed. + bool AllInputsConsumed() const noexcept { return m_input_tail == m_inputs.size(); } }; //! Utility function to add all of a transaction's outputs to a cache.