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 <[email protected]> AuthorDate: Tue May 12 17:52:06 2026 +0200 MINIFICPP-2791 Fix deadlock during C2 flow update Signed-off-by: Ferenc Gerlits <[email protected]> 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) {
