Fix TSAN warnings in indexing_test (#5149)

.daily_canary

@@ -1,4 +1,4 @@
    ___     ___
   (- -)   (o o)   |     Y   &  +-
  (  V  ) z  O  z  O    +---=---'
-/--x-m- /--m-m---xXx--/--yY------
+/--x-m- /--m-m---xXx--/--yY-----

.threading_canary

@@ -1 +1 @@
-This looks like a "job" for Threading Canary!!!!!!
+This looks like a "job" for Threading Canary!!!!

include/ccf/indexing/strategies/seqnos_by_key_in_memory.h

@@ -69,6 +69,7 @@ namespace ccf::indexing::strategies
 
     std::optional<SeqNoCollection> get_all_write_txs(const typename M::Key& key)
     {
+      std::lock_guard<ccf::pal::Mutex> guard(current_txid_lock);
       return get_write_txs_in_range(key, 0, current_txid.seqno);
     }
   };

include/ccf/indexing/strategies/visit_each_entry_in_map.h

@@ -4,6 +4,7 @@
 
 #include "ccf/byte_vector.h"
 #include "ccf/indexing/strategy.h"
+#include "ccf/pal/locking.h"
 
 namespace ccf::indexing::strategies
 {
@@ -14,6 +15,10 @@ namespace ccf::indexing::strategies
   {
   protected:
     std::string map_name;
+
+    // Protect access to current_txid
+    ccf::pal::Mutex current_txid_lock;
+
     ccf::TxID current_txid = {};
 
     virtual void visit_entry(
@@ -34,6 +39,6 @@ namespace ccf::indexing::strategies
 
     nlohmann::json describe() override;
 
-    ccf::TxID get_indexed_watermark() const;
+    ccf::TxID get_indexed_watermark();
   };
 }

src/apps/external_executor/external_executor_indexing.cpp

@@ -24,7 +24,7 @@ namespace externalexecutor
 
       // We were already trying to fetch this. If it's finished fetching,
       // parse and store the result
-      const auto fetch_result = bucket_value->fetch_result;
+      const auto fetch_result = bucket_value->fetch_result.load();
       switch (fetch_result)
       {
         case (ccf::indexing::FetchResult::Fetching):
@@ -149,4 +149,4 @@ namespace externalexecutor
   {
     return impl_index->fetch_data(key);
   }
-}
+}

src/indexing/lfs_interface.h

@@ -20,13 +20,14 @@ namespace ccf::indexing
 
   struct FetchResult
   {
-    enum
+    enum FetchResultType
     {
       Fetching,
       Loaded,
       NotFound,
       Corrupt,
-    } fetch_result;
+    };
+    std::atomic<FetchResultType> fetch_result;
 
     LFSKey key;
 

src/indexing/strategies/seqnos_by_key_bucketed.cpp

@@ -33,15 +33,20 @@ namespace ccf::indexing::strategies
     using BucketValue = std::pair<FetchResultPtr, SeqNoCollection>;
     LRU<BucketKey, BucketValue> old_results;
 
+    // When results_access and current_txid_lock need to be
+    // taken at the same time, results_access should be first
     ccf::pal::Mutex results_access;
 
     std::string name;
 
     std::shared_ptr<AbstractLFSAccess> lfs_access;
+
+    ccf::pal::Mutex& current_txid_lock;
     ccf::TxID& current_txid;
 
     Impl(
       const std::string& name_,
+      ccf::pal::Mutex& current_txid_lock_,
       ccf::TxID& current_txid_,
       const std::shared_ptr<AbstractLFSAccess>& lfs_access_,
       size_t seqnos_per_bucket_,
@@ -50,6 +55,7 @@ namespace ccf::indexing::strategies
       old_results(max_buckets_),
       name(name_),
       lfs_access(lfs_access_),
+      current_txid_lock(current_txid_lock_),
       current_txid(current_txid_)
     {
       if (lfs_access == nullptr)
@@ -166,7 +172,8 @@ namespace ccf::indexing::strategies
           fmt::format("Range goes backwards: {} -> {}", from, to));
       }
 
-      if (to > current_txid.seqno)
+      if (std::lock_guard<ccf::pal::Mutex> guard(current_txid_lock);
+          to > current_txid.seqno)
       {
         // If the requested range hasn't been populated yet, indicate
         // that with nullopt
@@ -221,7 +228,7 @@ namespace ccf::indexing::strategies
           // parse and store the result
           if (bucket_value.first != nullptr)
           {
-            const auto fetch_result = bucket_value.first->fetch_result;
+            const auto fetch_result = bucket_value.first->fetch_result.load();
             switch (fetch_result)
             {
               case (FetchResult::Fetching):
@@ -266,7 +273,10 @@ namespace ccf::indexing::strategies
                 // for safety, and consistency with the simple indexing
                 // strategies currently used, this re-indexes everything from
                 // the start of time.
-                current_txid = {};
+                {
+                  std::lock_guard<ccf::pal::Mutex> guard(current_txid_lock);
+                  current_txid = {};
+                }
                 old_results.clear();
                 current_results.clear();
 
@@ -436,6 +446,7 @@ namespace ccf::indexing::strategies
 
     impl = std::make_shared<Impl>(
       get_name(),
+      current_txid_lock,
       current_txid,
       node_context.get_subsystem<AbstractLFSAccess>(),
       seqnos_per_bucket_,

src/indexing/strategies/visit_each_entry_in_map.cpp

@@ -25,11 +25,13 @@ namespace ccf::indexing::strategies
       visit_entry(tx_id, k, v);
       return true;
     });
+    std::lock_guard<ccf::pal::Mutex> guard(current_txid_lock);
     current_txid = tx_id;
   }
 
   std::optional<ccf::SeqNo> VisitEachEntryInMap::next_requested()
   {
+    std::lock_guard<ccf::pal::Mutex> guard(current_txid_lock);
     return current_txid.seqno + 1;
   }
 
@@ -41,8 +43,9 @@ namespace ccf::indexing::strategies
     return j;
   }
 
-  ccf::TxID VisitEachEntryInMap::get_indexed_watermark() const
+  ccf::TxID VisitEachEntryInMap::get_indexed_watermark()
   {
+    std::lock_guard<ccf::pal::Mutex> guard(current_txid_lock);
     return current_txid;
   }
 }

src/indexing/test/indexing.cpp

@@ -456,7 +456,7 @@ TEST_CASE_TEMPLATE(
 }
 
 using namespace std::chrono_literals;
-const auto max_multithread_run_time = 10s;
+const auto max_multithread_run_time = 100s;
 
 // Uses the real classes, and access + update them concurrently
 TEST_CASE(
@@ -957,4 +957,4 @@ TEST_CASE(
   ringbuffer_flusher.join();
   index_ticker.join();
   watchdog.join();
-}
+}

tsan_env_suppressions

@@ -8,10 +8,5 @@
 deadlock:*/store.h
 deadlock:*/untyped_map.h
 
-# For indexing_test
-race:*/include/ccf/indexing/strategies/seqnos_by_key_in_memory.h
-race:*/indexing/strategies/visit_each_entry_in_map.cpp
-race:*/indexing/strategies/seqnos_by_key_bucketed.cpp
-
 # For governance_test
 race:*/node/*rpc/*frontend.h