diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-28 12:34:54 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-28 12:34:54 +0000 |
commit | 0915b3ef56dfac3113cce55a59a5765dc94976be (patch) | |
tree | a8fea11d50b4f083e1bf0f90025ece7f0824784a /lib/base/io-engine.cpp | |
parent | Initial commit. (diff) | |
download | icinga2-upstream.tar.xz icinga2-upstream.zip |
Adding upstream version 2.13.6.upstream/2.13.6upstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to '')
-rw-r--r-- | lib/base/io-engine.cpp | 154 |
1 files changed, 154 insertions, 0 deletions
diff --git a/lib/base/io-engine.cpp b/lib/base/io-engine.cpp new file mode 100644 index 0000000..c00bd78 --- /dev/null +++ b/lib/base/io-engine.cpp @@ -0,0 +1,154 @@ +/* Icinga 2 | (c) 2012 Icinga GmbH | GPLv2+ */ + +#include "base/exception.hpp" +#include "base/io-engine.hpp" +#include "base/lazy-init.hpp" +#include "base/logger.hpp" +#include <exception> +#include <memory> +#include <thread> +#include <boost/asio/io_context.hpp> +#include <boost/asio/spawn.hpp> +#include <boost/asio/post.hpp> +#include <boost/date_time/posix_time/ptime.hpp> +#include <boost/system/error_code.hpp> + +using namespace icinga; + +CpuBoundWork::CpuBoundWork(boost::asio::yield_context yc) + : m_Done(false) +{ + auto& ioEngine (IoEngine::Get()); + + for (;;) { + auto availableSlots (ioEngine.m_CpuBoundSemaphore.fetch_sub(1)); + + if (availableSlots < 1) { + ioEngine.m_CpuBoundSemaphore.fetch_add(1); + IoEngine::YieldCurrentCoroutine(yc); + continue; + } + + break; + } +} + +CpuBoundWork::~CpuBoundWork() +{ + if (!m_Done) { + IoEngine::Get().m_CpuBoundSemaphore.fetch_add(1); + } +} + +void CpuBoundWork::Done() +{ + if (!m_Done) { + IoEngine::Get().m_CpuBoundSemaphore.fetch_add(1); + + m_Done = true; + } +} + +IoBoundWorkSlot::IoBoundWorkSlot(boost::asio::yield_context yc) + : yc(yc) +{ + IoEngine::Get().m_CpuBoundSemaphore.fetch_add(1); +} + +IoBoundWorkSlot::~IoBoundWorkSlot() +{ + auto& ioEngine (IoEngine::Get()); + + for (;;) { + auto availableSlots (ioEngine.m_CpuBoundSemaphore.fetch_sub(1)); + + if (availableSlots < 1) { + ioEngine.m_CpuBoundSemaphore.fetch_add(1); + IoEngine::YieldCurrentCoroutine(yc); + continue; + } + + break; + } +} + +LazyInit<std::unique_ptr<IoEngine>> IoEngine::m_Instance ([]() { return std::unique_ptr<IoEngine>(new IoEngine()); }); + +IoEngine& IoEngine::Get() +{ + return *m_Instance.Get(); +} + +boost::asio::io_context& IoEngine::GetIoContext() +{ + return m_IoContext; +} + +IoEngine::IoEngine() : m_IoContext(), m_KeepAlive(boost::asio::make_work_guard(m_IoContext)), m_Threads(decltype(m_Threads)::size_type(std::thread::hardware_concurrency() * 2u)), m_AlreadyExpiredTimer(m_IoContext) +{ + m_AlreadyExpiredTimer.expires_at(boost::posix_time::neg_infin); + m_CpuBoundSemaphore.store(std::thread::hardware_concurrency() * 3u / 2u); + + for (auto& thread : m_Threads) { + thread = std::thread(&IoEngine::RunEventLoop, this); + } +} + +IoEngine::~IoEngine() +{ + for (auto& thread : m_Threads) { + boost::asio::post(m_IoContext, []() { + throw TerminateIoThread(); + }); + } + + for (auto& thread : m_Threads) { + thread.join(); + } +} + +void IoEngine::RunEventLoop() +{ + for (;;) { + try { + m_IoContext.run(); + + break; + } catch (const TerminateIoThread&) { + break; + } catch (const std::exception& e) { + Log(LogCritical, "IoEngine", "Exception during I/O operation!"); + Log(LogDebug, "IoEngine") << "Exception during I/O operation: " << DiagnosticInformation(e); + } + } +} + +AsioConditionVariable::AsioConditionVariable(boost::asio::io_context& io, bool init) + : m_Timer(io) +{ + m_Timer.expires_at(init ? boost::posix_time::neg_infin : boost::posix_time::pos_infin); +} + +void AsioConditionVariable::Set() +{ + m_Timer.expires_at(boost::posix_time::neg_infin); +} + +void AsioConditionVariable::Clear() +{ + m_Timer.expires_at(boost::posix_time::pos_infin); +} + +void AsioConditionVariable::Wait(boost::asio::yield_context yc) +{ + boost::system::error_code ec; + m_Timer.async_wait(yc[ec]); +} + +void Timeout::Cancel() +{ + m_Cancelled.store(true); + + boost::system::error_code ec; + m_Timer.cancel(ec); +} |