Skip to content

Commit

Permalink
CpuBoundWork#CpuBoundWork(): don't spin on atomic int to acquire slot
Browse files Browse the repository at this point in the history
This is inefficient and involves unfair scheduling. The latter implies
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.
  • Loading branch information
Al2Klimov committed Feb 20, 2024
1 parent b899611 commit bf74280
Show file tree
Hide file tree
Showing 2 changed files with 60 additions and 13 deletions.
59 changes: 48 additions & 11 deletions lib/base/io-engine.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -16,28 +16,61 @@

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());
auto& sem (ioEngine.m_CpuBoundSemaphore);
std::unique_lock<std::mutex> lock (sem.Mutex);

for (;;) {
auto availableSlots (ioEngine.m_CpuBoundSemaphore.fetch_sub(1));
if (sem.FreeSlots) {
--sem.FreeSlots;
return;
}

if (availableSlots < 1) {
ioEngine.m_CpuBoundSemaphore.fetch_add(1);
IoEngine::YieldCurrentCoroutine(yc);
continue;
}
AsioConditionVariable cv (ioEngine.GetIoContext());

sem.Waiting.emplace(&strand, &cv);
lock.unlock();

break;
try {
cv.Wait(yc);
} catch (...) {
Done();
throw;
}
}

class SetAsioCV
{
public:
inline SetAsioCV(AsioConditionVariable& cv) : m_CV(&cv)
{
}

inline void operator()()
{
m_CV->Set();
}

private:
AsioConditionVariable* m_CV;
};

void CpuBoundWork::Done()
{
if (!m_Done) {
IoEngine::Get().m_CpuBoundSemaphore.fetch_add(1);
auto& sem (IoEngine::Get().m_CpuBoundSemaphore);
std::unique_lock<std::mutex> lock (sem.Mutex);

if (sem.Waiting.empty()) {
++sem.FreeSlots;
} else {
auto next (sem.Waiting.front());
sem.Waiting.pop();

boost::asio::post(*next.first, SetAsioCV(*next.second));
}

m_Done = true;
}
Expand All @@ -58,7 +91,11 @@ boost::asio::io_context& IoEngine::GetIoContext()
IoEngine::IoEngine() : m_IoContext(), m_KeepAlive(boost::asio::make_work_guard(m_IoContext)), m_Threads(decltype(m_Threads)::size_type(Configuration::Concurrency * 2u)), m_AlreadyExpiredTimer(m_IoContext)
{
m_AlreadyExpiredTimer.expires_at(boost::posix_time::neg_infin);
m_CpuBoundSemaphore.store(Configuration::Concurrency * 3u / 2u);

{
std::unique_lock<std::mutex> lock (m_CpuBoundSemaphore.Mutex);
m_CpuBoundSemaphore.FreeSlots = Configuration::Concurrency * 3u / 2u;
}

for (auto& thread : m_Threads) {
thread = std::thread(&IoEngine::RunEventLoop, this);
Expand Down
14 changes: 12 additions & 2 deletions lib/base/io-engine.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,11 @@
#include "base/logger.hpp"
#include "base/shared-object.hpp"
#include <atomic>
#include <cstdint>
#include <exception>
#include <memory>
#include <mutex>
#include <queue>
#include <thread>
#include <utility>
#include <vector>
Expand All @@ -31,7 +34,7 @@ namespace icinga
class CpuBoundWork
{
public:
CpuBoundWork(boost::asio::yield_context yc, boost::asio::io_context::strand&);
CpuBoundWork(boost::asio::yield_context yc, boost::asio::io_context::strand& strand);
CpuBoundWork(const CpuBoundWork&) = delete;
CpuBoundWork(CpuBoundWork&&) = delete;
CpuBoundWork& operator=(const CpuBoundWork&) = delete;
Expand All @@ -48,6 +51,8 @@ class CpuBoundWork
bool m_Done;
};

class AsioConditionVariable;

/**
* Async I/O engine
*
Expand Down Expand Up @@ -120,7 +125,12 @@ 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;

struct {
std::mutex Mutex;
uint_fast32_t FreeSlots;
std::queue<std::pair<boost::asio::io_context::strand*, AsioConditionVariable*>> Waiting;
} m_CpuBoundSemaphore;
};

class TerminateIoThread : public std::exception
Expand Down

0 comments on commit bf74280

Please sign in to comment.