(nifi-minifi-cpp) 02/04: MINIFICPP-2505 SchedulingAgent::scheduled_processors_ isnt thread safe

[sz...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git

commit 212b1458c72dab812bc99eb3b47fd887cc3c190f
Author: Martin Zink <ma...@apache.org>
AuthorDate: Sun Dec 29 21:00:29 2024 +0100

    MINIFICPP-2505 SchedulingAgent::scheduled_processors_ isnt thread safe
    
    Closes #1910
    
    Signed-off-by: Marton Szasz <sz...@apache.org>
---
 libminifi/include/SchedulingAgent.h |  4 ++--
 libminifi/src/SchedulingAgent.cpp   | 33 ++++++++++++++++++---------------
 2 files changed, 20 insertions(+), 17 deletions(-)

diff --git a/libminifi/include/SchedulingAgent.h b/libminifi/include/SchedulingAgent.h
index 99ebd62c8..f50ddc571 100644
--- a/libminifi/include/SchedulingAgent.h
+++ b/libminifi/include/SchedulingAgent.h
@@ -150,8 +150,8 @@ class SchedulingAgent {
   };
 
   std::shared_ptr<core::logging::Logger> logger_;
-  mutable std::mutex watchdog_mtx_;  // used to protect the set below
-  std::set<SchedulingInfo> scheduled_processors_;  // set was chosen to avoid iterator invalidation
+  mutable std::mutex watchdog_mtx_;  // used to protect the vector below
+  std::vector<gsl::not_null<SchedulingInfo*>> scheduled_processors_;
   std::unique_ptr<utils::CallBackTimer> watchDogTimer_;
   std::chrono::milliseconds alert_time_;
 };
diff --git a/libminifi/src/SchedulingAgent.cpp b/libminifi/src/SchedulingAgent.cpp
index f42ca057f..90da81971 100644
--- a/libminifi/src/SchedulingAgent.cpp
+++ b/libminifi/src/SchedulingAgent.cpp
@@ -18,10 +18,12 @@
  * limitations under the License.
  */
 #include "SchedulingAgent.h"
+
 #include <chrono>
+#include <memory>
 #include <thread>
 #include <utility>
-#include <memory>
+
 #include "core/Processor.h"
 #include "utils/gsl.h"
 
@@ -68,18 +70,19 @@ nonstd::expected<void, std::exception_ptr> SchedulingAgent::triggerAndCommit(cor
     return {};
   }
 
-  auto schedule_it = scheduled_processors_.end();
-
+  auto processor_scheduling_info = SchedulingInfo(processor);
   {
     std::lock_guard<std::mutex> lock(watchdog_mtx_);
-    schedule_it = scheduled_processors_.emplace(processor).first;
+    scheduled_processors_.push_back(gsl::make_not_null(&processor_scheduling_info));
   }
 
-  const auto guard = gsl::finally([this, &schedule_it](){
+  const auto guard = gsl::finally([this, &processor_scheduling_info](){
     std::lock_guard<std::mutex> lock(watchdog_mtx_);
-    scheduled_processors_.erase(schedule_it);
+    [[maybe_unused]] const auto erased_scheduling_infos_count = std::erase(scheduled_processors_, gsl::make_not_null(&processor_scheduling_info));
+    gsl_Assert(1 == erased_scheduling_infos_count);
   });
 
+
   processor->incrementActiveTasks();
   auto decrement_task = gsl::finally([processor]() { processor->decrementActiveTask(); });
 
@@ -107,16 +110,16 @@ nonstd::expected<bool, std::exception_ptr> SchedulingAgent::trigger(core::Proces
     return false;
   }
 
-  auto schedule_it = scheduled_processors_.end();
-
+  auto processor_scheduling_info = SchedulingInfo(processor);
   {
     std::lock_guard<std::mutex> lock(watchdog_mtx_);
-    schedule_it = scheduled_processors_.emplace(processor).first;
+    scheduled_processors_.push_back(gsl::make_not_null(&processor_scheduling_info));
   }
 
-  const auto guard = gsl::finally([this, &schedule_it](){
+  const auto guard = gsl::finally([this, &processor_scheduling_info](){
     std::lock_guard<std::mutex> lock(watchdog_mtx_);
-    scheduled_processors_.erase(schedule_it);
+    [[maybe_unused]] const auto erased_scheduling_infos_count = std::erase(scheduled_processors_, gsl::make_not_null(&processor_scheduling_info));
+    gsl_Assert(1 == erased_scheduling_infos_count);
   });
 
   processor->incrementActiveTasks();
@@ -141,11 +144,11 @@ void SchedulingAgent::watchDogFunc() {
   std::lock_guard<std::mutex> lock(watchdog_mtx_);
   auto now = std::chrono::steady_clock::now();
   for (const auto& info : scheduled_processors_) {
-    auto elapsed = now - info.last_alert_time_;
+    auto elapsed = now - info->last_alert_time_;
     if (elapsed > alert_time_) {
-      int64_t elapsed_ms{ std::chrono::duration_cast<std::chrono::milliseconds>(now - info.start_time_).count() };
-      logger_->log_warn("{}::onTrigger has been running for {}  ms in {}", info.name_, elapsed_ms, info.uuid_);
-      info.last_alert_time_ = now;
+      int64_t elapsed_ms{ std::chrono::duration_cast<std::chrono::milliseconds>(now - info->start_time_).count() };
+      logger_->log_warn("{}::onTrigger has been running for {}  ms in {}", info->name_, elapsed_ms, info->uuid_);
+      info->last_alert_time_ = now;
     }
   }
 }