[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();
 
@@ -232,7 +234,7 @@ void CheckerComponent::ExecuteCheckHelper(const Checkable::Ptr& checkable)
 	try {
 		checkable->ExecuteCheck();
 	} catch (const std::exception& ex) {
-		CheckResult::Ptr cr = new CheckResult();
+		CheckResult::Ptr cr = new CheckResult(this);
 		cr->SetState(ServiceUnknown);
 
 		String output = "Exception occurred while checking '" + checkable->GetName() + "': " + DiagnosticInformation(ex);

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

@@ -33,6 +33,51 @@ double CheckResult::CalculateLatency() const
 	return latency;
 }
 
+bool CheckResultProducerComponent::try_lock_shared() noexcept
+{
+	auto expected (m_State.load());
+	decltype(expected) desired;
+
+	do {
+		if (!expected.m_InstanceIsActive) {
+			return false;
+		}
+
+		desired = expected;
+		++desired.m_ProcessingCheckResults;
+	} while (!m_State.compare_exchange_weak(expected, desired));
+
+	return true;
+}
+
+void CheckResultProducerComponent::unlock_shared() noexcept
+{
+	auto state (ModifyState([](auto& desired) { --desired.m_ProcessingCheckResults; }));
+
+	if (!state.m_ProcessingCheckResults) {
+		m_CV.notify_all();
+	}
+}
+
+/**
+ * Allow processing check results.
+ */
+void CheckResultProducerComponent::Start()
+{
+	ModifyState([](auto& desired) { desired.m_InstanceIsActive = 1; });
+}
+
+/**
+ * Disallow processing new check results, wait for all currently processed ones to finish.
+ */
+void CheckResultProducerComponent::Stop()
+{
+	ModifyState([](auto& desired) { desired.m_InstanceIsActive = 0; });
+
+	std::unique_lock lock (m_Mutex);
+	m_CV.wait(lock, [this] { return !m_State.load().m_ProcessingCheckResults; });
+}
+
 ObjectFactory TypeHelper<CheckResult, false>::GetFactory()
 {
 	return &Factory;

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>
 #include <utility>
 
 namespace icinga
@@ -60,6 +64,42 @@ class CheckResult final : public ObjectImpl<CheckResult>
 	CheckResultProducer::Ptr m_Producer;
 };
 
+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 m_InstanceIsActive = 0;
+		uint32_t m_ProcessingCheckResults = 0;
+	};
+
+	Atomic<State> m_State {State{}};
+	std::mutex m_Mutex;
+	std::condition_variable m_CV;
+
+	template<class F>
+	State ModifyState(const F& func)
+	{
+		auto expected (m_State.load());
+		decltype(expected) desired;
+
+		do {
+			desired = expected;
+			func(desired);
+		} while (!m_State.compare_exchange_weak(expected, desired));
+
+		return desired;
+	}
+};
+
 template<>
 struct TypeHelper<CheckResult, false>
 {