granicus.if.org Git - icinga2/blob - lib/base/io-engine.cpp

   1 /******************************************************************************
   2  * Icinga 2                                                                   *
   3  * Copyright (C) 2012-2018 Icinga Development Team (https://icinga.com/)      *
   4  *                                                                            *
   5  * This program is free software; you can redistribute it and/or              *
   6  * modify it under the terms of the GNU General Public License                *
   7  * as published by the Free Software Foundation; either version 2             *
   8  * of the License, or (at your option) any later version.                     *
   9  *                                                                            *
  10  * This program is distributed in the hope that it will be useful,            *
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of             *
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the              *
  13  * GNU General Public License for more details.                               *
  14  *                                                                            *
  15  * You should have received a copy of the GNU General Public License          *
  16  * along with this program; if not, write to the Free Software Foundation     *
  17  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA.             *
  18  ******************************************************************************/
  19
  20 #include "base/exception.hpp"
  21 #include "base/io-engine.hpp"
  22 #include "base/lazy-init.hpp"
  23 #include "base/logger.hpp"
  24 #include <exception>
  25 #include <memory>
  26 #include <thread>
  27 #include <boost/asio/io_service.hpp>
  28 #include <boost/asio/spawn.hpp>
  29 #include <boost/date_time/posix_time/ptime.hpp>
  30 #include <boost/system/error_code.hpp>
  31
  32 using namespace icinga;
  33
  34 CpuBoundWork::CpuBoundWork(boost::asio::yield_context yc)
  35         : m_Done(false)
  36 {
  37         auto& ioEngine (IoEngine::Get());
  38
  39         for (;;) {
  40                 auto availableSlots (ioEngine.m_CpuBoundSemaphore.fetch_sub(1));
  41
  42                 if (availableSlots < 1) {
  43                         ioEngine.m_CpuBoundSemaphore.fetch_add(1);
  44                         ioEngine.m_AlreadyExpiredTimer.async_wait(yc);
  45                         continue;
  46                 }
  47
  48                 break;
  49         }
  50 }
  51
  52 CpuBoundWork::~CpuBoundWork()
  53 {
  54         if (!m_Done) {
  55                 IoEngine::Get().m_CpuBoundSemaphore.fetch_add(1);
  56         }
  57 }
  58
  59 void CpuBoundWork::Done()
  60 {
  61         if (!m_Done) {
  62                 IoEngine::Get().m_CpuBoundSemaphore.fetch_add(1);
  63
  64                 m_Done = true;
  65         }
  66 }
  67
  68 IoBoundWorkSlot::IoBoundWorkSlot(boost::asio::yield_context yc)
  69         : yc(yc)
  70 {
  71         IoEngine::Get().m_CpuBoundSemaphore.fetch_add(1);
  72 }
  73
  74 IoBoundWorkSlot::~IoBoundWorkSlot()
  75 {
  76         auto& ioEngine (IoEngine::Get());
  77
  78         for (;;) {
  79                 auto availableSlots (ioEngine.m_CpuBoundSemaphore.fetch_sub(1));
  80
  81                 if (availableSlots < 1) {
  82                         ioEngine.m_CpuBoundSemaphore.fetch_add(1);
  83                         ioEngine.m_AlreadyExpiredTimer.async_wait(yc);
  84                         continue;
  85                 }
  86
  87                 break;
  88         }
  89 }
  90
  91 LazyInit<std::unique_ptr<IoEngine>> IoEngine::m_Instance ([]() { return std::unique_ptr<IoEngine>(new IoEngine()); });
  92
  93 IoEngine& IoEngine::Get()
  94 {
  95         return *m_Instance.Get();
  96 }
  97
  98 boost::asio::io_service& IoEngine::GetIoService()
  99 {
 100         return m_IoService;
 101 }
 102
 103 IoEngine::IoEngine() : m_IoService(), m_KeepAlive(m_IoService), m_Threads(decltype(m_Threads)::size_type(std::thread::hardware_concurrency() * 2u)), m_AlreadyExpiredTimer(m_IoService)
 104 {
 105         m_AlreadyExpiredTimer.expires_at(boost::posix_time::neg_infin);
 106         m_CpuBoundSemaphore.store(std::thread::hardware_concurrency() * 3u / 2u);
 107
 108         for (auto& thread : m_Threads) {
 109                 thread = std::thread(&IoEngine::RunEventLoop, this);
 110         }
 111 }
 112
 113 IoEngine::~IoEngine()
 114 {
 115         for (auto& thread : m_Threads) {
 116                 m_IoService.post([]() {
 117                         throw TerminateIoThread();
 118                 });
 119         }
 120
 121         for (auto& thread : m_Threads) {
 122                 thread.join();
 123         }
 124 }
 125
 126 void IoEngine::RunEventLoop()
 127 {
 128         for (;;) {
 129                 try {
 130                         m_IoService.run();
 131
 132                         break;
 133                 } catch (const TerminateIoThread&) {
 134                         break;
 135                 } catch (const std::exception& e) {
 136                         Log(LogCritical, "IoEngine", "Exception during I/O operation!");
 137                         Log(LogDebug, "IoEngine") << "Exception during I/O operation: " << DiagnosticInformation(e);
 138                 }
 139         }
 140 }
 141
 142 AsioConditionVariable::AsioConditionVariable(boost::asio::io_service& io, bool init)
 143         : m_Timer(io)
 144 {
 145         m_Timer.expires_at(init ? boost::posix_time::neg_infin : boost::posix_time::pos_infin);
 146 }
 147
 148 void AsioConditionVariable::Set()
 149 {
 150         m_Timer.expires_at(boost::posix_time::neg_infin);
 151 }
 152
 153 void AsioConditionVariable::Clear()
 154 {
 155         m_Timer.expires_at(boost::posix_time::pos_infin);
 156 }
 157
 158 void AsioConditionVariable::Wait(boost::asio::yield_context yc)
 159 {
 160         boost::system::error_code ec;
 161         m_Timer.async_wait(yc[ec]);
 162 }