Merge bitcoin/bitcoin#32467: checkqueue: make the queue non-optional for CCheckQueueControl and drop legacy locking macro usage

fd29073 validation: clean up and clarify CheckInputScripts logic (Cory Fields)
1a37507 validation: use a lock for CCheckQueueControl (Cory Fields)
c3b0e6c validation: make CCheckQueueControl's CCheckQueue non-optional (Cory Fields)
4c8c90b validation: only create a CCheckQueueControl if it's actually going to be used (Cory Fields)
11fed83 threading: add LOCK_ARGS macro (Cory Fields)

Pull request description:

  As part of an effort to cleanup our threading primitives and add safe `SharedMutex`/`SharedLock` impls, I'd like to get rid of the last of our legacy `ENTER_CRITICAL_SECTION`/`LEAVE_CRITICAL_SECTION` usage. This, along with a follow-up [after fixing REVERSE_LOCK](bitcoin/bitcoin#32465) will allow us to do that.

  This replaces the old macros with an RAII lock, while simplifying `CCheckQueueControl`. It now requires a `CCheckQueue`, and optionality is handled externally. In the case of validation, it is wrapped in a `std::optional`.

  It also adds an `LOCK_ARGS` macro for `UniqueLock` initialization which may be helpful elsewhere.

ACKs for top commit:
  fjahr:
    re-ACK fd29073
  ryanofsky:
    Code review ACK fd29073, just removing assert since last review. Thanks for considering all the comments and feedback!
  TheCharlatan:
    Re-ACK fd29073

Tree-SHA512: 54b9db604ee1bda7d11bce1653a88d3dcbc4f525eed6a85abdd4d6409138674af4bb8b00afa4e0d3d29dadd045a3a39de253a45f0ef9c05f56cba1aac5b59303

Author: fanquake

src/bench/checkqueue.cpp

@@ -56,7 +56,7 @@ static void CCheckQueueSpeedPrevectorJob(benchmark::Bench& bench)
 
     bench.minEpochIterations(10).batch(BATCH_SIZE * BATCHES).unit("job").run([&] {
         // Make insecure_rand here so that each iteration is identical.
-        CCheckQueueControl<PrevectorJob> control(&queue);
+        CCheckQueueControl<PrevectorJob> control(queue);
         for (auto vChecks : vBatches) {
             control.Add(std::move(vChecks));
         }

src/checkqueue.h

@@ -205,46 +205,35 @@ class CCheckQueue
  * queue is finished before continuing.
  */
 template <typename T, typename R = std::remove_cvref_t<decltype(std::declval<T>()().value())>>
-class CCheckQueueControl
+class SCOPED_LOCKABLE CCheckQueueControl
 {
 private:
-    CCheckQueue<T, R> * const pqueue;
+    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> * 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);
-        }
-    }
+    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()
     {
-        if (pqueue == nullptr) return std::nullopt;
-        auto ret = pqueue->Complete();
+        auto ret = m_queue.Complete();
         fDone = true;
         return ret;
     }
 
     void Add(std::vector<T>&& vChecks)
     {
-        if (pqueue != nullptr) {
-            pqueue->Add(std::move(vChecks));
-        }
+        m_queue.Add(std::move(vChecks));
     }
 
-    ~CCheckQueueControl()
+    ~CCheckQueueControl() UNLOCK_FUNCTION()
     {
         if (!fDone)
             Complete();
-        if (pqueue != nullptr) {
-            LEAVE_CRITICAL_SECTION(pqueue->m_control_mutex);
-        }
     }
 };
 

src/sync.h

