MINIFICPP-2877 Fix a lock ordering issue

controller/tests/ControllerTests.cpp

@@ -102,8 +102,8 @@ class TestUpdateSink : public minifi::state::StateMonitor {
     func(*controller);
   }
 
-  void executeOnAllComponents(std::function<void(minifi::state::StateController&)> func) override {
-    func(*controller);
+  std::vector<minifi::state::ComponentState> getComponentStates() const override {
+    return {{.name = controller->getComponentName(), .uuid = controller->getComponentUUID(), .running = controller->isRunning()}};
   }
 
   std::string getComponentName() const override {

libminifi/include/FlowController.h

@@ -106,7 +106,7 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi
   }
 
   void executeOnComponent(const std::string& id_or_name, std::function<void(state::StateController&)> func) override;
-  void executeOnAllComponents(std::function<void(state::StateController&)> func) override;
+  std::vector<state::ComponentState> getComponentStates() const override;
 
   int16_t clearConnection(const std::string &connection) override;
 
@@ -167,7 +167,6 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi
   std::unique_ptr<core::ProcessGroup> loadInitialFlow();
 
   void loadFlowRepo();
-  std::vector<state::StateController*> getAllComponents();
   state::StateController* getComponent(const std::string& id_or_name);
   gsl::not_null<std::unique_ptr<state::ProcessorController>> createController(core::Processor& processor);
 

libminifi/include/RootProcessGroupWrapper.h

@@ -32,6 +32,8 @@
 #include "minifi-cpp/utils/gsl.h"
 #include "minifi-cpp/core/logging/Logger.h"
 #include "c2/ControllerSocketProtocol.h"
+#include "core/state/UpdateController.h"
+#include "core/state/ComponentStateCache.h"
 
 namespace org::apache::nifi::minifi {
 
@@ -46,12 +48,15 @@ class RootProcessGroupWrapper {
     if (metrics_publisher_store_) {
       metrics_publisher_store_->clearMetricNodes();
     }
+    component_cache_.setRoot(nullptr);
   }
 
   void updatePropertyValue(const std::string& processor_name, const std::string& property_name, const std::string& property_value);
   std::string getName() const;
   utils::Identifier getComponentUUID() const;
   std::string getVersion() const;
+
+  std::vector<state::ComponentState> getProcessorStates() const;
   const core::ProcessGroup* getRoot() const { return root_.get(); }
   void setNewRoot(std::unique_ptr<core::ProcessGroup> new_root);
   void restoreBackup();
@@ -74,9 +79,6 @@ class RootProcessGroupWrapper {
   state::StateController* getProcessorController(const std::string& id_or_name,
       const std::function<gsl::not_null<std::unique_ptr<state::ProcessorController>>(core::Processor&)>& controllerFactory);
 
-  std::optional<std::vector<state::StateController*>> getAllProcessorControllers(
-          const std::function<gsl::not_null<std::unique_ptr<state::ProcessorController>>(core::Processor&)>& controllerFactory);
-
   void setControllerSocketProtocol(c2::ControllerSocketProtocol* controller_socket_protocol) {
     controller_socket_protocol_ = controller_socket_protocol;
   }
@@ -91,6 +93,8 @@ class RootProcessGroupWrapper {
   std::unordered_map<utils::Identifier, std::unique_ptr<state::ProcessorController>> processor_to_controller_;
   std::chrono::milliseconds shutdown_check_interval_{1000};
   c2::ControllerSocketProtocol* controller_socket_protocol_{};
+  state::ComponentStateCache component_cache_;
+
   std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<RootProcessGroupWrapper>::getLogger();
 };
 

libminifi/include/core/state/ComponentStateCache.h

@@ -0,0 +1,53 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include <shared_mutex>
+#include <string>
+#include <vector>
+
+#include "utils/Id.h"
+#include "core/state/UpdateController.h"
+
+namespace org::apache::nifi::minifi::core {
+class Processor;
+class ProcessGroup;
+}  // namespace org::apache::nifi::minifi::core
+
+namespace org::apache::nifi::minifi::state {
+
+// Identity of a root process group, copied out of it whenever the root changes, so that it can be read from another
+// thread (eg. a metrics thread) without touching the process group itself.
+struct RootIdentity {
+  std::string name;
+  utils::Identifier uuid;
+  std::string version{"0"};
+};
+
+// Thread-safe snapshot of a root process group's identity and processor list, guarded by a leaf mutex: it is never
+// held while any other minifi mutex is taken, so metrics threads can read it without inverting the lock order.
+struct ComponentStateCache {
+  void setRoot(core::ProcessGroup* root);
+  RootIdentity getIdentity() const;
+  std::vector<ComponentState> getProcessorStates() const;
+
+  mutable std::shared_mutex mutex;
+  std::vector<core::Processor*> processors;
+  RootIdentity identity;
+};
+
+}  // namespace org::apache::nifi::minifi::state

libminifi/include/core/state/UpdateController.h

@@ -26,6 +26,7 @@
 #include "utils/BackTrace.h"
 #include "minifi-cpp/io/InputStream.h"
 #include "utils/expected.h"
+#include "utils/Id.h"
 
 namespace org::apache::nifi::minifi::state {
 
@@ -116,61 +117,39 @@ class StateController : public Pausable {
   [[nodiscard]] virtual std::string getComponentName() const = 0;
 
   [[nodiscard]] virtual utils::Identifier getComponentUUID() const = 0;
-  /**
-   * Start the client
-   */
+
   virtual int16_t start() = 0;
-  /**
-   * Stop the client
-   */
+
   virtual int16_t stop() = 0;
 
   virtual bool isRunning() const = 0;
 };
 
-/**
- * Justification and Purpose: Update sink is an abstract class with most functions being purely virtual.
- * This is meant to reflect the operational sink for the state of the client. Functions, below, represent
- * a small and tight interface for operations that can be performed from external controllers.
- *
- */
+struct ComponentState {
+  std::string name;
+  utils::Identifier uuid;
+  bool running = false;
+};
+
 class StateMonitor : public StateController {
  public:
   ~StateMonitor() override = default;
 
-  // Execute callback func on the component. Thread safe, locking mutex_, preventing concurrent flow update
   virtual void executeOnComponent(const std::string &id_or_name, std::function<void(state::StateController&)> func) = 0;
 
-  // Execute callback func on the all components. Thread safe, locking mutex_, preventing concurrent flow update
-  virtual void executeOnAllComponents(std::function<void(state::StateController&)> func) = 0;
+  virtual std::vector<ComponentState> getComponentStates() const = 0;
 
-  /**
-   * Drain repositories
-   */
   virtual int16_t drainRepositories() = 0;
 
-  /**
-   * Clear connection for the agent.
-   */
   virtual int16_t clearConnection(const std::string &connection) = 0;
 
   virtual std::vector<std::string> getSupportedConfigurationFormats() const = 0;
 
-  /**
-   * Apply an update with the provided string.
-   */
   virtual std::expected<void, std::string> applyUpdate(const std::string & source, const std::string &configuration,
-                                                          bool persist = false, const std::optional<std::string>& flow_id = std::nullopt) = 0;
-  /**
-   * Returns uptime for this module.
-   * @return uptime for the current state monitor.
-   */
+      bool persist = false, const std::optional<std::string>& flow_id = std::nullopt) = 0;
+
   virtual uint64_t getUptime() = 0;
 
-  /**
-   * Returns a vector of backtraces
-   * @return backtraces from the state monitor.
-   */
   virtual std::vector<BackTrace> getTraces() = 0;
 
   virtual std::map<std::string, std::unique_ptr<io::InputStream>> getDebugInfo() = 0;

libminifi/include/core/state/nodes/ResponseNodeLoader.h

@@ -57,6 +57,7 @@ class ResponseNodeLoaderImpl : public ResponseNodeLoader {
   std::vector<SharedResponseNode> getComponentMetricsNodes(const std::string& metrics_class) const;
   std::expected<SharedResponseNode, std::string> getSystemMetricsNode(const std::string& clazz);
   std::vector<SharedResponseNode> getResponseNodes(const std::string& clazz);
+  void initializeSystemMetricsNode(const SharedResponseNode& response_node) const;
   void initializeRepositoryMetrics(const SharedResponseNode& response_node) const;
   void initializeQueueMetrics(const SharedResponseNode& response_node) const;
   void initializeAgentIdentifier(const SharedResponseNode& response_node) const;
@@ -71,11 +72,9 @@ class ResponseNodeLoaderImpl : public ResponseNodeLoader {
   mutable std::mutex root_mutex_;
   mutable std::mutex component_metrics_mutex_;
   mutable std::mutex system_metrics_mutex_;
-  mutable std::mutex initialization_mutex_;
   core::ProcessGroup* root_{};
   std::unordered_map<std::string, std::vector<SharedResponseNode>> component_metrics_;
   std::unordered_map<std::string, SharedResponseNode> system_metrics_;
-  std::unordered_set<std::string> initialized_metrics_;
   std::shared_ptr<Configure> configuration_;
   std::vector<std::shared_ptr<core::RepositoryMetricsSource>> repository_metric_sources_;
   std::shared_ptr<core::FlowConfiguration> flow_configuration_;

libminifi/src/FlowController.cpp

@@ -404,14 +404,12 @@ int16_t FlowController::clearConnection(const std::string &connection) {
   return -1;
 }
 
-void FlowController::executeOnAllComponents(std::function<void(state::StateController&)> func) {
-  if (updating_.isUpdating()) {
-    return;
-  }
-  std::lock_guard<std::recursive_mutex> lock(mutex_);
-  for (auto* component: getAllComponents()) {
-    func(*component);
-  }
+std::vector<state::ComponentState> FlowController::getComponentStates() const {
+  // Deliberately does not lock mutex_: this is called from metrics threads while a metrics publisher's mutex
+  // is held, and locking mutex_ here would invert the lock order against the flow update path.
+  auto states = root_wrapper_.getProcessorStates();
+  states.push_back({.name = getComponentName(), .uuid = root_wrapper_.getComponentUUID(), .running = isRunning()});
+  return states;
 }
 
 void FlowController::executeOnComponent(const std::string &id_or_name, std::function<void(state::StateController&)> func) {
@@ -426,15 +424,6 @@ void FlowController::executeOnComponent(const std::string &id_or_name, std::func
   }
 }
 
-std::vector<state::StateController*> FlowController::getAllComponents() {
-  if (auto components = root_wrapper_.getAllProcessorControllers([this](core::Processor& p) { return createController(p); })) {
-    components->push_back(this);
-    return *components;
-  }
-
-  return {this};
-}
-
 state::StateController* FlowController::getComponent(const std::string& id_or_name) {
   if (id_or_name == getUUIDStr() || id_or_name == "FlowController") {
     return this;

libminifi/src/RootProcessGroupWrapper.cpp

@@ -25,24 +25,19 @@ void RootProcessGroupWrapper::updatePropertyValue(const std::string& processor_n
 }
 
 std::string RootProcessGroupWrapper::getName() const {
-  if (root_ != nullptr) {
-    return root_->getName();
-  }
-  return "";
+  return component_cache_.getIdentity().name;
 }
 
 utils::Identifier RootProcessGroupWrapper::getComponentUUID() const {
-  if (!root_) {
-    return {};
-  }
-  return root_->getUUID();
+  return component_cache_.getIdentity().uuid;
 }
 
 std::string RootProcessGroupWrapper::getVersion() const {
-  if (root_ != nullptr) {
-    return std::to_string(root_->getVersion());
-  }
-  return "0";
+  return component_cache_.getIdentity().version;
+}
+
+std::vector<state::ComponentState> RootProcessGroupWrapper::getProcessorStates() const {
+  return component_cache_.getProcessorStates();
 }
 
 void RootProcessGroupWrapper::setNewRoot(std::unique_ptr<core::ProcessGroup> new_root) {
@@ -55,9 +50,11 @@ void RootProcessGroupWrapper::setNewRoot(std::unique_ptr<core::ProcessGroup> new
   if (controller_socket_protocol_) {
     controller_socket_protocol_->setRoot(nullptr);
   }
+  component_cache_.setRoot(nullptr);
   backup_root_ = std::move(root_);
   root_ = std::move(new_root);
   processor_to_controller_.clear();
+  component_cache_.setRoot(root_.get());
   if (metrics_publisher_store_) { metrics_publisher_store_->loadMetricNodes(root_.get()); }
   if (controller_socket_protocol_) {
     controller_socket_protocol_->setRoot(root_.get());
@@ -66,9 +63,17 @@ void RootProcessGroupWrapper::setNewRoot(std::unique_ptr<core::ProcessGroup> new
 
 void RootProcessGroupWrapper::restoreBackup() {
   if (metrics_publisher_store_) { metrics_publisher_store_->clearMetricNodes(); }
+  if (controller_socket_protocol_) {
+    controller_socket_protocol_->setRoot(nullptr);
+  }
+  component_cache_.setRoot(nullptr);
   root_ = std::move(backup_root_);
   processor_to_controller_.clear();
+  component_cache_.setRoot(root_.get());
   if (metrics_publisher_store_) { metrics_publisher_store_->loadMetricNodes(root_.get()); }
+  if (controller_socket_protocol_) {
+    controller_socket_protocol_->setRoot(root_.get());
+  }
 }
 
 void RootProcessGroupWrapper::clearBackup() {
@@ -140,27 +145,6 @@ void RootProcessGroupWrapper::clearConnection(const std::string &connection) {
   }
 }
 
-std::optional<std::vector<state::StateController*>> RootProcessGroupWrapper::getAllProcessorControllers(
-    const std::function<gsl::not_null<std::unique_ptr<state::ProcessorController>>(core::Processor&)>& controllerFactory) {
-  if (!root_) {
-    return std::nullopt;
-  }
-  std::vector<state::StateController*> controllerVec;
-  std::vector<core::Processor*> processorVec;
-  root_->getAllProcessors(processorVec);
-
-  for (const auto& processor : processorVec) {
-    // reference to the existing or newly created controller
-    auto& controller = processor_to_controller_[processor->getUUID()];
-    if (!controller) {
-      controller = controllerFactory(*processor);
-    }
-    controllerVec.push_back(controller.get());
-  }
-
-  return controllerVec;
-}
-
 state::StateController* RootProcessGroupWrapper::getProcessorController(const std::string& id_or_name,
     const std::function<gsl::not_null<std::unique_ptr<state::ProcessorController>>(core::Processor&)>& controllerFactory) {
   if (!root_) {
@@ -170,7 +154,8 @@ state::StateController* RootProcessGroupWrapper::getProcessorController(const st
   return utils::Identifier::parse(id_or_name)
     | utils::andThen([this](utils::Identifier id) { return utils::optional_from_ptr(root_->findProcessorById(id)); })
     | utils::orElse([this, &id_or_name] { return utils::optional_from_ptr(root_->findProcessorByName(id_or_name)); })
-    | utils::transform([this, &controllerFactory](gsl::not_null<core::Processor*> proc) -> gsl::not_null<state::ProcessorController*> {
+    | utils::transform([this, &controllerFactory](core::Processor* proc) -> gsl::not_null<state::ProcessorController*> {
+      gsl_Expects(proc != nullptr);
       return utils::optional_from_ptr(processor_to_controller_[proc->getUUID()].get())
           | utils::valueOrElse([this, proc, &controllerFactory] {
             return gsl::make_not_null((processor_to_controller_[proc->getUUID()] = controllerFactory(*proc)).get());

libminifi/src/c2/ControllerSocketProtocol.cpp

@@ -271,17 +271,14 @@ void ControllerSocketProtocol::writeQueueSizesResponse(io::BaseStream &stream) {
 }
 
 void ControllerSocketProtocol::writeComponentsResponse(io::BaseStream &stream) {
-  std::vector<std::pair<std::string, bool>> components;
-  update_sink_.executeOnAllComponents([&components](state::StateController& component) {
-    components.emplace_back(component.getComponentName(), component.isRunning());
-  });
+  const auto components = update_sink_.getComponentStates();
   io::BufferStream resp;
   auto op = static_cast<uint8_t>(Operation::describe);
   resp.write(&op, 1);
   resp.write(gsl::narrow<uint16_t>(components.size()));
-  for (const auto& [name, is_running] : components) {
-    resp.write(name);
-    resp.write(is_running ? "true" : "false");
+  for (const auto& component : components) {
+    resp.write(component.name);
+    resp.write(component.running ? "true" : "false");
   }
 
   stream.write(resp.getBuffer());

libminifi/src/c2/FlowStatusBuilder.cpp

@@ -219,6 +219,8 @@ std::expected<void, std::string> FlowStatusBuilder::addInstanceStatus(rapidjson:
     bulletins = bulletin_store_->getBulletins(std::chrono::minutes(5));
   }
 
+  std::lock_guard<std::mutex> guard(root_mutex_);
+
   if (options.contains(FlowStatusQueryOption::health)) {
     instance_status.AddMember("instanceHealth", rapidjson::Value(rapidjson::kObjectType), allocator);
     uint64_t queued_count = 0;