Fix race condition associated to module reload and modulemanager

src/agent/include/agent.hpp

@@ -13,7 +13,9 @@
 
 #include <sysInfo.hpp>
 
+#include <atomic>
 #include <memory>
+#include <mutex>
 #include <string>
 
 /// @brief Agent class
@@ -76,4 +78,13 @@ class Agent
 
     /// @brief Centralized configuration
     centralized_configuration::CentralizedConfiguration m_centralizedConfiguration;
+
+    /// @brief Mutex to coordinate agent reload
+    std::mutex m_reloadMutex;
+
+    /// @brief Indicates if the agent is running
+    std::atomic<bool> m_running = true;
+
+    /// @brief Agent thread count
+    size_t m_agentThreadCount;
 };

src/agent/src/agent.cpp

@@ -58,16 +58,14 @@ Agent::Agent(const std::string& configFilePath, std::unique_ptr<ISignalHandler>
 
     m_centralizedConfiguration.ReloadModulesFunction([this]() { ReloadModules(); });
 
-    auto agentThreadCount =
+    m_agentThreadCount =
         m_configurationParser->GetConfig<size_t>("agent", "thread_count").value_or(config::DEFAULT_THREAD_COUNT);
 
-    if (agentThreadCount < config::DEFAULT_THREAD_COUNT)
+    if (m_agentThreadCount < config::DEFAULT_THREAD_COUNT)
     {
         LogWarn("thread_count must be greater than {}. Using default value.", config::DEFAULT_THREAD_COUNT);
-        agentThreadCount = config::DEFAULT_THREAD_COUNT;
+        m_agentThreadCount = config::DEFAULT_THREAD_COUNT;
     }
-
-    m_taskManager.Start(agentThreadCount);
 }
 
 Agent::~Agent()
@@ -77,15 +75,34 @@ Agent::~Agent()
 
 void Agent::ReloadModules()
 {
-    LogInfo("Reloading Modules");
-    m_configurationParser->ReloadConfiguration();
-    m_moduleManager.Stop();
-    m_moduleManager.Setup();
-    m_moduleManager.Start();
+    std::lock_guard<std::mutex> lock(m_reloadMutex);
+
+    if (m_running.load())
+    {
+        try
+        {
+            LogInfo("Reloading Modules");
+            m_configurationParser->ReloadConfiguration();
+            m_moduleManager.Stop();
+            m_moduleManager.Setup();
+            m_moduleManager.Start();
+            LogInfo("Modules reloaded");
+        }
+        catch (const std::exception& e)
+        {
+            LogError("Error reloading modules: {}", e.what());
+        }
+    }
+    else
+    {
+        LogWarn("Agent cannot reload modules while start up or shutdown is in progress.");
+    }
 }
 
 void Agent::Run()
 {
+    m_taskManager.Start(m_agentThreadCount);
+
     // Check if the server recognizes the agent
     m_communicator.SendAuthenticationRequest();
 
@@ -144,8 +161,18 @@ void Agent::Run()
             }),
         "CommandsProcessing");
 
+    {
+        std::unique_lock<std::mutex> lock(m_reloadMutex);
+        m_running.store(true);
+    }
+
     m_signalHandler->WaitForSignal();
 
+    {
+        std::unique_lock<std::mutex> lock(m_reloadMutex);
+        m_running.store(false);
+    }
+
     m_commandHandler.Stop();
     m_communicator.Stop();
     m_moduleManager.Stop();

src/agent/src/signal_handler.cpp

@@ -17,6 +17,7 @@ void SignalHandler::HandleSignal([[maybe_unused]] int signal)
 
 void SignalHandler::WaitForSignal()
 {
+    KeepRunning.store(true);
     std::unique_lock<std::mutex> lock(m_cvMutex);
     m_cv.wait(lock, [] { return !KeepRunning.load(); });
 }

src/agent/task_manager/include/task_manager.hpp

@@ -7,6 +7,7 @@
 
 #include <functional>
 #include <memory>
+#include <mutex>
 #include <string>
 #include <thread>
 #include <vector>
@@ -53,4 +54,7 @@ class TaskManager : public ITaskManager<boost::asio::awaitable<void>>
 
     /// @brief Number of enqueued threads
     size_t m_numEnqueuedThreads = 0;
+
+    /// @brief Mutex to control Start and Stop operations
+    mutable std::mutex m_mutex;
 };

src/agent/task_manager/src/task_manager.cpp

