fix deadlock in TimeoutFlush (#2158)

* fix deadlock in TimeoutFlush

Author: henryzhx8

Diff for: core/collection_pipeline/CollectionPipeline.cpp

@@ -359,7 +359,7 @@ bool CollectionPipeline::Init(CollectionConfig&& config) {
 }
 
 void CollectionPipeline::Start() {
-    // #ifndef APSARA_UNIT_TEST_MAIN
+    TimeoutFlushManager::GetInstance()->RegisterFlushers(mName, mFlushers);
     //  TODO: 应该保证指定时间内返回，如果无法返回，将配置放入startDisabled里
     for (const auto& flusher : mFlushers) {
         flusher->Start();
@@ -381,7 +381,6 @@ void CollectionPipeline::Start() {
 
     SET_GAUGE(mStartTime,
               chrono::duration_cast<chrono::seconds>(chrono::system_clock::now().time_since_epoch()).count());
-    // #endif
     LOG_INFO(sLogger, ("pipeline start", "succeeded")("config", mName));
 }
 
@@ -439,7 +438,7 @@ bool CollectionPipeline::FlushBatch() {
     for (auto& flusher : mFlushers) {
         allSucceeded = flusher->FlushAll() && allSucceeded;
     }
-    TimeoutFlushManager::GetInstance()->ClearRecords(mName);
+    TimeoutFlushManager::GetInstance()->UnregisterFlushers(mName, mFlushers);
     return allSucceeded;
 }
 

Diff for: core/collection_pipeline/batch/TimeoutFlushManager.cpp

@@ -20,7 +20,7 @@ namespace logtail {
 
 void TimeoutFlushManager::UpdateRecord(
     const string& config, size_t index, size_t key, uint32_t timeoutSecs, Flusher* f) {
-    lock_guard<recursive_mutex> lock(mMux);
+    lock_guard<mutex> lock(mTimeoutRecordsMux);
     auto& item = mTimeoutRecords[config];
     auto it = item.find({index, key});
     if (it == item.end()) {
@@ -31,27 +31,52 @@ void TimeoutFlushManager::UpdateRecord(
 }
 
 void TimeoutFlushManager::FlushTimeoutBatch() {
-    lock_guard<recursive_mutex> lock(mMux);
-    vector<pair<Flusher*, size_t>> records;
-    for (auto& item : mTimeoutRecords) {
-        for (auto it = item.second.begin(); it != item.second.end();) {
-            if (time(nullptr) - it->second.mUpdateTime >= it->second.mTimeoutSecs) {
-                // cannot flush here, since flush may also update record, which might invalidate map iterator
-                records.emplace_back(it->second.mFlusher, it->second.mKey);
-                it = item.second.erase(it);
-            } else {
-                ++it;
+    multimap<string, pair<Flusher*, size_t>> records;
+    {
+        lock_guard<mutex> lock(mTimeoutRecordsMux);
+        for (auto& item : mTimeoutRecords) {
+            for (auto it = item.second.begin(); it != item.second.end();) {
+                if (time(nullptr) - it->second.mUpdateTime >= it->second.mTimeoutSecs) {
+                    // cannot flush here, since flush may also update record, which might invalidate map iterator and
+                    // lead to deadlock
+                    records.emplace(item.first, make_pair(it->second.mFlusher, it->second.mKey));
+                    it = item.second.erase(it);
+                } else {
+                    ++it;
+                }
             }
         }
     }
-    for (auto& item : records) {
-        item.first->Flush(item.second);
+    {
+        lock_guard<mutex> lock(mDeletedFlushersMux);
+        for (auto& item : records) {
+            if (mDeletedFlushers.find(make_pair(item.first, item.second.first)) == mDeletedFlushers.end()) {
+                item.second.first->Flush(item.second.second);
+            }
+        }
+        mDeletedFlushers.clear();
+    }
+}
+
+void TimeoutFlushManager::UnregisterFlushers(const string& config,
+                                             const vector<unique_ptr<FlusherInstance>>& flushers) {
+    {
+        lock_guard<mutex> lock(mTimeoutRecordsMux);
+        mTimeoutRecords.erase(config);
+    }
+    {
+        lock_guard<mutex> lock(mDeletedFlushersMux);
+        for (const auto& flusher : flushers) {
+            mDeletedFlushers.emplace(make_pair(config, flusher->GetPlugin()));
+        }
     }
 }
 
-void TimeoutFlushManager::ClearRecords(const string& config) {
-    lock_guard<recursive_mutex> lock(mMux);
-    mTimeoutRecords.erase(config);
+void TimeoutFlushManager::RegisterFlushers(const string& config, const vector<unique_ptr<FlusherInstance>>& flushers) {
+    lock_guard<mutex> lock(mDeletedFlushersMux);
+    for (const auto& flusher : flushers) {
+        mDeletedFlushers.erase(make_pair(config, flusher->GetPlugin()));
+    }
 }
 
 } // namespace logtail

Diff for: core/collection_pipeline/batch/TimeoutFlushManager.h

@@ -21,9 +21,11 @@
 
 #include <map>
 #include <mutex>
+#include <set>
 #include <string>
 #include <vector>
 
+#include "collection_pipeline/plugin/instance/FlusherInstance.h"
 #include "collection_pipeline/plugin/interface/Flusher.h"
 
 namespace logtail {
@@ -52,15 +54,21 @@ class TimeoutFlushManager {
 
     void UpdateRecord(const std::string& config, size_t index, size_t key, uint32_t timeoutSecs, Flusher* f);
     void FlushTimeoutBatch();
-    void ClearRecords(const std::string& config);
+    void UnregisterFlushers(const std::string& config, const std::vector<std::unique_ptr<FlusherInstance>>& flushers);
+    void RegisterFlushers(const std::string& config, const std::vector<std::unique_ptr<FlusherInstance>>& flushers);
 
 private:
     TimeoutFlushManager() = default;
     ~TimeoutFlushManager() = default;
 
-    std::recursive_mutex mMux;
+    // visited by all processor runner threads
+    mutable std::mutex mTimeoutRecordsMux;
     std::map<std::string, std::map<std::pair<size_t, size_t>, TimeoutRecord>> mTimeoutRecords;
 
+    // visited by main thread and num 0 processor runner thread
+    mutable std::mutex mDeletedFlushersMux;
+    std::set<std::pair<std::string, const Flusher*>> mDeletedFlushers;
+
 #ifdef APSARA_UNIT_TEST_MAIN
     friend class PipelineUnittest;
     friend class TimeoutFlushManagerUnittest;

Diff for: core/unittest/batch/TimeoutFlushManagerUnittest.cpp

@@ -24,7 +24,7 @@ class TimeoutFlushManagerUnittest : public ::testing::Test {
 public:
     void TestUpdateRecord();
     void TestFlushTimeoutBatch();
-    void TestClearRecords();
+    void TestUnregisterFlushers();
 
 protected:
     static void SetUpTestCase() {
@@ -77,16 +77,28 @@ void TimeoutFlushManagerUnittest::TestFlushTimeoutBatch() {
     APSARA_TEST_EQUAL(1U, TimeoutFlushManager::GetInstance()->mTimeoutRecords.size());
 }
 
-void TimeoutFlushManagerUnittest::TestClearRecords() {
-    TimeoutFlushManager::GetInstance()->UpdateRecord("test_config", 0, 1, 3, sFlusher.get());
-    TimeoutFlushManager::GetInstance()->ClearRecords("test_config");
+void TimeoutFlushManagerUnittest::TestUnregisterFlushers() {
+    auto flusher = new FlusherMock();
+    flusher->SetContext(sCtx);
+    flusher->SetMetricsRecordRef(FlusherMock::sName, "1");
+    auto instance = unique_ptr<FlusherInstance>(new FlusherInstance(flusher, PluginInstance::PluginMeta("1")));
+    vector<unique_ptr<FlusherInstance>> flushers;
+    flushers.push_back(std::move(instance));
+
+    TimeoutFlushManager::GetInstance()->UpdateRecord("test_config", 0, 1, 3, flusher);
+    TimeoutFlushManager::GetInstance()->UnregisterFlushers("test_config", flushers);
 
+    APSARA_TEST_EQUAL(1U, TimeoutFlushManager::GetInstance()->mDeletedFlushers.size());
     APSARA_TEST_EQUAL(0U, TimeoutFlushManager::GetInstance()->mTimeoutRecords.size());
+
+    TimeoutFlushManager::GetInstance()->FlushTimeoutBatch();
+    APSARA_TEST_TRUE(TimeoutFlushManager::GetInstance()->mDeletedFlushers.empty());
+    APSARA_TEST_TRUE(TimeoutFlushManager::GetInstance()->mTimeoutRecords.empty());
 }
 
 UNIT_TEST_CASE(TimeoutFlushManagerUnittest, TestUpdateRecord)
 UNIT_TEST_CASE(TimeoutFlushManagerUnittest, TestFlushTimeoutBatch)
-UNIT_TEST_CASE(TimeoutFlushManagerUnittest, TestClearRecords)
+UNIT_TEST_CASE(TimeoutFlushManagerUnittest, TestUnregisterFlushers)
 
 } // namespace logtail
 

Diff for: core/unittest/pipeline/PipelineUnittest.cpp

@@ -2917,15 +2917,19 @@ void PipelineUnittest::TestFlushBatch() const {
         TimeoutFlushManager::GetInstance()->UpdateRecord(configName, 0, 1, 3, nullptr);
         TimeoutFlushManager::GetInstance()->UpdateRecord(configName, 1, 1, 3, nullptr);
         APSARA_TEST_TRUE(pipeline.FlushBatch());
-        APSARA_TEST_TRUE(TimeoutFlushManager::GetInstance()->mTimeoutRecords.empty());
+        APSARA_TEST_EQUAL(0U, TimeoutFlushManager::GetInstance()->mTimeoutRecords.size());
+        APSARA_TEST_EQUAL(2U, TimeoutFlushManager::GetInstance()->mDeletedFlushers.size());
+        TimeoutFlushManager::GetInstance()->FlushTimeoutBatch();
     }
     {
         // some failed
         const_cast<FlusherMock*>(static_cast<const FlusherMock*>(pipeline.mFlushers[0]->GetPlugin()))->mIsValid = false;
         TimeoutFlushManager::GetInstance()->UpdateRecord(configName, 0, 1, 3, nullptr);
         TimeoutFlushManager::GetInstance()->UpdateRecord(configName, 1, 1, 3, nullptr);
         APSARA_TEST_FALSE(pipeline.FlushBatch());
-        APSARA_TEST_TRUE(TimeoutFlushManager::GetInstance()->mTimeoutRecords.empty());
+        APSARA_TEST_EQUAL(0U, TimeoutFlushManager::GetInstance()->mTimeoutRecords.size());
+        APSARA_TEST_EQUAL(2U, TimeoutFlushManager::GetInstance()->mDeletedFlushers.size());
+        TimeoutFlushManager::GetInstance()->FlushTimeoutBatch();
     }
 }