/* Icinga 2 | (c) 2012 Icinga GmbH | GPLv2+ */ #ifndef IO_ENGINE_H #define IO_ENGINE_H #include "base/atomic.hpp" #include "base/debug.hpp" #include "base/exception.hpp" #include "base/lazy-init.hpp" #include "base/logger.hpp" #include "base/shared.hpp" #include #include #include #include #include #include #include #include #include #include #include namespace icinga { /** * Scope lock for CPU-bound work done in an I/O thread * * @ingroup base */ class CpuBoundWork { public: CpuBoundWork(boost::asio::yield_context yc); CpuBoundWork(const CpuBoundWork&) = delete; CpuBoundWork(CpuBoundWork&&) = delete; CpuBoundWork& operator=(const CpuBoundWork&) = delete; CpuBoundWork& operator=(CpuBoundWork&&) = delete; ~CpuBoundWork(); void Done(); private: bool m_Done; }; /** * Scope break for CPU-bound work done in an I/O thread * * @ingroup base */ class IoBoundWorkSlot { public: IoBoundWorkSlot(boost::asio::yield_context yc); IoBoundWorkSlot(const IoBoundWorkSlot&) = delete; IoBoundWorkSlot(IoBoundWorkSlot&&) = delete; IoBoundWorkSlot& operator=(const IoBoundWorkSlot&) = delete; IoBoundWorkSlot& operator=(IoBoundWorkSlot&&) = delete; ~IoBoundWorkSlot(); private: boost::asio::yield_context yc; }; /** * Async I/O engine * * @ingroup base */ class IoEngine { friend CpuBoundWork; friend IoBoundWorkSlot; public: IoEngine(const IoEngine&) = delete; IoEngine(IoEngine&&) = delete; IoEngine& operator=(const IoEngine&) = delete; IoEngine& operator=(IoEngine&&) = delete; ~IoEngine(); static IoEngine& Get(); boost::asio::io_context& GetIoContext(); static inline size_t GetCoroutineStackSize() { #ifdef _WIN32 // Increase the stack size for Windows coroutines to prevent exception corruption. // Rationale: Low cost Windows agent only & https://github.com/Icinga/icinga2/issues/7431 return 8 * 1024 * 1024; #else /* _WIN32 */ // Increase the stack size for Linux/Unix coroutines for many JSON objects on the stack. // This may help mitigate possible stack overflows. https://github.com/Icinga/icinga2/issues/7532 return 256 * 1024; //return boost::coroutines::stack_allocator::traits_type::default_size(); // Default 64 KB #endif /* _WIN32 */ } template static void SpawnCoroutine(Handler& h, Function f) { boost::asio::spawn(h, [f](boost::asio::yield_context yc) { try { f(yc); } catch (const boost::coroutines::detail::forced_unwind &) { // Required for proper stack unwinding when coroutines are destroyed. // https://github.com/boostorg/coroutine/issues/39 throw; } catch (const std::exception& ex) { Log(LogCritical, "IoEngine") << "Exception in coroutine: " << DiagnosticInformation(ex); } catch (...) { Log(LogCritical, "IoEngine", "Exception in coroutine!"); } }, boost::coroutines::attributes(GetCoroutineStackSize()) // Set a pre-defined stack size. ); } static inline void YieldCurrentCoroutine(boost::asio::yield_context yc) { Get().m_AlreadyExpiredTimer.async_wait(yc); } private: IoEngine(); void RunEventLoop(); static LazyInit> m_Instance; boost::asio::io_context m_IoContext; boost::asio::executor_work_guard m_KeepAlive; std::vector m_Threads; boost::asio::deadline_timer m_AlreadyExpiredTimer; std::atomic_int_fast32_t m_CpuBoundSemaphore; }; class TerminateIoThread : public std::exception { }; /** * Condition variable which doesn't block I/O threads * * @ingroup base */ class AsioConditionVariable { public: AsioConditionVariable(boost::asio::io_context& io, bool init = false); void Set(); void Clear(); void Wait(boost::asio::yield_context yc); private: boost::asio::deadline_timer m_Timer; }; /** * I/O timeout emulator * * This class provides a workaround for Boost.ASIO's lack of built-in timeout support. * While Boost.ASIO handles asynchronous operations, it does not natively support timeouts for these operations. * This class uses a boost::asio::deadline_timer to emulate a timeout by scheduling a callback to be triggered * after a specified duration, effectively adding timeout behavior where none exists. * The callback is executed within the provided strand, ensuring thread-safety. * * The constructor returns immediately after scheduling the timeout callback. * The callback itself is invoked asynchronously when the timeout occurs. * This allows the caller to continue execution while the timeout is running in the background. * * The class provides a Cancel() method to unschedule any pending callback. If the callback has already been run, * calling Cancel() has no effect. This method can be used to abort the timeout early if the monitored operation * completes before the callback has been run. The Timeout destructor also automatically cancels any pending callback. * A callback is considered pending even if the timeout has already expired, * but the callback has not been executed yet due to a busy strand. * * @ingroup base */ class Timeout { public: using Timer = boost::asio::deadline_timer; /** * Schedules onTimeout to be triggered after timeoutFromNow on strand. * * @param strand The strand in which the callback will be executed. * The caller must also run in this strand, as well as Cancel() and the destructor! * @param timeoutFromNow The duration after which the timeout callback will be triggered. * @param onTimeout The callback to invoke when the timeout occurs. */ template Timeout(boost::asio::io_context::strand& strand, const Timer::duration_type& timeoutFromNow, OnTimeout onTimeout) : m_Timer(strand.context(), timeoutFromNow), m_Cancelled(Shared>::Make(false)) { VERIFY(strand.running_in_this_thread()); m_Timer.async_wait(boost::asio::bind_executor( strand, [cancelled = m_Cancelled, onTimeout = std::move(onTimeout)](boost::system::error_code ec) { if (!ec && !cancelled->load()) { onTimeout(); } } )); } Timeout(const Timeout&) = delete; Timeout(Timeout&&) = delete; Timeout& operator=(const Timeout&) = delete; Timeout& operator=(Timeout&&) = delete; /** * Cancels any pending timeout callback. * * Must be called in the strand in which the callback was scheduled! */ ~Timeout() { Cancel(); } void Cancel(); private: Timer m_Timer; /** * Indicates whether the Timeout has been cancelled. * * This must be Shared<> between the lambda in the constructor and Cancel() for the case * the destructor calls Cancel() while the lambda is already queued in the strand. * The whole Timeout instance can't be kept alive by the lambda because this would delay the destructor. */ Shared>::Ptr m_Cancelled; }; } #endif /* IO_ENGINE_H */