@@ -15,6 +15,8 @@ TaskManager::~TaskManager()
 
 void TaskManager::Start(size_t numThreads)
 {
+    std::lock_guard<std::mutex> lock(m_mutex);
+
     if (m_work || !m_threads.empty())
     {
         LogError("Task manager already started");
@@ -32,9 +34,11 @@ void TaskManager::Start(size_t numThreads)
 
 void TaskManager::Stop()
 {
+    std::lock_guard<std::mutex> lock(m_mutex);
+
     if (m_work)
     {
-        m_work->reset();
+        m_work.reset();
     }
 
     if (!m_ioContext.stopped())
@@ -59,6 +63,8 @@ void TaskManager::Stop()
 
 void TaskManager::EnqueueTask(std::function<void()> task, const std::string& taskID)
 {
+    std::lock_guard<std::mutex> lock(m_mutex);
+
     if (++m_numEnqueuedThreads > m_threads.size())
     {
         LogError("Enqueued more threaded tasks than available threads");
@@ -82,6 +88,8 @@ void TaskManager::EnqueueTask(std::function<void()> task, const std::string& tas
 
 void TaskManager::EnqueueTask(boost::asio::awaitable<void> task, const std::string& taskID)
 {
+    std::lock_guard<std::mutex> lock(m_mutex);
+
     // NOLINTBEGIN(cppcoreguidelines-avoid-capturing-lambda-coroutines)
     boost::asio::co_spawn(
         m_ioContext,
@@ -105,5 +113,6 @@ void TaskManager::EnqueueTask(boost::asio::awaitable<void> task, const std::stri
 
 size_t TaskManager::GetNumEnqueuedThreads() const
 {
+    std::lock_guard<std::mutex> lock(m_mutex);
     return m_numEnqueuedThreads;
 }

src/modules/include/moduleManager.hpp

@@ -1,15 +1,18 @@
 #pragma once
 
-#include <map>
-#include <memory>
-#include <string>
-#include <thread>
 #include <multitype_queue.hpp>
 #include <moduleWrapper.hpp>
 #include <task_manager.hpp>
 
 #include <boost/asio/awaitable.hpp>
 
+#include <map>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <thread>
+
+
 class ModuleManager {
 public:
     ModuleManager(const std::function<int(Message)>& pushMessage, std::shared_ptr<configuration::ConfigurationParser> configurationParser, std::string uuid)
@@ -46,6 +49,15 @@ class ModuleManager {
 
     void AddModules();
     std::shared_ptr<ModuleWrapper> GetModule(const std::string & name);
+
+    /// @brief Start the modules
+    ///
+    /// This function begins the procedure to start the modules and blocks until the Start function
+    /// for each module has been called. However, it does not guarantee that the modules are fully
+    /// operational upon return; they may still be in the process of initializing.
+    ///
+    /// @note Call this function before interacting with the modules to ensure the startup process is initiated.
+    /// @warning Ensure the modules have fully started before performing any operations that depend on them.
     void Start();
     void Setup();
     void Stop();
@@ -56,4 +68,6 @@ class ModuleManager {
     std::function<int(Message)> m_pushMessage;
     std::shared_ptr<configuration::ConfigurationParser> m_configurationParser;
     std::string m_agentUUID;
+    std::mutex m_mutex;
+    std::atomic<int> m_started {0};
 };

src/modules/src/moduleManager.cpp

@@ -24,31 +24,62 @@ void ModuleManager::AddModules() {
     Setup();
 }
 
-std::shared_ptr<ModuleWrapper> ModuleManager::GetModule(const std::string & name) {
-    auto it = m_modules.find(name);
-    if (it != m_modules.end()) {
+std::shared_ptr<ModuleWrapper> ModuleManager::GetModule(const std::string & name)
+{
+    if (auto it = m_modules.find(name); it != m_modules.end())
+    {
         return it->second;
     }
     return nullptr;
 }
 
-void ModuleManager::Start() {
+void ModuleManager::Start()
+{
+    std::unique_lock<std::mutex> lock(m_mutex);
+
     m_taskManager.Start(m_modules.size());
+    std::condition_variable cv;
+
     for (const auto &[_, module] : m_modules)
     {
-        m_taskManager.EnqueueTask([module]() { module->Start(); }, module->Name());
+        m_taskManager.EnqueueTask(
+            [module, this, &cv]
+            {
+                ++m_started;
+                cv.notify_one();
+                module->Start();
+            }
+            , module->Name()
+        );
     }
+
+    cv.wait(
+        lock,
+        [this]
+        {
+            return m_started.load() == static_cast<int>(m_modules.size());
+        }
+    );
 }
 
-void ModuleManager::Setup() {
-    for (const auto &[_, module] : m_modules) {
+void ModuleManager::Setup()
+{
+    std::lock_guard<std::mutex> lock(m_mutex);
+
+    for (const auto &[_, module] : m_modules)
+    {
         module->Setup(m_configurationParser);
     }
 }
 
-void ModuleManager::Stop() {
-    for (const auto &[_, module] : m_modules) {
+void ModuleManager::Stop()
+{
+    std::lock_guard<std::mutex> lock(m_mutex);
+
+    for (const auto &[_, module] : m_modules)
+    {
         module->Stop();
+        m_started--;
     }
     m_taskManager.Stop();
 }