Ensure that SetCorked() works properly

diff --git a/lib/base/socketevents-poll.cpp b/lib/base/socketevents-poll.cpp
index 76c13cfc64be1b66c2972916c0ca009eddb415eb..705ebed211972ac71a7723e80cb5760a60285c90 100644 (file)
--- a/lib/base/socketevents-poll.cpp
+++ b/lib/base/socketevents-poll.cpp
@@ -57,11 +57,17 @@ void SocketEventEnginePoll::ThreadProc(int tid)
                                        if (desc.second.Events == 0)
                                                continue;
 
-                                       if (desc.second.EventInterface)
+                                       int events = desc.second.Events;
+
+                                       if (desc.second.EventInterface) {
                                                desc.second.EventInterface->m_EnginePrivate = &pfds[i];
 
+                                               if (!desc.second.EventInterface->m_Events)
+                                                       events = 0;
+                                       }
+
                                        pfds[i].fd = desc.first;
-                                       pfds[i].events = desc.second.Events;
+                                       pfds[i].events = events;
                                        descriptors[i] = desc.second;
 
                                        i++;

diff --git a/lib/base/tlsstream.cpp b/lib/base/tlsstream.cpp
index 403daf336089a49fb70a684658a400a3df3ac680..3d310c0bf79a1aa7189b2e6103301366e393e95e 100644 (file)
--- a/lib/base/tlsstream.cpp
+++ b/lib/base/tlsstream.cpp
@@ -174,6 +174,8 @@ void TlsStream::OnEvent(int revents)
         */
        ERR_clear_error();
 
+       size_t readTotal = 0;
+
        switch (m_CurrentAction) {
                case TlsActionRead:
                        do {
@@ -182,8 +184,10 @@ void TlsStream::OnEvent(int revents)
                                if (rc > 0) {
                                        m_RecvQ->Write(buffer, rc);
                                        success = true;
+
+                                       readTotal += rc;
                                }
-                       } while (rc > 0);
+                       } while (rc > 0 && readTotal < 64 * 1024);
 
                        if (success)
                                m_CV.notify_all();
@@ -265,7 +269,7 @@ void TlsStream::OnEvent(int revents)
 
                lock.unlock();
 
-               while (m_RecvQ->IsDataAvailable() && IsHandlingEvents())
+               while (!IsCorked() && m_RecvQ->IsDataAvailable() && IsHandlingEvents())
                        SignalDataAvailable();
        }
 

diff --git a/lib/remote/httpserverconnection.cpp b/lib/remote/httpserverconnection.cpp
index fdb87097dac6f7c2d7565dfe84117c941915cf50..69d57343d7eb61df2b59024de2e800851ea9445d 100644 (file)
--- a/lib/remote/httpserverconnection.cpp
+++ b/lib/remote/httpserverconnection.cpp
@@ -174,9 +174,7 @@ bool HttpServerConnection::ProcessMessage(void)
                return res;
        }
 
-       m_Stream->SetCorked(true);
-
-       m_RequestQueue.Enqueue(std::bind(&HttpServerConnection::ProcessMessageAsync,
+       m_RequestQueue.Enqueue(boost::bind(&HttpServerConnection::ProcessMessageAsync,
                HttpServerConnection::Ptr(this), m_CurrentRequest, response, m_AuthenticatedUser));
 
        m_Seen = Utility::GetTime();
@@ -348,6 +346,8 @@ void HttpServerConnection::DataAvailableHandler(void)
        if (!m_Stream->IsEof()) {
                boost::mutex::scoped_lock lock(m_DataHandlerMutex);
 
+               m_Stream->SetCorked(true);
+
                try {
                        while (ProcessMessage())
                                ; /* empty loop body */
@@ -357,6 +357,8 @@ void HttpServerConnection::DataAvailableHandler(void)
 
                        close = true;
                }
+
+               m_RequestQueue.Enqueue(boost::bind(&Stream::SetCorked, m_Stream, false));
        } else
                close = true;
 

diff --git a/lib/remote/jsonrpcconnection.cpp b/lib/remote/jsonrpcconnection.cpp
index ac23a10600f8ac878cde7cc75cdc1de0dd5f00fa..add6e1a7077e144bf2ad7edf5f99861a0db76046 100644 (file)
--- a/lib/remote/jsonrpcconnection.cpp
+++ b/lib/remote/jsonrpcconnection.cpp
@@ -144,8 +144,6 @@ void JsonRpcConnection::MessageHandlerWrapper(const String& jsonString)
 
        try {
                MessageHandler(jsonString);
-
-               m_Stream->SetCorked(false);
        } catch (const std::exception& ex) {
                Log(LogWarning, "JsonRpcConnection")
                    << "Error while reading JSON-RPC message for identity '" << m_Identity
@@ -249,8 +247,6 @@ bool JsonRpcConnection::ProcessMessage(void)
        if (srs != StatusNewItem)
                return false;
 
-       m_Stream->SetCorked(true);
-
        l_JsonRpcConnectionWorkQueues[m_ID % l_JsonRpcConnectionWorkQueueCount].Enqueue(boost::bind(&JsonRpcConnection::MessageHandlerWrapper, JsonRpcConnection::Ptr(this), message));
 
        return true;
@@ -266,6 +262,8 @@ void JsonRpcConnection::DataAvailableHandler(void)
        if (!m_Stream->IsEof()) {
                boost::mutex::scoped_lock lock(m_DataHandlerMutex);
 
+               m_Stream->SetCorked(true);
+
                try {
                        while (ProcessMessage())
                                ; /* empty loop body */
@@ -278,6 +276,8 @@ void JsonRpcConnection::DataAvailableHandler(void)
 
                        return;
                }
+
+               l_JsonRpcConnectionWorkQueues[m_ID % l_JsonRpcConnectionWorkQueueCount].Enqueue(boost::bind(&Stream::SetCorked, m_Stream, false));
        } else
                close = true;