Merge pull request microsoft#393 from ChristophAlbert/cpprestsdk

Race condition in ws_client_wspp.cpp

Author: ras0219-msft

Release/src/CMakeLists.txt

@@ -45,12 +45,14 @@ elseif(CPPREST_WEBSOCKETS_IMPL STREQUAL "winrt")
     websockets/client/ws_msg.cpp
     websockets/client/ws_client.cpp
     websockets/client/ws_client_winrt.cpp
+    websockets/client/ws_client_impl.h
   )
 elseif(CPPREST_WEBSOCKETS_IMPL STREQUAL "wspp")
   list(APPEND SOURCES
     websockets/client/ws_msg.cpp
     websockets/client/ws_client.cpp
     websockets/client/ws_client_wspp.cpp
+    websockets/client/ws_client_impl.h
   )
 endif()
 

Release/src/build/common.vcxitems

@@ -86,6 +86,7 @@
     <ClInclude Include="$(MSBuildThisFileDirectory)..\http\client\http_client_impl.h" />
     <ClInclude Include="$(MSBuildThisFileDirectory)..\http\listener\http_server_impl.h" />
     <ClInclude Include="$(MSBuildThisFileDirectory)..\http\common\internal_http_helpers.h" />
+    <ClInclude Include="$(MSBuildThisFileDirectory)..\websockets\client\ws_client_impl.h" />
     <ClInclude Include="$(MSBuildThisFileDirectory)..\pch\stdafx.h" />
   </ItemGroup>
   <ItemGroup>

Release/src/build/common.vcxitems.filters

@@ -218,7 +218,12 @@
     <ClInclude Include="$(MSBuildThisFileDirectory)..\http\listener\http_server_impl.h">
       <Filter>Header Files\private</Filter>
     </ClInclude>
-    <ClInclude Include="$(MSBuildThisFileDirectory)..\http\common\internal_http_helpers.h" />
+    <ClInclude Include="$(MSBuildThisFileDirectory)..\websockets\client\ws_client_impl.h">
+      <Filter>Header Files\private</Filter>
+    </ClInclude>
+    <ClInclude Include="$(MSBuildThisFileDirectory)..\http\common\internal_http_helpers.h">
+      <Filter>Header Files\private</Filter>
+    </ClInclude>
   </ItemGroup>
   <ItemGroup>
     <None Include="$(MSBuildThisFileDirectory)..\..\include\cpprest\details\http_constants.dat">

Release/src/websockets/client/ws_client_impl.h

@@ -0,0 +1,58 @@
+#pragma once
+
+#include <queue>
+#include <mutex>
+#include "cpprest/ws_client.h"
+#include "cpprest/ws_msg.h"
+
+namespace web
+{
+namespace websockets
+{
+namespace client
+{
+namespace details
+{
+
+struct outgoing_msg_queue
+{
+    enum class state
+    {
+        was_empty,
+        was_not_empty,
+    };
+
+    state push(websocket_outgoing_message& msg)
+    {
+        state ret = state::was_not_empty;
+        std::lock_guard<std::mutex> lock(m_lock);
+        if (m_queue.empty())
+        {
+            ret = state::was_empty;
+        }
+
+        m_queue.push(msg);
+        return ret;
+    }
+
+    bool pop_and_peek(websocket_outgoing_message& msg)
+    {
+        std::lock_guard<std::mutex> lock(m_lock);
+
+        m_queue.pop();
+
+        if (m_queue.empty())
+        {
+            return false;
+        }
+        msg = m_queue.front();
+        return true;
+    }
+
+private:
+    std::mutex m_lock;
+    std::queue<websocket_outgoing_message> m_queue;
+};
+
+
+}}}}

Release/src/websockets/client/ws_client_winrt.cpp

@@ -15,6 +15,8 @@
 
 #if !defined(CPPREST_EXCLUDE_WEBSOCKETS)
 
+#include "ws_client_impl.h"
+
 using namespace ::Windows::Foundation;
 using namespace ::Windows::Storage;
 using namespace ::Windows::Storage::Streams;
@@ -230,19 +232,10 @@ class winrt_callback_client : public websocket_client_callback_impl, public std:
             return pplx::task_from_exception<void>(websocket_exception("Message size too large. Ensure message length is less than UINT_MAX."));
         }
 
-        bool msg_pending = false;
-        {
-            std::lock_guard<std::mutex> lock(m_send_lock);
-            if (m_outgoing_msg_queue.size() > 0)
-            {
-                msg_pending = true;
-            }
-
-            m_outgoing_msg_queue.push(msg);
-        }
+        auto msg_pending = m_out_queue.push(msg);
 
         // No sends in progress
-        if (msg_pending == false)
+        if (msg_pending == outgoing_msg_queue::state::was_empty)
         {
             // Start sending the message
             send_msg(msg);
@@ -379,22 +372,8 @@ class winrt_callback_client : public websocket_client_callback_impl, public std:
                 msg.signal_body_sent();
             }
 
-            bool msg_pending = false;
             websocket_outgoing_message next_msg;
