Fix race conditions in shard

Builds/CMake/RippledCore.cmake

@@ -157,6 +157,7 @@ install (
     src/ripple/basics/safe_cast.h
     src/ripple/basics/Slice.h
     src/ripple/basics/StringUtilities.h
+    src/ripple/basics/ThreadSafetyAnalysis.h
     src/ripple/basics/ToString.h
     src/ripple/basics/UnorderedContainers.h
     src/ripple/basics/XRPAmount.h

CMakeLists.txt

@@ -21,6 +21,12 @@ if(Git_FOUND)
     endif()
 endif() #git
 
+if (thread_safety_analysis)
+  add_compile_options(-Wthread-safety -D_LIBCPP_ENABLE_THREAD_SAFETY_ANNOTATIONS -DRIPPLE_ENABLE_THREAD_SAFETY_ANNOTATIONS)
+  add_compile_options("-stdlib=libc++")
+  add_link_options("-stdlib=libc++")
+endif()
+
 list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_SOURCE_DIR}/Builds/CMake")
 list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_SOURCE_DIR}/Builds/CMake/deps")
 

src/ripple/basics/ThreadSafetyAnalysis.h

@@ -0,0 +1,63 @@
+#ifndef RIPPLE_BASICS_THREAD_SAFTY_ANALYSIS_H_INCLUDED
+#define RIPPLE_BASICS_THREAD_SAFTY_ANALYSIS_H_INCLUDED
+
+#ifdef RIPPLE_ENABLE_THREAD_SAFETY_ANNOTATIONS
+#define THREAD_ANNOTATION_ATTRIBUTE__(x) __attribute__((x))
+#else
+#define THREAD_ANNOTATION_ATTRIBUTE__(x)  // no-op
+#endif
+
+#define CAPABILITY(x) THREAD_ANNOTATION_ATTRIBUTE__(capability(x))
+
+#define SCOPED_CAPABILITY THREAD_ANNOTATION_ATTRIBUTE__(scoped_lockable)
+
+#define GUARDED_BY(x) THREAD_ANNOTATION_ATTRIBUTE__(guarded_by(x))
+
+#define PT_GUARDED_BY(x) THREAD_ANNOTATION_ATTRIBUTE__(pt_guarded_by(x))
+
+#define ACQUIRED_BEFORE(...) \
+    THREAD_ANNOTATION_ATTRIBUTE__(acquired_before(__VA_ARGS__))
+
+#define ACQUIRED_AFTER(...) \
+    THREAD_ANNOTATION_ATTRIBUTE__(acquired_after(__VA_ARGS__))
+
+#define REQUIRES(...) \
+    THREAD_ANNOTATION_ATTRIBUTE__(requires_capability(__VA_ARGS__))
+
+#define REQUIRES_SHARED(...) \
+    THREAD_ANNOTATION_ATTRIBUTE__(requires_shared_capability(__VA_ARGS__))
+
+#define ACQUIRE(...) \
+    THREAD_ANNOTATION_ATTRIBUTE__(acquire_capability(__VA_ARGS__))
+
+#define ACQUIRE_SHARED(...) \
+    THREAD_ANNOTATION_ATTRIBUTE__(acquire_shared_capability(__VA_ARGS__))
+
+#define RELEASE(...) \
+    THREAD_ANNOTATION_ATTRIBUTE__(release_capability(__VA_ARGS__))
+
+#define RELEASE_SHARED(...) \
+    THREAD_ANNOTATION_ATTRIBUTE__(release_shared_capability(__VA_ARGS__))
+
+#define RELEASE_GENERIC(...) \
+    THREAD_ANNOTATION_ATTRIBUTE__(release_generic_capability(__VA_ARGS__))
+
+#define TRY_ACQUIRE(...) \
+    THREAD_ANNOTATION_ATTRIBUTE__(try_acquire_capability(__VA_ARGS__))
+
+#define TRY_ACQUIRE_SHARED(...) \
+    THREAD_ANNOTATION_ATTRIBUTE__(try_acquire_shared_capability(__VA_ARGS__))
+
+#define EXCLUDES(...) THREAD_ANNOTATION_ATTRIBUTE__(locks_excluded(__VA_ARGS__))
+
+#define ASSERT_CAPABILITY(x) THREAD_ANNOTATION_ATTRIBUTE__(assert_capability(x))
+
+#define ASSERT_SHARED_CAPABILITY(x) \
+    THREAD_ANNOTATION_ATTRIBUTE__(assert_shared_capability(x))
+
+#define RETURN_CAPABILITY(x) THREAD_ANNOTATION_ATTRIBUTE__(lock_returned(x))
+
+#define NO_THREAD_SAFETY_ANALYSIS \
+    THREAD_ANNOTATION_ATTRIBUTE__(no_thread_safety_analysis)
+
+#endif

