polish code

alibaba · Dec 11, 2023 · 2a01ca6 · 2a01ca6
1 parent 94ff42a
commit 2a01ca6
Show file tree

Hide file tree

Showing 10 changed files with 42 additions and 35 deletions.
diff --git a/core/common/Thread.h b/core/common/Thread.h
@@ -15,15 +15,14 @@
  */
 
 #pragma once
+
 #include <boost/thread.hpp>
 
 #include <functional>
 #include <memory>
 #include <thread>
 #include <utility>
 
-#include "boost/thread.hpp"
-
 namespace logtail {
 
 class Thread {

diff --git a/core/config/provider/CommonConfigProvider.cpp b/core/config/provider/CommonConfigProvider.cpp
@@ -14,12 +14,12 @@
 
 #include "config/provider/CommonConfigProvider.h"
 
+#include <json/json.h>
+
 #include <filesystem>
 #include <iostream>
 #include <random>
 
-#include <json/json.h>
-
 #include "app_config/AppConfig.h"
 #include "application/Application.h"
 #include "common/LogtailCommonFlags.h"
@@ -28,8 +28,8 @@
 #include "logger/Logger.h"
 #include "monitor/LogFileProfiler.h"
 #include "sdk/Common.h"
-#include "sdk/Exception.h"
 #include "sdk/CurlImp.h"
+#include "sdk/Exception.h"
 
 using namespace std;
 
@@ -38,7 +38,17 @@ DEFINE_FLAG_INT32(config_update_interval, "second", 10);
 namespace logtail {
 
 CommonConfigProvider::~CommonConfigProvider() {
-    mThreadIsRunning = false;
+    {
+        lock_guard<mutex> lock(mThreadRunningMux);
+        mIsThreadRunning = false;
+    }
+    mStopCV.notify_one();
+    future_status s = mThreadRes.wait_for(chrono::seconds(1));
+    if (s == future_status::ready) {
+        LOG_INFO(sLogger, ("common config provider", "stopped successfully"));
+    } else {
+        LOG_WARNING(sLogger, ("common config provider", "forced to stopped"));
+    }
 }
 
 void CommonConfigProvider::Init(const string& dir) {
@@ -82,23 +92,23 @@ void CommonConfigProvider::Init(const string& dir) {
         LOG_INFO(sLogger, ("ilogtail_configserver_tags", confJson["ilogtail_tags"].toStyledString()));
     }
 
-    mCheckUpdateThread = thread([this] { CheckUpdateThread(); });
+    mThreadRes = async(launch::async, &CommonConfigProvider::CheckUpdateThread, this);
 }
 
 void CommonConfigProvider::CheckUpdateThread() {
+    LOG_INFO(sLogger, ("common config provider", "started"));
     usleep((rand() % 10) * 100 * 1000);
     int32_t lastCheckTime = 0;
-    mThreadIsRunning = true;
-    while (mThreadIsRunning.load()) {
+    unique_lock<mutex> lock(mThreadRunningMux);
+    while (mIsThreadRunning) {
         int32_t curTime = time(NULL);
         if (curTime - lastCheckTime >= INT32_FLAG(config_update_interval)) {
             GetConfigUpdate();
             lastCheckTime = curTime;
         }
-        if (mThreadIsRunning.load())
-            sleep(1);
-        else
+        if (mStopCV.wait_for(lock, std::chrono::seconds(3), [this]() { return !mIsThreadRunning; })) {
             break;
+        }
     }
 }
 
@@ -313,9 +323,9 @@ void CommonConfigProvider::UpdateRemoteConfig(
                 error_code ec;
                 filesystem::rename(tmpFilePath, filePath, ec);
                 if (ec) {
-                    LOG_WARNING(
-                        sLogger,
-                        ("failed to dump config file", filePath.string())("error code", ec.value())("error msg", ec.message()));
+                    LOG_WARNING(sLogger,
+                                ("failed to dump config file",
+                                 filePath.string())("error code", ec.value())("error msg", ec.message()));
                     filesystem::remove(tmpFilePath, ec);
                 }
                 break;

diff --git a/core/config/provider/CommonConfigProvider.h b/core/config/provider/CommonConfigProvider.h
@@ -16,8 +16,9 @@
 
 #pragma once
 
-#include <atomic>
+#include <condition_variable>
 #include <cstdint>
+#include <future>
 #include <string>
 #include <unordered_map>
 #include <vector>
@@ -73,8 +74,10 @@ class CommonConfigProvider : public ConfigProvider {
     int mConfigServerAddressId = 0;
     std::vector<std::string> mConfigServerTags;
 
-    JThread mCheckUpdateThread;
-    std::atomic_bool mThreadIsRunning = false;
+    std::future<void> mThreadRes;
+    std::mutex mThreadRunningMux;
+    bool mIsThreadRunning = true;
+    std::condition_variable mStopCV;
     std::unordered_map<std::string, int64_t> mConfigNameVersionMap;
     bool mConfigServerAvailable = false;
 };

diff --git a/core/event_handler/LogInput.cpp b/core/event_handler/LogInput.cpp
@@ -88,7 +88,6 @@ void LogInput::Start() {
 
     mInteruptFlag = false;
     new Thread([this]() { ProcessLoop(); });
-    LOG_INFO(sLogger, ("event handle daemon", "started"));
 }
 
 void LogInput::Resume() {
@@ -344,7 +343,7 @@ void LogInput::UpdateCriticalMetric(int32_t curTime) {
 }
 
 void* LogInput::ProcessLoop() {
-    LOG_DEBUG(sLogger, ("LogInputThread", "Start"));
+    LOG_INFO(sLogger, ("event handle daemon", "started"));
     EventDispatcher* dispatcher = EventDispatcher::GetInstance();
     dispatcher->StartTimeCount();
     int32_t prevTime = time(NULL);

diff --git a/core/monitor/LogtailAlarm.cpp b/core/monitor/LogtailAlarm.cpp
@@ -104,7 +104,6 @@ LogtailAlarm::LogtailAlarm() {
 
 void LogtailAlarm::Init() {
     mThreadRes = async(launch::async, &LogtailAlarm::SendAlarmLoop, this);
-    LOG_INFO(sLogger, ("alarm gathering", "started"));
 }
 
 void LogtailAlarm::Stop() {
@@ -123,9 +122,9 @@ void LogtailAlarm::Stop() {
 }
 
 bool LogtailAlarm::SendAlarmLoop() {
+    LOG_INFO(sLogger, ("alarm gathering", "started"));
     {
         unique_lock<mutex> lock(mThreadRunningMux);
-        mIsThreadRunning = true;
         while (mIsThreadRunning) {
             SendAllRegionAlarm();
             if (mStopCV.wait_for(lock, std::chrono::seconds(3), [this]() { return !mIsThreadRunning; })) {

diff --git a/core/monitor/LogtailAlarm.h b/core/monitor/LogtailAlarm.h
@@ -118,8 +118,8 @@ struct LogtailAlarmMessage {
 class LogtailAlarm {
 public:
     static LogtailAlarm* GetInstance() {
-        static LogtailAlarm ptr;
-        return &ptr;
+        static LogtailAlarm instance;
+        return &instance;
     }
 
     void Init();
@@ -147,7 +147,7 @@ class LogtailAlarm {
 
     std::future<bool> mThreadRes;
     std::mutex mThreadRunningMux;
-    bool mIsThreadRunning = false;
+    bool mIsThreadRunning = true;
     std::condition_variable mStopCV;
 
 

diff --git a/core/monitor/Monitor.cpp b/core/monitor/Monitor.cpp
@@ -113,7 +113,6 @@ bool LogtailMonitor::Init() {
 
     // Initialize monitor thread.
     mThreadRes = async(launch::async, &LogtailMonitor::Monitor, this);
-    LOG_INFO(sLogger, ("profiling", "started"));
     return true;
 }
 
@@ -131,12 +130,15 @@ void LogtailMonitor::Stop() {
 }
 
 void LogtailMonitor::Monitor() {
+    LOG_INFO(sLogger, ("profiling", "started"));
     int32_t lastMonitorTime = time(NULL);
     CpuStat curCpuStat;
     {
         unique_lock<mutex> lock(mThreadRunningMux);
-        mIsThreadRunning = true;
         while (mIsThreadRunning) {
+            if (mStopCV.wait_for(lock, std::chrono::seconds(1), [this]() { return !mIsThreadRunning; })) {
+                break;
+            }
             GetCpuStat(curCpuStat);
 
             // Update mRealtimeCpuStat for InputFlowControl.
@@ -196,10 +198,6 @@ void LogtailMonitor::Monitor() {
                 if (!DumpMonitorInfo(monitorTime))
                     LOG_ERROR(sLogger, ("Fail to dump monitor info", ""));
             }
-
-            if (mStopCV.wait_for(lock, std::chrono::seconds(1), [this]() { return !mIsThreadRunning; })) {
-                break;
-            }
         }
     }
     SendStatusProfile(true);

diff --git a/core/monitor/Monitor.h b/core/monitor/Monitor.h
@@ -146,7 +146,7 @@ class LogtailMonitor : public MetricStore {
 
     std::future<void> mThreadRes;
     std::mutex mThreadRunningMux;
-    bool mIsThreadRunning = false;
+    bool mIsThreadRunning = true;
     std::condition_variable mStopCV;
 
     // Control report status profile frequency.

diff --git a/core/polling/PollingDirFile.cpp b/core/polling/PollingDirFile.cpp
@@ -70,7 +70,6 @@ void PollingDirFile::Start() {
     ClearCache();
     mRuningFlag = true;
     mThreadPtr = CreateThread([this]() { Polling(); });
-    LOG_INFO(sLogger, ("polling discovery", "started"));
 }
 
 void PollingDirFile::Stop() {
@@ -123,6 +122,7 @@ void PollingDirFile::CheckConfigPollingStatCount(const int32_t lastStatCount,
 }
 
 void PollingDirFile::Polling() {
+    LOG_INFO(sLogger, ("polling discovery", "started"));
     mHoldOnFlag = false;
     while (mRuningFlag) {
         LOG_DEBUG(sLogger, ("start dir file polling, mCurrentRound", mCurrentRound));

diff --git a/core/polling/PollingModify.cpp b/core/polling/PollingModify.cpp
@@ -48,7 +48,6 @@ void PollingModify::Start() {
     ClearCache();
     mRuningFlag = true;
     mThreadPtr = CreateThread([this]() { Polling(); });
-    LOG_INFO(sLogger, ("polling modify", "started"));
 }
 
 void PollingModify::Stop() {
@@ -237,7 +236,7 @@ bool PollingModify::UpdateDeletedFile(const SplitedFilePath& filePath,
 }
 
 void PollingModify::Polling() {
-    LOG_INFO(sLogger, ("PollingModify::Polling", "start"));
+    LOG_INFO(sLogger, ("polling modify", "started"));
     mHoldOnFlag = false;
     while (mRuningFlag) {
         {