-            {
-                // Only hold the lock when actually touching the queue.
-                std::lock_guard<std::mutex> lock(this_client->m_send_lock);
-
-                // First message in queue has been sent
-                this_client->m_outgoing_msg_queue.pop();
-
-                if (this_client->m_outgoing_msg_queue.size() > 0)
-                {
-                    next_msg = this_client->m_outgoing_msg_queue.front();
-                    msg_pending = true;
-                }
-            }
-
+            bool msg_pending = this_client->m_out_queue.pop_and_peek(next_msg);
             if (msg_pending)
             {
                 this_client->send_msg(next_msg);
@@ -443,11 +422,8 @@ class winrt_callback_client : public websocket_client_callback_impl, public std:
     std::function<void(websocket_incoming_message)> m_external_message_handler;
     std::function<void(websocket_close_status, const utility::string_t&, const std::error_code&)> m_external_close_handler;
 
-    // The implementation has to ensure ordering of send requests
-    std::mutex m_send_lock;
-
     // Queue to track pending sends
-    std::queue<websocket_outgoing_message> m_outgoing_msg_queue;
+    outgoing_msg_queue m_out_queue;
 };
 
 void ReceiveContext::OnReceive(MessageWebSocket^ sender, MessageWebSocketMessageReceivedEventArgs^ args)

Release/src/websockets/client/ws_client_wspp.cpp

@@ -18,6 +18,8 @@
 #include "cpprest/details/x509_cert_utilities.h"
 #include "pplx/threadpool.h"
 
+#include "ws_client_impl.h"
+
 // Force websocketpp to use C++ std::error_code instead of Boost.
 #define _WEBSOCKETPP_CPP11_SYSTEM_ERROR_
 #if defined(_MSC_VER)
@@ -117,8 +119,7 @@ class wspp_callback_client : public websocket_client_callback_impl, public std::
 public:
     wspp_callback_client(websocket_client_config config) :
         websocket_client_callback_impl(std::move(config)),
-        m_state(CREATED),
-        m_num_sends(0)
+        m_state(CREATED)
 #if defined(__APPLE__) || (defined(ANDROID) || defined(__ANDROID__)) || defined(_WIN32)
         , m_openssl_failed(false)
 #endif
@@ -402,10 +403,10 @@ class wspp_callback_client : public websocket_client_callback_impl, public std::
         {
         case websocket_message_type::text_message:
         case websocket_message_type::binary_message:
-		case websocket_message_type::pong:
+        case websocket_message_type::pong:
             break;
         default:
-            return pplx::task_from_exception<void>(websocket_exception("Invalid message type"));
+            return pplx::task_from_exception<void>(websocket_exception("Message Type not supported."));
         }
 
         const auto length = msg.m_length;
@@ -418,18 +419,13 @@ class wspp_callback_client : public websocket_client_callback_impl, public std::
             return pplx::task_from_exception<void>(websocket_exception("Message size too large. Ensure message length is less than UINT_MAX."));
         }
 
+        auto msg_pending = m_out_queue.push(msg);
+
+        // No sends in progress
+        if (msg_pending == outgoing_msg_queue::state::was_empty)
         {
-            if (++m_num_sends == 1) // No sends in progress
-            {
-                // Start sending the message
-                send_msg(msg);
-            }
-            else
-            {
-                // Only actually have to take the lock if touching the queue.
-                std::lock_guard<std::mutex> lock(m_send_lock);
-                m_outgoing_msg_queue.push(msg);
-            }
+            // Start sending the message
+            send_msg(msg);
         }
 
         return pplx::create_task(msg.body_sent());
@@ -565,16 +561,12 @@ class wspp_callback_client : public websocket_client_callback_impl, public std::
                 msg.signal_body_sent();
             }
 
-            if (--this_client->m_num_sends > 0)
+            websocket_outgoing_message next_msg;
+            bool msg_pending = this_client->m_out_queue.pop_and_peek(next_msg);
+
+            if (msg_pending)
             {
-                // Only hold the lock when actually touching the queue.
-                websocket_outgoing_message next_msg;
-                {
-                    std::lock_guard<std::mutex> lock(this_client->m_send_lock);
-                    next_msg = this_client->m_outgoing_msg_queue.front();
-                    this_client->m_outgoing_msg_queue.pop();
-                }
-                this_client->send_msg(next_msg);
+              this_client->send_msg(next_msg);
             }
         });
     }
@@ -669,19 +661,19 @@ class wspp_callback_client : public websocket_client_callback_impl, public std::
                 ec);
             break;
         case websocket_message_type::binary_message:
-			client.send(
+            client.send(
                 this_client->m_con,
                 sp_allocated.get(),
                 length,
                 websocketpp::frame::opcode::binary,
                 ec);
             break;
-		case websocket_message_type::pong:
-			client.pong(
-				this_client->m_con,
-				"",
-				ec);
-			break;
+        case websocket_message_type::pong:
+            client.pong(
+                this_client->m_con,
+                "",
+                ec);
+            break;
         default:
             // This case should have already been filtered above.
             std::abort();
@@ -763,14 +755,8 @@ class wspp_callback_client : public websocket_client_callback_impl, public std::
     State m_state;
     std::unique_ptr<websocketpp_client_base> m_client;
 
-    // Guards access to m_outgoing_msg_queue
-    std::mutex m_send_lock;
-
-    // Queue to order the sends
-    std::queue<websocket_outgoing_message> m_outgoing_msg_queue;
-
-    // Number of sends in progress and queued up.
-    std::atomic<int> m_num_sends;
+    // Queue to track pending sends
+    outgoing_msg_queue m_out_queue;
 
     // External callback for handling received and close event
     std::function<void(websocket_incoming_message)> m_external_message_handler;