fix disk manager thread conflict

vesoft-inc · Jul 5, 2021 · f988991 · f988991
1 parent 8f38df5
commit f988991
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 3 deletions.
diff --git a/src/kvstore/DiskManager.cpp b/src/kvstore/DiskManager.cpp
@@ -70,6 +70,7 @@ StatusOr<std::string> DiskManager::path(GraphSpaceID spaceId, PartitionID partId
 void DiskManager::addPartToPath(GraphSpaceID spaceId,
                                 PartitionID partId,
                                 const std::string& path) {
+    std::lock_guard<std::mutex> lg(lock_);
     try {
         auto canonical = boost::filesystem::canonical(path);
         auto dataPath = canonical.parent_path().parent_path();
@@ -85,6 +86,7 @@ void DiskManager::addPartToPath(GraphSpaceID spaceId,
 void DiskManager::removePartFromPath(GraphSpaceID spaceId,
                                      PartitionID partId,
                                      const std::string& path) {
+    std::lock_guard<std::mutex> lg(lock_);
     try {
         auto canonical = boost::filesystem::canonical(path);
         auto dataPath = canonical.parent_path().parent_path();

diff --git a/src/kvstore/DiskManager.h b/src/kvstore/DiskManager.h
@@ -65,6 +65,9 @@ class DiskManager {
 
     // the index in dataPaths_ for a given space + part
     std::unordered_map<GraphSpaceID, std::unordered_map<PartitionID, size_t>> partIndex_;
+
+    // lock used to protect partPath_ and partIndex_
+    std::mutex lock_;
 };
 
 }  // namespace kvstore

diff --git a/src/kvstore/raftex/RaftPart.cpp b/src/kvstore/raftex/RaftPart.cpp
@@ -768,7 +768,7 @@ void RaftPart::appendLogsInternal(AppendLogsIterator iter, TermID termId) {
     } while (false);
 
     if (!checkAppendLogResult(res)) {
-        LOG(ERROR) << idStr_ << "Failed append logs";
+        LOG_EVERY_N(WARNING, 100) << idStr_ << "Failed to write wal";
         return;
     }
     // Step 2: Replicate to followers
@@ -805,7 +805,7 @@ void RaftPart::replicateLogs(folly::EventBase* eb,
     } while (false);
 
     if (!checkAppendLogResult(res)) {
-        LOG(ERROR) << idStr_ << "Replicate logs failed";
+        LOG(WARNING) << idStr_ << "replicateLogs failed because of not leader or term changed";
         return;
     }
 
@@ -911,7 +911,9 @@ void RaftPart::processAppendLogResponses(
         } while (false);
 
         if (!checkAppendLogResult(res)) {
-            LOG(ERROR) << idStr_ << "processAppendLogResponses failed!";
+            LOG(WARNING)
+                << idStr_
+                << "processAppendLogResponses failed because of not leader or term changed";
             return;
         }