tyler.durden
/
icinga2
mirror of https://github.com/Icinga/icinga2.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Ensure that SetCorked() works properly

This commit is contained in:
Gunnar Beutner 2018-03-06 08:49:43 +01:00
parent c16dbd19c8
commit 9cd5298d8b
4 changed files with 22 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
lib/base/socketevents-poll.cpp
View File

@ -57,11 +57,17 @@ void SocketEventEnginePoll::ThreadProc(int tid)
if (desc.second.Events == 0) if (desc.second.Events == 0)
continue; continue;
if (desc.second.EventInterface) int events = desc.second.Events;
if (desc.second.EventInterface) {
desc.second.EventInterface->m_EnginePrivate = &pfds[i]; desc.second.EventInterface->m_EnginePrivate = &pfds[i];
if (!desc.second.EventInterface->m_Events)
events = 0;
}
pfds[i].fd = desc.first; pfds[i].fd = desc.first;
pfds[i].events = desc.second.Events; pfds[i].events = events;
descriptors[i] = desc.second; descriptors[i] = desc.second;
i++; i++;

8
lib/base/tlsstream.cpp
View File

@ -173,6 +173,8 @@ void TlsStream::OnEvent(int revents)
*/ */
ERR_clear_error(); ERR_clear_error();
size_t readTotal = 0;
switch (m_CurrentAction) { switch (m_CurrentAction) {
case TlsActionRead: case TlsActionRead:
do { do {
@ -181,8 +183,10 @@ void TlsStream::OnEvent(int revents)
if (rc > 0) { if (rc > 0) {
m_RecvQ->Write(buffer, rc); m_RecvQ->Write(buffer, rc);
success = true; success = true;
readTotal += rc;
} }
} while (rc > 0); } while (rc > 0 && readTotal < 64 * 1024);
if (success) if (success)
m_CV.notify_all(); m_CV.notify_all();
@ -264,7 +268,7 @@ void TlsStream::OnEvent(int revents)
lock.unlock(); lock.unlock();
while (m_RecvQ->IsDataAvailable() && IsHandlingEvents()) while (!IsCorked() && m_RecvQ->IsDataAvailable() && IsHandlingEvents())
SignalDataAvailable(); SignalDataAvailable();
} }

6
lib/remote/httpserverconnection.cpp
View File

@ -173,8 +173,6 @@ bool HttpServerConnection::ProcessMessage()
return res; return res;
} }
m_Stream->SetCorked(true);
m_RequestQueue.Enqueue(std::bind(&HttpServerConnection::ProcessMessageAsync, m_RequestQueue.Enqueue(std::bind(&HttpServerConnection::ProcessMessageAsync,
HttpServerConnection::Ptr(this), m_CurrentRequest, response, m_AuthenticatedUser)); HttpServerConnection::Ptr(this), m_CurrentRequest, response, m_AuthenticatedUser));
@ -347,6 +345,8 @@ void HttpServerConnection::DataAvailableHandler()
if (!m_Stream->IsEof()) { if (!m_Stream->IsEof()) {
boost::mutex::scoped_lock lock(m_DataHandlerMutex); boost::mutex::scoped_lock lock(m_DataHandlerMutex);
m_Stream->SetCorked(true);
try { try {
while (ProcessMessage()) while (ProcessMessage())
; /* empty loop body */ ; /* empty loop body */
@ -356,6 +356,8 @@ void HttpServerConnection::DataAvailableHandler()
close = true; close = true;
} }
m_RequestQueue.Enqueue(std::bind(&Stream::SetCorked, m_Stream, false));
} else } else
close = true; close = true;

8
lib/remote/jsonrpcconnection.cpp
View File

@ -155,8 +155,6 @@ void JsonRpcConnection::MessageHandlerWrapper(const String& jsonString)
try { try {
MessageHandler(jsonString); MessageHandler(jsonString);
m_Stream->SetCorked(false);
} catch (const std::exception& ex) { } catch (const std::exception& ex) {
Log(LogWarning, "JsonRpcConnection") Log(LogWarning, "JsonRpcConnection")
<< "Error while reading JSON-RPC message for identity '" << m_Identity << "Error while reading JSON-RPC message for identity '" << m_Identity
@ -262,8 +260,6 @@ bool JsonRpcConnection::ProcessMessage()
if (srs != StatusNewItem) if (srs != StatusNewItem)
return false; return false;
m_Stream->SetCorked(true);
l_JsonRpcConnectionWorkQueues[m_ID % l_JsonRpcConnectionWorkQueueCount].Enqueue(std::bind(&JsonRpcConnection::MessageHandlerWrapper, JsonRpcConnection::Ptr(this), message)); l_JsonRpcConnectionWorkQueues[m_ID % l_JsonRpcConnectionWorkQueueCount].Enqueue(std::bind(&JsonRpcConnection::MessageHandlerWrapper, JsonRpcConnection::Ptr(this), message));
return true; return true;
@ -279,6 +275,8 @@ void JsonRpcConnection::DataAvailableHandler()
if (!m_Stream->IsEof()) { if (!m_Stream->IsEof()) {
boost::mutex::scoped_lock lock(m_DataHandlerMutex); boost::mutex::scoped_lock lock(m_DataHandlerMutex);
m_Stream->SetCorked(true);
try { try {
while (ProcessMessage()) while (ProcessMessage())
; /* empty loop body */ ; /* empty loop body */
@ -291,6 +289,8 @@ void JsonRpcConnection::DataAvailableHandler()
return; return;
} }
l_JsonRpcConnectionWorkQueues[m_ID % l_JsonRpcConnectionWorkQueueCount].Enqueue(std::bind(&Stream::SetCorked, m_Stream, false));
} else } else
close = true; close = true;