@@ -258,8 +258,9 @@ inline MutexType* MaybeCheckNotHeld(MutexType* m) LOCKS_EXCLUDED(m) LOCK_RETURNE
 #define LOCK2(cs1, cs2)                                               \
     UniqueLock criticalblock1(MaybeCheckNotHeld(cs1), #cs1, __FILE__, __LINE__); \
     UniqueLock criticalblock2(MaybeCheckNotHeld(cs2), #cs2, __FILE__, __LINE__)
-#define TRY_LOCK(cs, name) UniqueLock name(MaybeCheckNotHeld(cs), #cs, __FILE__, __LINE__, true)
-#define WAIT_LOCK(cs, name) UniqueLock name(MaybeCheckNotHeld(cs), #cs, __FILE__, __LINE__)
+#define LOCK_ARGS(cs) MaybeCheckNotHeld(cs), #cs, __FILE__, __LINE__
+#define TRY_LOCK(cs, name) UniqueLock name(LOCK_ARGS(cs), true)
+#define WAIT_LOCK(cs, name) UniqueLock name(LOCK_ARGS(cs))
 
 #define ENTER_CRITICAL_SECTION(cs)                            \
     {                                                         \

src/test/checkqueue_tests.cpp

@@ -165,7 +165,7 @@ void CheckQueueTest::Correct_Queue_range(std::vector<size_t> range)
     for (const size_t i : range) {
         size_t total = i;
         FakeCheckCheckCompletion::n_calls = 0;
-        CCheckQueueControl<FakeCheckCheckCompletion> control(small_queue.get());
+        CCheckQueueControl<FakeCheckCheckCompletion> control(*small_queue);
         while (total) {
             vChecks.clear();
             vChecks.resize(std::min<size_t>(total, m_rng.randrange(10)));
@@ -220,7 +220,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Catches_Failure)
 {
     auto fixed_queue = std::make_unique<Fixed_Queue>(QUEUE_BATCH_SIZE, SCRIPT_CHECK_THREADS);
     for (size_t i = 0; i < 1001; ++i) {
-        CCheckQueueControl<FixedCheck> control(fixed_queue.get());
+        CCheckQueueControl<FixedCheck> control(*fixed_queue);
         size_t remaining = i;
         while (remaining) {
             size_t r = m_rng.randrange(10);
@@ -246,7 +246,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Recovers_From_Failure)
     auto fail_queue = std::make_unique<Fixed_Queue>(QUEUE_BATCH_SIZE, SCRIPT_CHECK_THREADS);
     for (auto times = 0; times < 10; ++times) {
         for (const bool end_fails : {true, false}) {
-            CCheckQueueControl<FixedCheck> control(fail_queue.get());
+            CCheckQueueControl<FixedCheck> control(*fail_queue);
             {
                 std::vector<FixedCheck> vChecks;
                 vChecks.resize(100, FixedCheck(std::nullopt));
@@ -268,7 +268,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_UniqueCheck)
     size_t COUNT = 100000;
     size_t total = COUNT;
     {
-        CCheckQueueControl<UniqueCheck> control(queue.get());
+        CCheckQueueControl<UniqueCheck> control(*queue);
         while (total) {
             size_t r = m_rng.randrange(10);
             std::vector<UniqueCheck> vChecks;
@@ -300,7 +300,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Memory)
     for (size_t i = 0; i < 1000; ++i) {
         size_t total = i;
         {
-            CCheckQueueControl<MemoryCheck> control(queue.get());
+            CCheckQueueControl<MemoryCheck> control(*queue);
             while (total) {
                 size_t r = m_rng.randrange(10);
                 std::vector<MemoryCheck> vChecks;
@@ -324,7 +324,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_FrozenCleanup)
     auto queue = std::make_unique<FrozenCleanup_Queue>(QUEUE_BATCH_SIZE, SCRIPT_CHECK_THREADS);
     bool fails = false;
     std::thread t0([&]() {
-        CCheckQueueControl<FrozenCleanupCheck> control(queue.get());
+        CCheckQueueControl<FrozenCleanupCheck> control(*queue);
         std::vector<FrozenCleanupCheck> vChecks(1);
         control.Add(std::move(vChecks));
         auto result = control.Complete(); // Hangs here
@@ -364,7 +364,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueueControl_Locks)
         for (size_t i = 0; i < 3; ++i) {
             tg.emplace_back(
                     [&]{
-                    CCheckQueueControl<FakeCheck> control(queue.get());
+                    CCheckQueueControl<FakeCheck> control(*queue);
                     // While sleeping, no other thread should execute to this point
                     auto observed = ++nThreads;
                     UninterruptibleSleep(std::chrono::milliseconds{10});
@@ -387,7 +387,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueueControl_Locks)
         {
             std::unique_lock<std::mutex> l(m);
             tg.emplace_back([&]{
-                    CCheckQueueControl<FakeCheck> control(queue.get());
+                    CCheckQueueControl<FakeCheck> control(*queue);
                     std::unique_lock<std::mutex> ll(m);
                     has_lock = true;
                     cv.notify_one();

src/test/fuzz/checkqueue.cpp

@@ -49,7 +49,7 @@ FUZZ_TARGET(checkqueue)
         (void)check_queue_1.Complete();
     }
 
-    CCheckQueueControl<DumbCheck> check_queue_control{&check_queue_2};
+    CCheckQueueControl<DumbCheck> check_queue_control{check_queue_2};
     if (fuzzed_data_provider.ConsumeBool()) {
         check_queue_control.Add(std::move(checks_2));
     }

src/test/transaction_tests.cpp

@@ -568,7 +568,7 @@ BOOST_AUTO_TEST_CASE(test_big_witness_transaction)
     // check all inputs concurrently, with the cache
     PrecomputedTransactionData txdata(tx);
     CCheckQueue<CScriptCheck> scriptcheckqueue(/*batch_size=*/128, /*worker_threads_num=*/20);
-    CCheckQueueControl<CScriptCheck> control(&scriptcheckqueue);
+    CCheckQueueControl<CScriptCheck> control(scriptcheckqueue);
 
     std::vector<Coin> coins;
     for(uint32_t i = 0; i < mtx.vin.size(); i++) {

src/validation.cpp

@@ -2421,7 +2421,6 @@ bool Chainstate::ConnectBlock(const CBlock& block, BlockValidationState& state,
 
     uint256 block_hash{block.GetHash()};
     assert(*pindex->phashBlock == block_hash);
-    const bool parallel_script_checks{m_chainman.GetCheckQueue().HasThreads()};
 
     const auto time_start{SteadyClock::now()};
     const CChainParams& params{m_chainman.GetParams()};
@@ -2611,7 +2610,9 @@ bool Chainstate::ConnectBlock(const CBlock& block, BlockValidationState& state,
     // in multiple threads). Preallocate the vector size so a new allocation
     // doesn't invalidate pointers into the vector, and keep txsdata in scope
     // for as long as `control`.
-    CCheckQueueControl<CScriptCheck> control(fScriptChecks && parallel_script_checks ? &m_chainman.GetCheckQueue() : nullptr);
+    std::optional<CCheckQueueControl<CScriptCheck>> control;
+    if (auto& queue = m_chainman.GetCheckQueue(); queue.HasThreads() && fScriptChecks) control.emplace(queue);
+
     std::vector<PrecomputedTransactionData> txsdata(block.vtx.size());
 
     std::vector<int> prevheights;
@@ -2669,18 +2670,26 @@ bool Chainstate::ConnectBlock(const CBlock& block, BlockValidationState& state,
             break;
         }
 
-        if (!tx.IsCoinBase())
+        if (!tx.IsCoinBase() && fScriptChecks)
         {
-            std::vector<CScriptCheck> vChecks;
             bool fCacheResults = fJustCheck; /* Don't cache results if we're actually connecting blocks (still consult the cache, though) */
+            bool tx_ok;
             TxValidationState tx_state;
-            if (fScriptChecks && !CheckInputScripts(tx, tx_state, view, flags, fCacheResults, fCacheResults, txsdata[i], m_chainman.m_validation_cache, parallel_script_checks ? &vChecks : nullptr)) {
+            // If CheckInputScripts is called with a pointer to a checks vector, the resulting checks are appended to it. In that case
+            // they need to be added to control which runs them asynchronously. Otherwise, CheckInputScripts runs the checks before returning.
+            if (control) {
+                std::vector<CScriptCheck> vChecks;
+                tx_ok = CheckInputScripts(tx, tx_state, view, flags, fCacheResults, fCacheResults, txsdata[i], m_chainman.m_validation_cache, &vChecks);
+                if (tx_ok) control->Add(std::move(vChecks));
+            } else {
+                tx_ok = CheckInputScripts(tx, tx_state, view, flags, fCacheResults, fCacheResults, txsdata[i], m_chainman.m_validation_cache);
+            }
+            if (!tx_ok) {
                 // Any transaction validation failure in ConnectBlock is a block consensus failure
                 state.Invalid(BlockValidationResult::BLOCK_CONSENSUS,
                               tx_state.GetRejectReason(), tx_state.GetDebugMessage());
                 break;
             }
-            control.Add(std::move(vChecks));
         }
 
         CTxUndo undoDummy;
@@ -2702,10 +2711,11 @@ bool Chainstate::ConnectBlock(const CBlock& block, BlockValidationState& state,
         state.Invalid(BlockValidationResult::BLOCK_CONSENSUS, "bad-cb-amount",
                       strprintf("coinbase pays too much (actual=%d vs limit=%d)", block.vtx[0]->GetValueOut(), blockReward));
     }
-
-    auto parallel_result = control.Complete();
-    if (parallel_result.has_value() && state.IsValid()) {
-        state.Invalid(BlockValidationResult::BLOCK_CONSENSUS, strprintf("mandatory-script-verify-flag-failed (%s)", ScriptErrorString(parallel_result->first)), parallel_result->second);
+    if (control) {
+        auto parallel_result = control->Complete();
+        if (parallel_result.has_value() && state.IsValid()) {
+            state.Invalid(BlockValidationResult::BLOCK_CONSENSUS, strprintf("mandatory-script-verify-flag-failed (%s)", ScriptErrorString(parallel_result->first)), parallel_result->second);
+        }
     }
     if (!state.IsValid()) {
         LogInfo("Block validation error: %s", state.ToString());