[WIP] CheckerComponent#Stop(): wait for own Checkable#ProcessCheckResult() to finish

Author: Al2Klimov

lib/checker/checkercomponent.cpp

@@ -60,6 +60,7 @@ void CheckerComponent::OnConfigLoaded()
 void CheckerComponent::Start(bool runtimeCreated)
 {
 	ObjectImpl<CheckerComponent>::Start(runtimeCreated);
+	CheckResultProducerComponent::Start();
 
 	Log(LogInformation, "CheckerComponent")
 		<< "'" << GetName() << "' started.";
@@ -81,6 +82,7 @@ void CheckerComponent::Stop(bool runtimeRemoved)
 		m_CV.notify_all();
 	}
 
+	CheckResultProducerComponent::Stop();
 	m_ResultTimer->Stop(true);
 	m_Thread.join();
 
@@ -244,7 +246,7 @@ void CheckerComponent::ExecuteCheckHelper(const Checkable::Ptr& checkable)
 		cr->SetExecutionStart(now);
 		cr->SetExecutionEnd(now);
 
-		checkable->ProcessCheckResult(cr);
+		checkable->ProcessCheckResult(cr, this);
 
 		Log(LogCritical, "checker", output);
 	}

lib/checker/checkercomponent.hpp

@@ -46,7 +46,7 @@ struct CheckableNextCheckExtractor
 /**
  * @ingroup checker
  */
-class CheckerComponent final : public ObjectImpl<CheckerComponent>
+class CheckerComponent final : public ObjectImpl<CheckerComponent>, public CheckResultProducerComponent
 {
 public:
 	DECLARE_OBJECT(CheckerComponent);

lib/icinga/checkresult.cpp

@@ -41,3 +41,58 @@ bool UnitTestCRP::try_lock_shared() noexcept
 void UnitTestCRP::unlock_shared() noexcept
 {
 }
+
+bool CheckResultProducerComponent::try_lock_shared() noexcept
+{
+	auto state (ModifyState(
+		[](auto current) { return current.InstanceIsActive; },
+		[](auto& desired) { ++desired.ProcessingCheckResults; }
+	));
+
+	return state.InstanceIsActive;
+}
+
+void CheckResultProducerComponent::unlock_shared() noexcept
+{
+	std::unique_lock lock (m_Mutex, std::defer_lock);
+
+	auto state (ModifyState([&lock](auto& desired) {
+		--desired.ProcessingCheckResults;
+
+		if (!desired.ProcessingCheckResults && !desired.InstanceIsActive && !lock) {
+			lock.lock();
+		}
+	}));
+
+	if (!state.ProcessingCheckResults && !state.InstanceIsActive) {
+		m_CV.notify_all();
+	}
+}
+
+/**
+ * Allow processing check results.
+ */
+void CheckResultProducerComponent::Start()
+{
+	ModifyState([](auto& desired) { desired.InstanceIsActive = 1; });
+}
+
+/**
+ * Disallow processing new check results, wait for all currently processed ones to finish.
+ */
+void CheckResultProducerComponent::Stop()
+{
+	std::unique_lock lock (m_Mutex, std::defer_lock);
+
+	auto state (ModifyState([&lock](auto& desired) {
+		desired.InstanceIsActive = 0;
+
+		if (desired.ProcessingCheckResults && !lock) {
+			lock.lock();
+		}
+	}));
+
+	if (state.ProcessingCheckResults) {
+		m_CV.wait(lock, [this] { return !m_State.load().ProcessingCheckResults; });
+	}
+}

lib/icinga/checkresult.hpp

@@ -3,8 +3,12 @@
 #ifndef CHECKRESULT_H
 #define CHECKRESULT_H
 
+#include "base/atomic.hpp"
 #include "icinga/i2-icinga.hpp"
 #include "icinga/checkresult-ti.hpp"
+#include <condition_variable>
+#include <cstdint>
+#include <mutex>
 
 namespace icinga
 {
@@ -54,6 +58,58 @@ class CheckResult final : public ObjectImpl<CheckResult>
 	double CalculateLatency() const;
 };
 
+class CheckResultProducerComponent : public CheckResultProducer
+{
+public:
+	bool try_lock_shared() noexcept override;
+	void unlock_shared() noexcept override;
+
+protected:
+	void Start();
+	void Stop();
+
+private:
+	struct State
+	{
+		uint32_t InstanceIsActive = 0;
+		uint32_t ProcessingCheckResults = 0;
+	};
+
+	Atomic<State> m_State {State{}};
+	std::mutex m_Mutex;
+	std::condition_variable m_CV;
+
+	/**
+	 * Load m_State into x and if cond(x), pass x to mod by reference and try to store x back.
+	 * If m_State has changed in the meantime, repeat the process.
+	 *
+	 * @return The (not) updated m_State.
+	 */
+	template<class C, class M>
+	State ModifyState(const C& cond, const M& mod)
+	{
+		auto expected (m_State.load());
+		decltype(expected) desired;
+
+		do {
+			if (!cond(expected)) {
+				return expected;
+			}
+
+			desired = expected;
+			mod(desired);
+		} while (!m_State.compare_exchange_weak(expected, desired));
+
+		return desired;
+	}
+
+	template<class M>
+	State ModifyState(const M& mod)
+	{
+		return ModifyState([](auto) { return true; }, mod);
+	}
+};
+
 }
 
 #endif /* CHECKRESULT_H */