1 /* Icinga 2 | (c) 2012 Icinga GmbH | GPLv2+ */
2
3 #include "base/exception.hpp"
4 #include "base/io-engine.hpp"
5 #include "base/lazy-init.hpp"
6 #include "base/logger.hpp"
7 #include <exception>
8 #include <memory>
9 #include <thread>
10 #include <boost/asio/io_context.hpp>
11 #include <boost/asio/spawn.hpp>
12 #include <boost/asio/post.hpp>
13 #include <boost/date_time/posix_time/ptime.hpp>
14 #include <boost/system/error_code.hpp>
15
16 using namespace icinga;
17
CpuBoundWork(boost::asio::yield_context yc)18 CpuBoundWork::CpuBoundWork(boost::asio::yield_context yc)
19 : m_Done(false)
20 {
21 auto& ioEngine (IoEngine::Get());
22
23 for (;;) {
24 auto availableSlots (ioEngine.m_CpuBoundSemaphore.fetch_sub(1));
25
26 if (availableSlots < 1) {
27 ioEngine.m_CpuBoundSemaphore.fetch_add(1);
28 ioEngine.m_AlreadyExpiredTimer.async_wait(yc);
29 continue;
30 }
31
32 break;
33 }
34 }
35
~CpuBoundWork()36 CpuBoundWork::~CpuBoundWork()
37 {
38 if (!m_Done) {
39 IoEngine::Get().m_CpuBoundSemaphore.fetch_add(1);
40 }
41 }
42
Done()43 void CpuBoundWork::Done()
44 {
45 if (!m_Done) {
46 IoEngine::Get().m_CpuBoundSemaphore.fetch_add(1);
47
48 m_Done = true;
49 }
50 }
51
IoBoundWorkSlot(boost::asio::yield_context yc)52 IoBoundWorkSlot::IoBoundWorkSlot(boost::asio::yield_context yc)
53 : yc(yc)
54 {
55 IoEngine::Get().m_CpuBoundSemaphore.fetch_add(1);
56 }
57
~IoBoundWorkSlot()58 IoBoundWorkSlot::~IoBoundWorkSlot()
59 {
60 auto& ioEngine (IoEngine::Get());
61
62 for (;;) {
63 auto availableSlots (ioEngine.m_CpuBoundSemaphore.fetch_sub(1));
64
65 if (availableSlots < 1) {
66 ioEngine.m_CpuBoundSemaphore.fetch_add(1);
67 ioEngine.m_AlreadyExpiredTimer.async_wait(yc);
68 continue;
69 }
70
71 break;
72 }
73 }
74
__anon6c12c1070102() 75 LazyInit<std::unique_ptr<IoEngine>> IoEngine::m_Instance ([]() { return std::unique_ptr<IoEngine>(new IoEngine()); });
76
Get()77 IoEngine& IoEngine::Get()
78 {
79 return *m_Instance.Get();
80 }
81
GetIoContext()82 boost::asio::io_context& IoEngine::GetIoContext()
83 {
84 return m_IoContext;
85 }
86
IoEngine()87 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)
88 {
89 m_AlreadyExpiredTimer.expires_at(boost::posix_time::neg_infin);
90 m_CpuBoundSemaphore.store(std::thread::hardware_concurrency() * 3u / 2u);
91
92 for (auto& thread : m_Threads) {
93 thread = std::thread(&IoEngine::RunEventLoop, this);
94 }
95 }
96
~IoEngine()97 IoEngine::~IoEngine()
98 {
99 for (auto& thread : m_Threads) {
100 boost::asio::post(m_IoContext, []() {
101 throw TerminateIoThread();
102 });
103 }
104
105 for (auto& thread : m_Threads) {
106 thread.join();
107 }
108 }
109
RunEventLoop()110 void IoEngine::RunEventLoop()
111 {
112 for (;;) {
113 try {
114 m_IoContext.run();
115
116 break;
117 } catch (const TerminateIoThread&) {
118 break;
119 } catch (const std::exception& e) {
120 Log(LogCritical, "IoEngine", "Exception during I/O operation!");
121 Log(LogDebug, "IoEngine") << "Exception during I/O operation: " << DiagnosticInformation(e);
122 }
123 }
124 }
125
AsioConditionVariable(boost::asio::io_context & io,bool init)126 AsioConditionVariable::AsioConditionVariable(boost::asio::io_context& io, bool init)
127 : m_Timer(io)
128 {
129 m_Timer.expires_at(init ? boost::posix_time::neg_infin : boost::posix_time::pos_infin);
130 }
131
Set()132 void AsioConditionVariable::Set()
133 {
134 m_Timer.expires_at(boost::posix_time::neg_infin);
135 }
136
Clear()137 void AsioConditionVariable::Clear()
138 {
139 m_Timer.expires_at(boost::posix_time::pos_infin);
140 }
141
Wait(boost::asio::yield_context yc)142 void AsioConditionVariable::Wait(boost::asio::yield_context yc)
143 {
144 boost::system::error_code ec;
145 m_Timer.async_wait(yc[ec]);
146 }
147
Cancel()148 void Timeout::Cancel()
149 {
150 m_Cancelled.store(true);
151
152 boost::system::error_code ec;
153 m_Timer.cancel(ec);
154 }
155