Copilot commented on code in PR #2297:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2297#discussion_r4523734709
##########
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;
Review Comment:
These getters now exclusively depend on `component_cache_`. In the shown
diff, the cache is updated in `setNewRoot()`/`restoreBackup()`/destructor, but
if the wrapper ever has a non-null initial `root_` before `setNewRoot()` is
first called (e.g., during initial flow load), these APIs will return
default/empty values. Ensure `component_cache_.setRoot(root_.get())` is invoked
whenever `root_` is initially established (constructor or equivalent
initialization path), not only during subsequent root swaps.
##########
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;
Review Comment:
These getters now exclusively depend on `component_cache_`. In the shown
diff, the cache is updated in `setNewRoot()`/`restoreBackup()`/destructor, but
if the wrapper ever has a non-null initial `root_` before `setNewRoot()` is
first called (e.g., during initial flow load), these APIs will return
default/empty values. Ensure `component_cache_.setRoot(root_.get())` is invoked
whenever `root_` is initially established (constructor or equivalent
initialization path), not only during subsequent root swaps.
##########
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();
}
Review Comment:
These getters now exclusively depend on `component_cache_`. In the shown
diff, the cache is updated in `setNewRoot()`/`restoreBackup()`/destructor, but
if the wrapper ever has a non-null initial `root_` before `setNewRoot()` is
first called (e.g., during initial flow load), these APIs will return
default/empty values. Ensure `component_cache_.setRoot(root_.get())` is invoked
whenever `root_` is initially established (constructor or equivalent
initialization path), not only during subsequent root swaps.
##########
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;
+};
Review Comment:
The comment claims the cache mutex is a strict 'leaf' that is never held
while other MiNiFi locks are taken, but `getProcessorStates()` (in the
corresponding .cpp) calls `processor->getName()`, `getUUID()`, and
`isRunning()` while holding the cache lock. If any of those acquire internal
locks, this guarantee is no longer true and could reintroduce lock-order
constraints. Either (mandatory) relax/adjust the comment to match the actual
locking behavior, or (optional but safer) redesign the cache to avoid calling
into `Processor` while holding the cache lock (e.g., store copied `name/uuid`
plus a safe running indicator, or store shared ownership to make it safe to
release the cache lock before dereferencing).
##########
libminifi/test/flow-tests/MetricsDuringReloadTests.cpp:
##########
@@ -0,0 +1,190 @@
+/**
+ * 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.
+ */
+#include <array>
+#include <atomic>
+#include <chrono>
+#include <memory>
+#include <ranges>
+#include <string>
+#include <string_view>
+#include <thread>
+#include <unordered_set>
+#include <vector>
+
+#include "c2/C2MetricsPublisher.h"
+#include "c2/C2Utils.h"
+#include "core/state/MetricsPublisherStore.h"
+#include "properties/Configuration.h"
+#include "CustomProcessors.h"
+#include "unit/Catch.h"
+#include "unit/TestControllerWithFlow.h"
+
+namespace {
+
+constexpr std::string_view SinkProcessorName = "Sink Processor";
+
+std::string createFlow(const std::string& generator_processor_name) {
+ return R"(
+Flow Controller:
+ name: MiNiFi Flow
+Processors:
+- name: )" + generator_processor_name + R"(
+ id: 2438e3c8-015a-1000-79ca-83af40ec1991
+ class: org.apache.nifi.processors.standard.TestFlowFileGenerator
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 10 ms
+ auto-terminated relationships list:
+ Properties:
+ Batch Size: 1
+- name: )" + std::string{SinkProcessorName} + R"(
+ id: 2438e3c8-015a-1000-79ca-83af40ec1992
+ class: org.apache.nifi.processors.standard.TestProcessor
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 10 ms
+ auto-terminated relationships list:
+ - apple
+ - banana
+Connections:
+- name: GeneratorToSink
+ id: 2438e3c8-015a-1000-79ca-83af40ec1997
+ source name: )" + generator_processor_name + R"(
+ source id: 2438e3c8-015a-1000-79ca-83af40ec1991
+ source relationship name: success
+ destination name: )" + std::string{SinkProcessorName} + R"(
+ destination id: 2438e3c8-015a-1000-79ca-83af40ec1992
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 sec
+Remote Processing Groups: []
+Controller Services: []
+)";
+}
+
+constexpr std::string_view UnparseableFlow = "this is not a valid flow
configuration";
+
+constexpr std::array DescribedClasses = {"FlowInformation", "QueueMetrics",
"processorMetrics/.*"};
+
+std::unique_ptr<TestControllerWithFlow> createTestController(const
std::string& flow) {
+ auto test_controller =
std::make_unique<TestControllerWithFlow>(flow.c_str(), /* setup_flow = */
false);
+
test_controller->configuration_->set(minifi::Configuration::nifi_c2_root_classes,
+ "DeviceInfoNode,AgentInformation,FlowInformation,QueueMetrics");
+ test_controller->setupFlow();
+ return test_controller;
+}
+
+std::unordered_set<std::string> processorNames(const
std::vector<minifi::state::ComponentState>& components) {
+ REQUIRE_FALSE(components.empty());
+ CHECK(components.back().name == "FlowController");
+ return components | std::views::take(components.size() - 1)
+ | std::views::transform([](const auto& component) { return
component.name; })
+ | std::ranges::to<std::unordered_set>();
Review Comment:
`std::ranges::to<...>()` is a C++23 feature; if the project is built as
C++20 (or earlier), this will not compile. Consider replacing it with an
explicit loop that inserts into `std::unordered_set<std::string>` (or using an
in-project utility) to keep the test compatible with the repository’s
configured language standard.
##########
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;
Review Comment:
Locking `root_mutex_` for the entire remainder of `addInstanceStatus()` can
increase contention with flow reload/root updates, since JSON construction and
option handling can be relatively expensive. Prefer narrowing the lock scope to
only the minimal sections that access `root_` (or copy required root data under
the lock, then build the JSON after releasing it).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
