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

   1 /* Icinga 2 | (c) 2012 Icinga GmbH | GPLv2+ */
   2
   3 #include "base/socketevents.hpp"
   4 #include "base/exception.hpp"
   5 #include "base/logger.hpp"
   6 #include "base/application.hpp"
   7 #include "base/scriptglobal.hpp"
   8 #include "base/utility.hpp"
   9 #include <boost/thread/once.hpp>
  10 #include <map>
  11 #ifdef __linux__
  12 #       include <sys/epoll.h>
  13 #endif /* __linux__ */
  14
  15 using namespace icinga;
  16
  17 static boost::once_flag l_SocketIOOnceFlag = BOOST_ONCE_INIT;
  18 static SocketEventEngine *l_SocketIOEngine;
  19
  20 int SocketEvents::m_NextID = 0;
  21
  22 void SocketEventEngine::Start()
  23 {
  24         for (int tid = 0; tid < SOCKET_IOTHREADS; tid++) {
  25                 Socket::SocketPair(m_EventFDs[tid]);
  26
  27                 Utility::SetNonBlockingSocket(m_EventFDs[tid][0]);
  28                 Utility::SetNonBlockingSocket(m_EventFDs[tid][1]);
  29
  30 #ifndef _WIN32
  31                 Utility::SetCloExec(m_EventFDs[tid][0]);
  32                 Utility::SetCloExec(m_EventFDs[tid][1]);
  33 #endif /* _WIN32 */
  34
  35                 InitializeThread(tid);
  36
  37                 m_Threads[tid] = std::thread(std::bind(&SocketEventEngine::ThreadProc, this, tid));
  38         }
  39 }
  40
  41 void SocketEventEngine::WakeUpThread(int sid, bool wait)
  42 {
  43         int tid = sid % SOCKET_IOTHREADS;
  44
  45         if (std::this_thread::get_id() == m_Threads[tid].get_id())
  46                 return;
  47
  48         if (wait) {
  49                 boost::mutex::scoped_lock lock(m_EventMutex[tid]);
  50
  51                 m_FDChanged[tid] = true;
  52
  53                 while (m_FDChanged[tid]) {
  54                         (void) send(m_EventFDs[tid][1], "T", 1, 0);
  55
  56                         boost::system_time const timeout = boost::get_system_time() + boost::posix_time::milliseconds(50);
  57                         m_CV[tid].timed_wait(lock, timeout);
  58                 }
  59         } else {
  60                 (void) send(m_EventFDs[tid][1], "T", 1, 0);
  61         }
  62 }
  63
  64 void SocketEvents::InitializeEngine()
  65 {
  66         String eventEngine = Configuration::EventEngine;
  67
  68         if (eventEngine.IsEmpty())
  69 #ifdef __linux__
  70                 eventEngine = "epoll";
  71 #else /* __linux__ */
  72                 eventEngine = "poll";
  73 #endif /* __linux__ */
  74
  75         if (eventEngine == "poll")
  76                 l_SocketIOEngine = new SocketEventEnginePoll();
  77 #ifdef __linux__
  78         else if (eventEngine == "epoll")
  79                 l_SocketIOEngine = new SocketEventEngineEpoll();
  80 #endif /* __linux__ */
  81         else {
  82                 Log(LogWarning, "SocketEvents")
  83                         << "Invalid event engine selected: " << eventEngine << " - Falling back to 'poll'";
  84
  85                 eventEngine = "poll";
  86
  87                 l_SocketIOEngine = new SocketEventEnginePoll();
  88         }
  89
  90         l_SocketIOEngine->Start();
  91
  92         Configuration::EventEngine = eventEngine;
  93 }
  94
  95 /**
  96  * Constructor for the SocketEvents class.
  97  */
  98 SocketEvents::SocketEvents(const Socket::Ptr& socket)
  99         : m_ID(m_NextID++), m_FD(socket->GetFD()), m_EnginePrivate(nullptr)
 100 {
 101         boost::call_once(l_SocketIOOnceFlag, &SocketEvents::InitializeEngine);
 102
 103         Register();
 104 }
 105
 106 SocketEvents::~SocketEvents()
 107 {
 108         VERIFY(m_FD == INVALID_SOCKET);
 109 }
 110
 111 void SocketEvents::Register()
 112 {
 113         l_SocketIOEngine->Register(this);
 114 }
 115
 116 void SocketEvents::Unregister()
 117 {
 118         l_SocketIOEngine->Unregister(this);
 119 }
 120
 121 void SocketEvents::ChangeEvents(int events)
 122 {
 123         l_SocketIOEngine->ChangeEvents(this, events);
 124 }
 125
 126 boost::mutex& SocketEventEngine::GetMutex(int tid)
 127 {
 128         return m_EventMutex[tid];
 129 }
 130
 131 bool SocketEvents::IsHandlingEvents() const
 132 {
 133         int tid = m_ID % SOCKET_IOTHREADS;
 134         boost::mutex::scoped_lock lock(l_SocketIOEngine->GetMutex(tid));
 135         return m_Events;
 136 }
 137
 138 void SocketEvents::OnEvent(int revents)
 139 {
 140
 141 }
 142