src/ripple/nodestore/impl/Shard.cpp

@@ -223,6 +223,7 @@ Shard::storeNodeObject(std::shared_ptr<NodeObject> const& nodeObject)
 
     try
     {
+        std::lock_guard lock(mutex_);
         backend_->store(nodeObject);
     }
     catch (std::exception const& e)
@@ -249,6 +250,7 @@ Shard::fetchNodeObject(uint256 const& hash, FetchReport& fetchReport)
     Status status;
     try
     {
+        std::lock_guard lock(mutex_);
         status = backend_->fetch(hash.data(), &nodeObject);
     }
     catch (std::exception const& e)
@@ -331,6 +333,7 @@ Shard::storeLedger(
 
         try
         {
+            std::lock_guard lock(mutex_);
             backend_->storeBatch(batch);
         }
         catch (std::exception const& e)
@@ -538,6 +541,7 @@ Shard::getWriteLoad()
     auto const scopedCount{makeBackendCount()};
     if (!scopedCount)
         return 0;
+    std::lock_guard lock(mutex_);
     return backend_->getWriteLoad();
 }
 
@@ -572,6 +576,8 @@ Shard::finalize(bool writeSQLite, std::optional<uint256> const& referenceHash)
 
     try
     {
+        std::lock_guard lock(mutex_);
+
         state_ = ShardState::finalizing;
         progress_ = 0;
 
@@ -759,8 +765,11 @@ Shard::finalize(bool writeSQLite, std::optional<uint256> const& referenceHash)
 
     try
     {
-        // Store final key's value, may already be stored
-        backend_->store(nodeObject);
+        {
+            // Store final key's value, may already be stored
+            std::lock_guard lock(mutex_);
+            backend_->store(nodeObject);
+        }
 
         // Do not allow all other threads work with the shard
         busy_ = true;
@@ -819,7 +828,7 @@ Shard::open(std::lock_guard<std::mutex> const& lock)
     using namespace boost::filesystem;
     Config const& config{app_.config()};
     auto preexist{false};
-    auto fail = [this, &preexist](std::string const& msg) {
+    auto fail = [this, &preexist](std::string const& msg) REQUIRES(mutex_) {
         backend_->close();
         lgrSQLiteDB_.reset();
         txSQLiteDB_.reset();
@@ -837,7 +846,7 @@ Shard::open(std::lock_guard<std::mutex> const& lock)
         }
         return false;
     };
-    auto createAcquireInfo = [this, &config]() {
+    auto createAcquireInfo = [this, &config]() REQUIRES(mutex_) {
         DatabaseCon::Setup setup;
         setup.startUp = config.standalone() ? config.LOAD : config.START_UP;
         setup.standAlone = config.standalone();
@@ -1024,6 +1033,8 @@ Shard::storeSQLite(std::shared_ptr<Ledger const> const& ledger)
 
     try
     {
+        std::lock_guard lock(mutex_);
+
         auto res = updateLedgerDBs(
             *txSQLiteDB_->checkoutDb(),
             *lgrSQLiteDB_->checkoutDb(),
@@ -1186,6 +1197,7 @@ Shard::verifyFetch(uint256 const& hash) const
 
     try
     {
+        std::lock_guard lock(mutex_);
         switch (backend_->fetch(hash.data(), &nodeObject))
         {
             case ok:

src/ripple/nodestore/impl/Shard.h

@@ -26,6 +26,7 @@
 #include <ripple/basics/KeyCache.h>
 #include <ripple/basics/MathUtilities.h>
 #include <ripple/basics/RangeSet.h>
+#include <ripple/basics/ThreadSafetyAnalysis.h>
 #include <ripple/core/DatabaseCon.h>
 #include <ripple/nodestore/NodeObject.h>
 #include <ripple/nodestore/Scheduler.h>
@@ -210,6 +211,7 @@ class Shard final
     std::string
     getStoredSeqs()
     {
+        std::lock_guard lock(mutex_);
         if (!acquireInfo_)
             return "";
 
@@ -316,29 +318,30 @@ class Shard final
     boost::filesystem::path const dir_;
 
     // Storage space utilized by the shard
-    std::uint64_t fileSz_{0};
+    GUARDED_BY(mutex_) std::uint64_t fileSz_{0};
 
     // Number of file descriptors required by the shard
-    std::uint32_t fdRequired_{0};
+    GUARDED_BY(mutex_) std::uint32_t fdRequired_{0};
 
     // NuDB key/value store for node objects
-    std::unique_ptr<Backend> backend_;
+    std::unique_ptr<Backend> backend_ GUARDED_BY(mutex_);
 
     std::atomic<std::uint32_t> backendCount_{0};
 
     // Ledger SQLite database used for indexes
-    std::unique_ptr<DatabaseCon> lgrSQLiteDB_;
+    std::unique_ptr<DatabaseCon> lgrSQLiteDB_ GUARDED_BY(mutex_);
 
     // Transaction SQLite database used for indexes
-    std::unique_ptr<DatabaseCon> txSQLiteDB_;
+    std::unique_ptr<DatabaseCon> txSQLiteDB_ GUARDED_BY(mutex_);
 
     // Tracking information used only when acquiring a shard from the network.
     // If the shard is finalized, this member will be null.
-    std::unique_ptr<AcquireInfo> acquireInfo_;
+    std::unique_ptr<AcquireInfo> acquireInfo_ GUARDED_BY(mutex_);
+    ;
 
     // Older shard without an acquire database or final key
     // Eventually there will be no need for this and should be removed
-    bool legacy_{false};
+    GUARDED_BY(mutex_) bool legacy_{false};
 
     // Determines if the shard needs to stop processing for shutdown
     std::atomic<bool> stop_{false};
@@ -356,16 +359,17 @@ class Shard final
     std::atomic<bool> removeOnDestroy_{false};
 
     // The time of the last access of a shard with a finalized state
-    std::chrono::steady_clock::time_point lastAccess_;
+    std::chrono::steady_clock::time_point lastAccess_ GUARDED_BY(mutex_);
+    ;
 
     // Open shard databases
     [[nodiscard]] bool
-    open(std::lock_guard<std::mutex> const& lock);
+    open(std::lock_guard<std::mutex> const& lock) REQUIRES(mutex_);
 
     // Open/Create SQLite databases
     // Lock over mutex_ required
     [[nodiscard]] bool
-    initSQLite(std::lock_guard<std::mutex> const&);
+    initSQLite(std::lock_guard<std::mutex> const&) REQUIRES(mutex_);
 
     // Write SQLite entries for this ledger
     [[nodiscard]] bool
@@ -374,7 +378,7 @@ class Shard final
     // Set storage and file descriptor usage stats
     // Lock over mutex_ required
     void
-    setFileStats(std::lock_guard<std::mutex> const&);
+    setFileStats(std::lock_guard<std::mutex> const&) REQUIRES(mutex_);
 
     // Verify this ledger by walking its SHAMaps and verifying its Merkle trees
     // Every node object verified will be stored in the deterministic shard