MINIFICPP-2669 - Remove ThreadManagementService, review changes

Author: adamdebreceni

core-framework/include/core/controller/ControllerServiceBase.h

@@ -33,12 +33,7 @@
 
 namespace org::apache::nifi::minifi::core::controller {
 
-/**
- * Controller Service base class that contains some pure virtual methods.
- *
- * Design: OnEnable is executed when the controller service is being enabled.
- * Note that keeping state here must be protected  in this function.
- */
+// A base class that helps with controller service development, contains common functionalities.
 class ControllerServiceBase : public ControllerServiceApi {
  public:
   explicit ControllerServiceBase(ControllerServiceMetadata metadata)
@@ -107,7 +102,9 @@ class ControllerServiceBase : public ControllerServiceApi {
   utils::Identifier uuid_;
   std::vector<std::shared_ptr<controller::ControllerServiceInterface> > linked_services_;
   std::shared_ptr<Configure> configuration_;
+  // valid during initialize, sink for supported properties
   ControllerServiceDescriptor* descriptor_{nullptr};
+  // valid during onEnable, provides property access
   ControllerServiceContext* context_{nullptr};
 
   std::shared_ptr<core::logging::Logger> logger_;

core-framework/include/utils/ThreadPool.h

@@ -37,7 +37,6 @@
 #include "MinifiConcurrentQueue.h"
 #include "Monitors.h"
 #include "core/expect.h"
-#include "minifi-cpp/controllers/ThreadManagementService.h"
 #include "minifi-cpp/core/logging/Logger.h"
 
 namespace org::apache::nifi::minifi::utils {
@@ -136,10 +135,7 @@ class WorkerThread {
  */
 class ThreadPool {
  public:
-  using ControllerServiceProvider = std::function<std::shared_ptr<core::controller::ControllerServiceInterface>(std::string_view)>;
-
-  ThreadPool(int max_worker_threads = 2,
-             ControllerServiceProvider controller_service_provider = nullptr, std::string name = "NamelessPool");
+  ThreadPool(int max_worker_threads = 2, std::string name = "NamelessPool");
 
   ThreadPool(const ThreadPool &other) = delete;
   ThreadPool& operator=(const ThreadPool &other) = delete;
@@ -232,20 +228,6 @@ class ThreadPool {
       start();
   }
 
-  void setControllerServiceProvider(ControllerServiceProvider controller_service_provider) {
-    std::lock_guard<std::recursive_mutex> lock(manager_mutex_);
-    bool was_running = running_;
-    if (was_running) {
-      shutdown();
-    }
-    controller_service_provider_ = controller_service_provider;
-    if (was_running)
-      start();
-  }
-
- private:
-  std::shared_ptr<controllers::ThreadManagementService> createThreadManager() const;
-
  protected:
   std::thread createThread(std::function<void()> &&functor) {
     return std::thread([ functor ]() mutable {
@@ -273,8 +255,6 @@ class ThreadPool {
   std::thread manager_thread_;
   std::thread delayed_scheduler_thread_;
   std::atomic<bool> running_;
-  ControllerServiceProvider controller_service_provider_;
-  std::shared_ptr<controllers::ThreadManagementService> thread_manager_;
   ConcurrentQueue<std::shared_ptr<WorkerThread>> deceased_thread_queue_;
   ConditionConcurrentQueue<Worker> worker_queue_;
   std::priority_queue<Worker, std::vector<Worker>, DelayedTaskComparator> delayed_worker_queue_;

core-framework/src/utils/ThreadPool.cpp

@@ -22,12 +22,11 @@ using namespace std::literals::chrono_literals;
 
 namespace org::apache::nifi::minifi::utils {
 
-ThreadPool::ThreadPool(int max_worker_threads, ControllerServiceProvider controller_service_provider, std::string name)
+ThreadPool::ThreadPool(int max_worker_threads, std::string name)
     : thread_reduction_count_(0),
       max_worker_threads_(max_worker_threads),
       current_workers_(0),
       running_(false),
-      controller_service_provider_(std::move(controller_service_provider)),
       name_(std::move(name)),
       logger_(core::logging::LoggerFactory<ThreadPool>::getLogger()) {
 }
@@ -141,71 +140,15 @@ void ThreadPool::manageWorkers() {
     }
   }
 
-  if (nullptr != thread_manager_) {
-    while (running_) {
-      auto wait_period = 500ms;
-      {
-        std::unique_lock<std::recursive_mutex> manager_lock(manager_mutex_, std::try_to_lock);
-        if (!manager_lock.owns_lock()) {
-          // Threadpool is being stopped/started or config is being changed, better wait a bit
-          std::this_thread::sleep_for(10ms);
-          continue;
-        }
-        if (thread_manager_->isAboveMax(current_workers_)) {
-          auto max = 1;  // thread_manager_->getMaxConcurrentTasks();
-          auto differential = current_workers_ - max;
-          thread_reduction_count_ += differential;
-        } else if (thread_manager_->shouldReduce()) {
-          if (current_workers_ > 1)
-            thread_reduction_count_++;
-          thread_manager_->reduce();
-        } else if (thread_manager_->canIncrease() && max_worker_threads_ > current_workers_) {  // increase slowly
-          std::unique_lock<std::mutex> worker_queue_lock(worker_queue_mutex_);
-          auto worker_thread = std::make_shared<WorkerThread>();
-          worker_thread->thread_ = createThread([this, worker_thread] { run_tasks(worker_thread); });
-          thread_queue_.push_back(worker_thread);
-          current_workers_++;
-        }
-        std::shared_ptr<WorkerThread> thread_ref;
-        while (deceased_thread_queue_.tryDequeue(thread_ref)) {
-          std::unique_lock<std::mutex> worker_queue_lock(worker_queue_mutex_);
-          if (thread_ref->thread_.joinable())
-            thread_ref->thread_.join();
-          thread_queue_.erase(std::remove(thread_queue_.begin(), thread_queue_.end(), thread_ref), thread_queue_.end());
-        }
-      }
-      std::this_thread::sleep_for(wait_period);
-    }
-  } else {
-    for (auto &thread : thread_queue_) {
-      if (thread->thread_.joinable())
-        thread->thread_.join();
-    }
-  }
-}
-
-std::shared_ptr<controllers::ThreadManagementService> ThreadPool::createThreadManager() const {
-  if (!controller_service_provider_) {
-    return nullptr;
-  }
-  auto service = controller_service_provider_("ThreadPoolManager");
-  if (!service) {
-    logger_->log_info("Could not find a ThreadPoolManager service");
-    return nullptr;
+  for (auto &thread : thread_queue_) {
+    if (thread->thread_.joinable())
+      thread->thread_.join();
   }
-  auto thread_manager_service = std::dynamic_pointer_cast<controllers::ThreadManagementService>(service);
-  if (!thread_manager_service) {
-    logger_->log_error("Found ThreadPoolManager, but it is not a ThreadManagementService");
-    return nullptr;
-  }
-  return thread_manager_service;
 }
 
 void ThreadPool::start() {
   std::lock_guard<std::recursive_mutex> lock(manager_mutex_);
   if (!running_) {
-    thread_manager_ = createThreadManager();
-
     running_ = true;
     worker_queue_.start();
     manager_thread_ = std::thread(&ThreadPool::manageWorkers, this);

libminifi/include/controllers/ThreadManagementService.h

@@ -38,12 +38,7 @@ enum ControllerServiceState {
   ENABLED
 };
 
-/**
- * Controller Service base class that contains some pure virtual methods.
- *
- * Design: OnEnable is executed when the controller service is being enabled.
- * Note that keeping state here must be protected  in this function.
- */
+// Wrapper class to manage common functionalities of controller service implementations (properties, name, id, etc)
 class ControllerService : public ConfigurableComponentImpl, public CoreComponentImpl {
   class ControllerServiceDescriptorImpl : public ControllerServiceDescriptor {
    public:
@@ -128,7 +123,7 @@ class ControllerService : public ConfigurableComponentImpl, public CoreComponent
 
   template<typename T = ControllerServiceInterface>
   gsl::not_null<T*> getImplementation() {
-    return gsl::make_not_null(dynamic_cast<T*>(impl_.get()));
+    return gsl::make_not_null(dynamic_cast<T*>(impl_->getControllerServiceInterface()));
   }
 
   bool supportsDynamicProperties() const final {

libminifi/include/core/controller/ControllerService.h

@@ -53,7 +53,7 @@ FlowController::FlowController(std::shared_ptr<core::Repository> provenance_repo
     : core::controller::ForwardingControllerServiceProvider(core::className<FlowController>()),
       running_(false),
       initialized_(false),
-      thread_pool_(5, nullptr, "Flowcontroller threadpool"),
+      thread_pool_(5, "Flowcontroller threadpool"),
       configuration_(std::move(configure)),
       provenance_repo_(std::move(provenance_repo)),
       flow_file_repo_(std::move(flow_file_repo)),
@@ -275,12 +275,6 @@ void FlowController::load(bool reload) {
   if (!thread_pool_.isRunning() || reload) {
     thread_pool_.shutdown();
     thread_pool_.setMaxConcurrentTasks(configuration_->getInt(Configure::nifi_flow_engine_threads, 5));
-    thread_pool_.setControllerServiceProvider([this] (std::string_view name) -> std::shared_ptr<core::controller::ControllerServiceInterface> {
-      if (auto service = this->getControllerService(std::string{name})) {
-        return {service, service->getImplementation()};
-      }
-      return {};
-    });
     thread_pool_.start();
   }
 

libminifi/src/FlowController.cpp

@@ -62,7 +62,7 @@ C2Agent::C2Agent(std::shared_ptr<Configure> configuration,
       configuration_(std::move(configuration)),
       node_reporter_(std::move(node_reporter)),
       filesystem_(std::move(filesystem)),
-      thread_pool_(2, nullptr, "C2 threadpool"),
+      thread_pool_(2, "C2 threadpool"),
       request_restart_(std::move(request_restart)),
       last_run_(std::chrono::steady_clock::now()),
       asset_manager_(asset_manager) {

libminifi/src/c2/C2Agent.cpp

@@ -716,7 +716,7 @@ TEST_CASE("The encrypted flow configuration can be decrypted with the correct ke
   REQUIRE(controller_service_node_before);
   const auto* const controller_service_node_after = process_group_after->findControllerService(controller_service_id);
   REQUIRE(controller_service_node_after);
-  const auto controller_service_after = controller_service_node_before->getControllerServiceImplementation();
+  const auto controller_service_after = controller_service_node_after->getControllerServiceImplementation();
   REQUIRE(controller_service_after);
   CHECK(controller_service_before->getProperty("CA Certificate") == controller_service_after->getProperty("CA Certificate"));
   CHECK(controller_service_before->getProperty("Passphrase") == controller_service_after->getProperty("Passphrase"));

libminifi/test/unit/JsonFlowSerializerTests.cpp

@@ -331,7 +331,7 @@ TEST_CASE("The encrypted flow configuration can be decrypted with the correct ke
   REQUIRE(controller_service_node_before);
   const auto* const controller_service_node_after = process_group_after->findControllerService(controller_service_id);
   REQUIRE(controller_service_node_after);
-  const auto controller_service_after = controller_service_node_before->getControllerServiceImplementation();
+  const auto controller_service_after = controller_service_node_after->getControllerServiceImplementation();
   REQUIRE(controller_service_after);
   CHECK(controller_service_before->getProperty("CA Certificate") == controller_service_after->getProperty("CA Certificate"));
   CHECK(controller_service_before->getProperty("Passphrase") == controller_service_after->getProperty("Passphrase"));

libminifi/test/unit/YamlFlowSerializerTests.cpp

@@ -1,6 +1,7 @@
 add_library(minifi-api-common INTERFACE)
 target_include_directories(minifi-api-common INTERFACE common/include)
 target_link_libraries(minifi-api-common INTERFACE gsl-lite)
+target_compile_definitions(minifi-api-common INTERFACE MINIFI_VERSION_STR="${MINIFI_VERSION_STR}")
 
 add_library(minifi-api INTERFACE)
 target_include_directories(minifi-api INTERFACE include)