You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by fg...@apache.org on 2026/05/12 18:48:37 UTC
(nifi-minifi-cpp) 02/03: MINIFICPP-2791 Fix deadlock during C2 flow update
This is an automated email from the ASF dual-hosted git repository.
fgerlits pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 037cda25be679b8e0d05238b976a13b22df383ba
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Tue May 12 17:52:06 2026 +0200
MINIFICPP-2791 Fix deadlock during C2 flow update
Signed-off-by: Ferenc Gerlits <fg...@gmail.com>
Closes #2163
---
core-framework/src/utils/CallBackTimer.cpp | 1 -
libminifi/include/FlowController.h | 1 -
libminifi/include/core/ProcessGroup.h | 1 +
libminifi/src/FlowController.cpp | 20 ++++++-------
libminifi/src/core/ProcessGroup.cpp | 45 ++++++++++++++++++++----------
5 files changed, 41 insertions(+), 27 deletions(-)
diff --git a/core-framework/src/utils/CallBackTimer.cpp b/core-framework/src/utils/CallBackTimer.cpp
index 70e26dc7a..3d1270063 100644
--- a/core-framework/src/utils/CallBackTimer.cpp
+++ b/core-framework/src/utils/CallBackTimer.cpp
@@ -26,7 +26,6 @@ CallBackTimer::CallBackTimer(std::chrono::milliseconds interval, const std::func
CallBackTimer::~CallBackTimer() {
stop();
- std::lock_guard<std::mutex> guard(mtx_);
if (thd_.joinable()) {
thd_.join();
}
diff --git a/libminifi/include/FlowController.h b/libminifi/include/FlowController.h
index 5473dd6d0..00ff3f756 100644
--- a/libminifi/include/FlowController.h
+++ b/libminifi/include/FlowController.h
@@ -168,7 +168,6 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi
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);
- std::unique_ptr<core::ProcessGroup> updateFromPayload(const std::string& url, const std::string& config_payload, const std::optional<std::string>& flow_id = std::nullopt);
template <typename T, typename = typename std::enable_if<std::is_base_of<SchedulingAgent, T>::value>::type>
void conditionalReloadScheduler(std::unique_ptr<T>& scheduler, const bool condition) {
diff --git a/libminifi/include/core/ProcessGroup.h b/libminifi/include/core/ProcessGroup.h
index f21068e02..06a3ef4df 100644
--- a/libminifi/include/core/ProcessGroup.h
+++ b/libminifi/include/core/ProcessGroup.h
@@ -239,6 +239,7 @@ class ProcessGroup : public CoreComponentImpl {
ProcessGroup(const ProcessGroup &parent);
ProcessGroup &operator=(const ProcessGroup &parent);
static std::shared_ptr<utils::IdGenerator> id_generator_;
+ std::mutex on_schedule_timer_mutex_;
std::unique_ptr<utils::CallBackTimer> onScheduleTimer_;
};
} // namespace core
diff --git a/libminifi/src/FlowController.cpp b/libminifi/src/FlowController.cpp
index a6b199b98..916299805 100644
--- a/libminifi/src/FlowController.cpp
+++ b/libminifi/src/FlowController.cpp
@@ -110,7 +110,8 @@ FlowController::~FlowController() {
std::expected<void, std::string> FlowController::applyConfiguration(const std::string &source, const std::string &configurePayload, const std::optional<std::string>& flow_id) {
std::unique_ptr<core::ProcessGroup> newRoot;
try {
- newRoot = updateFromPayload(source, configurePayload, flow_id);
+ // Parse the new flow here and clear controller services later to avoid race condition with the onScheduleTimer_ callback thread
+ newRoot = flow_configuration_->updateFromPayload(source, configurePayload, flow_id);
} catch (const std::exception& ex) {
logger_->log_error("Invalid configuration payload, type: {}, what: {}", typeid(ex).name(), ex.what());
return std::unexpected{fmt::format("Invalid configuration payload, type: {}, what: {}", typeid(ex).name(), ex.what())};
@@ -130,6 +131,10 @@ std::expected<void, std::string> FlowController::applyConfiguration(const std::s
std::lock_guard<std::recursive_mutex> flow_lock(mutex_);
stop();
+ // Clear the controller services now that all timer callback threads have been stopped
+ clearControllerServices();
+ controller_service_provider_impl_ = flow_configuration_->getControllerServiceProvider();
+
root_wrapper_.setNewRoot(std::move(newRoot));
initialized_ = false;
try {
@@ -241,7 +246,10 @@ std::unique_ptr<core::ProcessGroup> FlowController::loadInitialFlow() {
logger_->log_error("Couldn't fetch flow configuration from C2 server");
return nullptr;
}
- root = updateFromPayload(*opt_flow_url, *opt_source);
+ root = flow_configuration_->updateFromPayload(*opt_flow_url, *opt_source, std::nullopt);
+ // prepare to accept the new controller service provider from flow_configuration_
+ clearControllerServices();
+ controller_service_provider_impl_ = flow_configuration_->getControllerServiceProvider();
if (root) {
logger_->log_info("Successfully fetched valid flow configuration");
if (!flow_configuration_->persist(*root)) {
@@ -504,12 +512,4 @@ std::map<std::string, std::unique_ptr<io::InputStream>> FlowController::getDebug
return debug_info;
}
-std::unique_ptr<core::ProcessGroup> FlowController::updateFromPayload(const std::string& url, const std::string& config_payload, const std::optional<std::string>& flow_id) {
- auto root = flow_configuration_->updateFromPayload(url, config_payload, flow_id);
- // prepare to accept the new controller service provider from flow_configuration_
- clearControllerServices();
- controller_service_provider_impl_ = flow_configuration_->getControllerServiceProvider();
- return root;
-}
-
} // namespace org::apache::nifi::minifi
diff --git a/libminifi/src/core/ProcessGroup.cpp b/libminifi/src/core/ProcessGroup.cpp
index c60d96cd2..36d5e8946 100644
--- a/libminifi/src/core/ProcessGroup.cpp
+++ b/libminifi/src/core/ProcessGroup.cpp
@@ -68,8 +68,11 @@ ProcessGroup::ProcessGroup(ProcessGroupType type, std::string_view name)
}
ProcessGroup::~ProcessGroup() {
- if (onScheduleTimer_) {
- onScheduleTimer_->stop();
+ {
+ std::lock_guard<std::mutex> lock(on_schedule_timer_mutex_);
+ if (onScheduleTimer_) {
+ onScheduleTimer_->stop();
+ }
}
for (auto&& connection : connections_) {
@@ -169,15 +172,19 @@ void ProcessGroup::startProcessingProcessors(TimerDrivenSchedulingAgent& timeSch
// The admin yield duration comes from the configuration, should be equal in all three schedulers
std::chrono::milliseconds admin_yield_duration = timeScheduler.getAdminYieldDuration();
- if (!onScheduleTimer_ && !failed_processors.empty() && admin_yield_duration > 0ms) {
- logger_->log_info("Retrying failed processors in {}", admin_yield_duration);
- auto func = [this, eventScheduler = &eventScheduler, cronScheduler = &cronScheduler, timeScheduler = &timeScheduler]() {
- this->startProcessingProcessors(*timeScheduler, *eventScheduler, *cronScheduler);
- };
- onScheduleTimer_ = std::make_unique<utils::CallBackTimer>(admin_yield_duration, func);
- onScheduleTimer_->start();
- } else if (failed_processors.empty() && onScheduleTimer_) {
- onScheduleTimer_->stop();
+
+ {
+ std::lock_guard<std::mutex> lock(on_schedule_timer_mutex_);
+ if (!onScheduleTimer_ && !failed_processors.empty() && admin_yield_duration > 0ms) {
+ logger_->log_info("Retrying failed processors in {}", admin_yield_duration);
+ auto func = [this, eventScheduler = &eventScheduler, cronScheduler = &cronScheduler, timeScheduler = &timeScheduler]() {
+ this->startProcessingProcessors(*timeScheduler, *eventScheduler, *cronScheduler);
+ };
+ onScheduleTimer_ = std::make_unique<utils::CallBackTimer>(admin_yield_duration, func);
+ onScheduleTimer_->start();
+ } else if (failed_processors.empty() && onScheduleTimer_) {
+ onScheduleTimer_->stop();
+ }
}
{
@@ -223,12 +230,20 @@ void ProcessGroup::stopProcessing(TimerDrivenSchedulingAgent& timeScheduler, Eve
}
}
- if (onScheduleTimer_) {
- onScheduleTimer_->stop();
+ // Stop and join the retry timer outside the lock to avoid deadlock: the timer callback
+ // (startProcessingProcessors) acquires on_schedule_timer_mutex_ at the end, so the join
+ // must not be done while holding that lock.
+ // Loop because the timer callback may create a new timer while we held the lock
+ while (true) {
+ std::unique_ptr<utils::CallBackTimer> timer_to_destroy;
+ {
+ std::lock_guard<std::mutex> lock(on_schedule_timer_mutex_);
+ if (!onScheduleTimer_) break;
+ onScheduleTimer_->stop();
+ timer_to_destroy = std::move(onScheduleTimer_); // reset onScheduleTimer_ under the lock, and destroy the timer implicitly at the end of the loop
+ }
}
- onScheduleTimer_.reset();
-
try {
// Stop all the processor node, input and output ports
for (const auto &processor : processors) {