tyler.durden/icinga2
1
0
Fork 0
mirror of https://github.com/Icinga/icinga2.git synced 2025-07-28 08:04:14 +02:00
Code Issues Packages Projects Releases Wiki Activity

Fix stalled reconnection attempts for the cluster

Browse Source 
refs #8485
This commit is contained in:
Gunnar Beutner 2015-02-20 14:43:13 +01:00
parent 2a9ceffea9
commit e2815de8a6
1 changed files with 34 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
lib/base/socketevents.cpp
View File

@ -41,9 +41,12 @@ struct SocketEventDescriptor
{ } { }
}; };
static boost::thread l_SocketIOThread;
static boost::once_flag l_SocketIOOnceFlag = BOOST_ONCE_INIT; static boost::once_flag l_SocketIOOnceFlag = BOOST_ONCE_INIT;
static SOCKET l_SocketIOEventFDs[2]; static SOCKET l_SocketIOEventFDs[2];
static boost::mutex l_SocketIOMutex; static boost::mutex l_SocketIOMutex;
static boost::condition_variable l_SocketIOCV;
static bool l_SocketIOFDChanged;
static std::map<SOCKET, SocketEventDescriptor> l_SocketIOSockets; static std::map<SOCKET, SocketEventDescriptor> l_SocketIOSockets;
void SocketEvents::InitializeThread(void) void SocketEvents::InitializeThread(void)
@ -63,8 +66,7 @@ void SocketEvents::InitializeThread(void)
l_SocketIOSockets[l_SocketIOEventFDs[0]] = sed; l_SocketIOSockets[l_SocketIOEventFDs[0]] = sed;
boost::thread thread(&SocketEvents::ThreadProc); l_SocketIOThread = boost::thread(&SocketEvents::ThreadProc);
thread.detach();
} }
void SocketEvents::ThreadProc(void) void SocketEvents::ThreadProc(void)
@ -100,6 +102,16 @@ void SocketEvents::ThreadProc(void)
(void) poll(pfds, pfdcount, -1); (void) poll(pfds, pfdcount, -1);
#endif /* _WIN32 */ #endif /* _WIN32 */
{
boost::mutex::scoped_lock lock(l_SocketIOMutex);
if (l_SocketIOFDChanged) {
l_SocketIOFDChanged = false;
l_SocketIOCV.notify_all();
continue;
}
}
for (int i = 0; i < pfdcount; i++) { for (int i = 0; i < pfdcount; i++) {
if ((pfds[i].revents & (POLLIN | POLLOUT | POLLHUP | POLLERR)) == 0) if ((pfds[i].revents & (POLLIN | POLLOUT | POLLHUP | POLLERR)) == 0)
continue; continue;
@ -139,7 +151,16 @@ void SocketEvents::ThreadProc(void)
void SocketEvents::WakeUpThread(void) void SocketEvents::WakeUpThread(void)
{ {
if (boost::this_thread::get_id() != l_SocketIOThread.get_id()) {
boost::mutex::scoped_lock lock(l_SocketIOMutex);
l_SocketIOFDChanged = true;
(void) send(l_SocketIOEventFDs[1], "T", 1, 0); (void) send(l_SocketIOEventFDs[1], "T", 1, 0);
while (l_SocketIOFDChanged)
l_SocketIOCV.wait(lock);
}
} }
/** /**
@ -160,6 +181,8 @@ SocketEvents::~SocketEvents(void)
void SocketEvents::Register(Object *lifesupportObject) void SocketEvents::Register(Object *lifesupportObject)
{ {
boost::mutex::scoped_lock lock(l_SocketIOMutex);
VERIFY(m_FD != INVALID_SOCKET); VERIFY(m_FD != INVALID_SOCKET);
SocketEventDescriptor desc; SocketEventDescriptor desc;
@ -167,25 +190,21 @@ void SocketEvents::Register(Object *lifesupportObject)
desc.EventInterface = this; desc.EventInterface = this;
desc.LifesupportObject = lifesupportObject; desc.LifesupportObject = lifesupportObject;
{
boost::mutex::scoped_lock lock(l_SocketIOMutex);
VERIFY(l_SocketIOSockets.find(m_FD) == l_SocketIOSockets.end()); VERIFY(l_SocketIOSockets.find(m_FD) == l_SocketIOSockets.end());
l_SocketIOSockets[m_FD] = desc; l_SocketIOSockets[m_FD] = desc;
}
/* There's no need to wake up the I/O thread here. */ /* There's no need to wake up the I/O thread here. */
} }
void SocketEvents::Unregister(void) void SocketEvents::Unregister(void)
{ {
if (m_FD == INVALID_SOCKET)
return;
{ {
boost::mutex::scoped_lock lock(l_SocketIOMutex); boost::mutex::scoped_lock lock(l_SocketIOMutex);
if (m_FD == INVALID_SOCKET)
return;
l_SocketIOSockets.erase(m_FD); l_SocketIOSockets.erase(m_FD);
m_FD = INVALID_SOCKET; m_FD = INVALID_SOCKET;
} }
@ -195,12 +214,12 @@ void SocketEvents::Unregister(void)
void SocketEvents::ChangeEvents(int events) void SocketEvents::ChangeEvents(int events)
{ {
if (m_FD == INVALID_SOCKET)
BOOST_THROW_EXCEPTION(std::runtime_error("Tried to read/write from a closed socket."));
{ {
boost::mutex::scoped_lock lock(l_SocketIOMutex); boost::mutex::scoped_lock lock(l_SocketIOMutex);
if (m_FD == INVALID_SOCKET)
BOOST_THROW_EXCEPTION(std::runtime_error("Tried to read/write from a closed socket."));
std::map<SOCKET, SocketEventDescriptor>::iterator it = l_SocketIOSockets.find(m_FD); std::map<SOCKET, SocketEventDescriptor>::iterator it = l_SocketIOSockets.find(m_FD);
if (it == l_SocketIOSockets.end()) if (it == l_SocketIOSockets.end())