CpuBoundWork#CpuBoundWork(): don't spin on atomic int to acquire slot

This is inefficient and involves possible bad surprises regarding waiting durations on busy nodes. Instead, use AsioConditionVariable#Wait() if there are no free slots. It's notified by others' CpuBoundWork#~CpuBoundWork() once finished.

Author: Al2Klimov

lib/base/io-engine.cpp

@@ -16,30 +16,84 @@
 
 using namespace icinga;
 
-CpuBoundWork::CpuBoundWork(boost::asio::yield_context yc, boost::asio::io_context::strand&)
+CpuBoundWork::CpuBoundWork(boost::asio::yield_context yc, boost::asio::io_context::strand& strand)
 	: m_Done(false)
 {
-	auto& ioEngine (IoEngine::Get());
+	VERIFY(strand.running_in_this_thread());
 
-	for (;;) {
-		auto availableSlots (ioEngine.m_CpuBoundSemaphore.fetch_sub(1));
+	auto& ie (IoEngine::Get());
+	Shared<AsioConditionVariable>::Ptr cv;
+
+	auto fastPath = [&ie] {
+		auto freeSlots (ie.m_CpuBoundSemaphore.load());
 
-		if (availableSlots < 1) {
-			ioEngine.m_CpuBoundSemaphore.fetch_add(1);
-			IoEngine::YieldCurrentCoroutine(yc);
-			continue;
+		while (freeSlots > 0) {
+			// In case of failure, compare_exchange_weak() updates the first argument with the current value.
+			// https://en.cppreference.com/w/cpp/atomic/atomic/compare_exchange
+			if (ie.m_CpuBoundSemaphore.compare_exchange_weak(freeSlots, freeSlots - 1)) {
+				return true;
+			}
 		}
 
-		break;
+		return false;
+	};
+
+	while (!fastPath()) {
+		if (!cv) {
+			cv = Shared<AsioConditionVariable>::Make(ie.GetIoContext());
+
+			// The above line may take a little bit, so let's optimistically re-check
+			if (fastPath()) {
+				break;
+			}
+		}
+
+		{
+			std::unique_lock lock (ie.m_CpuBoundWaitingMutex);
+
+			// The above line may take even longer, so let's check again.
+			// Also mitigate lost wake-ups by synchronizing with Done().
+			// On details, see Done() implementation inline comments.
+			if (fastPath()) {
+				break;
+			}
+
+			ie.m_CpuBoundWaiting.emplace_back(strand, cv);
+		}
+
+		cv->Wait(yc);
 	}
 }
 
 void CpuBoundWork::Done()
 {
 	if (!m_Done) {
-		IoEngine::Get().m_CpuBoundSemaphore.fetch_add(1);
-
 		m_Done = true;
+
+		auto& ie (IoEngine::Get());
+
+		// The constructor takes the slow path only if the semaphore is full,
+		// so we only have to wake up constructors if the semaphore was full.
+		// This works because after fetch_add(), fast paths in the constructor will succeed.
+		if (ie.m_CpuBoundSemaphore.fetch_add(1) < 1) {
+			// So now there are only slow path subscribers from just before the fetch_add() to be woken up.
+			// Precisely, only subscribers from just before the fetch_add() which turned 0 to 1.
+
+			decltype(ie.m_CpuBoundWaiting) subscribers;
+
+			{
+				// Due to the mutex, wake-ups obviously may happen only before or after constructor subscribtions.
+				// Ideally, this wake-up happens after any ongoing subscription (obviously).
+				std::unique_lock lock (ie.m_CpuBoundWaitingMutex);
+				std::swap(subscribers, ie.m_CpuBoundWaiting);
+			}
+			// But even if it happens before (and gets lost), it's not a problem because now the ongoing subscriber
+			// will lock the mutex and re-check the semaphore which is already >0 (fast path) due to fetch_add().
+
+			for (auto& subscriber : subscribers) {
+				boost::asio::post(subscriber.first, [cv = std::move(subscriber.second)] { cv->NotifyOne(); });
+			}
+		}
 	}
 }
 

lib/base/io-engine.hpp

@@ -6,10 +6,12 @@
 #include "base/exception.hpp"
 #include "base/lazy-init.hpp"
 #include "base/logger.hpp"
+#include "base/shared.hpp"
 #include "base/shared-object.hpp"
 #include <atomic>
 #include <exception>
 #include <memory>
+#include <mutex>
 #include <thread>
 #include <utility>
 #include <vector>
@@ -138,7 +140,10 @@ class IoEngine
 	boost::asio::executor_work_guard<boost::asio::io_context::executor_type> m_KeepAlive;
 	std::vector<std::thread> m_Threads;
 	boost::asio::deadline_timer m_AlreadyExpiredTimer;
+
 	std::atomic_int_fast32_t m_CpuBoundSemaphore;
+	std::mutex m_CpuBoundWaitingMutex;
+	std::vector<std::pair<boost::asio::io_context::strand, Shared<AsioConditionVariable>::Ptr>> m_CpuBoundWaiting;
 };
 
 class TerminateIoThread : public std::exception