(mac ssl) rename DarwinSSL -> SecureTransport (see this too -> https://github.com/curl/curl/issues/3733)

.github/workflows/ccpp.yml

@@ -12,20 +12,31 @@ jobs:
     - name: make test
       run: make test
 
-  mac:
-    runs-on: macOS-latest
-    steps:
-    - uses: actions/checkout@v1
-    - name: make test
-      run: make test
-
-  tsan:
+  mac_tsan_sectransport:
     runs-on: macOS-latest
     steps:
     - uses: actions/checkout@v1
     - name: make test_tsan
       run: make test_tsan
 
+  mac_tsan_openssl:
+    runs-on: macOS-latest
+    steps:
+    - uses: actions/checkout@v1
+    - name: install openssl
+      run: brew install openssl
+    - name: make test
+      run: make test_tsan_openssl
+
+  mac_tsan_mbedtls:
+    runs-on: macOS-latest
+    steps:
+    - uses: actions/checkout@v1
+    - name: install mbedtls
+      run: brew install mbedtls
+    - name: make test
+      run: make test_tsan_mbedtls
+
   win:
     runs-on: windows-latest
     steps:

CMakeLists.txt

@@ -200,7 +200,9 @@ if (USE_TLS AND USE_MBED_TLS)
   endif()
 endif()
 
-find_package(ZLIB)
+if (NOT ZLIB_FOUND)
+  find_package(ZLIB)
+endif()
 if (ZLIB_FOUND)
   include_directories(${ZLIB_INCLUDE_DIRS})
   target_link_libraries(ixwebsocket ${ZLIB_LIBRARIES})

docs/CHANGELOG.md

@@ -1,6 +1,18 @@
 # Changelog
 All changes to this project will be documented in this file.
 
+## [9.1.2] - 2020-03-26
+
+(mac ssl) rename DarwinSSL -> SecureTransport (see this too -> https://github.com/curl/curl/issues/3733)
+
+## [9.1.1] - 2020-03-26
+
+(websocket) fix data race accessing _socket object without mutex protection when calling wakeUpFromPoll in WebSocketTransport.cpp
+
+## [9.1.0] - 2020-03-26
+
+(ixcobra) add explicit event types for handshake, authentication and subscription failure, and handle those by exiting in ws_cobra_subcribe and friends
+
 ## [9.0.3] - 2020-03-24
 
 (ws connect) display statistics about how much time it takes to stop the connection

ixbots/ixbots/IXCobraToSentryBot.cpp

@@ -37,6 +37,7 @@ namespace ix
         std::atomic<bool> errorSending(false);
         std::atomic<bool> stop(false);
         std::atomic<bool> throttled(false);
+        std::atomic<bool> fatalCobraError(false);
 
         QueueManager queueManager(maxQueueSize);
 
@@ -179,6 +180,7 @@ namespace ix
                                verbose,
                                &throttled,
                                &receivedCount,
+                               &fatalCobraError,
                                &queueManager](ix::CobraConnectionEventType eventType,
                                               const std::string& errMsg,
                                               const ix::WebSocketHttpHeaders& headers,
@@ -240,6 +242,21 @@ namespace ix
             {
                 spdlog::info("Received websocket pong");
             }
+            else if (eventType == ix::CobraConnection_EventType_Handshake_Error)
+            {
+                spdlog::error("Subscriber: Handshake error: {}", errMsg);
+                fatalCobraError = true;
+            }
+            else if (eventType == ix::CobraConnection_EventType_Authentication_Error)
+            {
+                spdlog::error("Subscriber: Authentication error: {}", errMsg);
+                fatalCobraError = true;
+            }
+            else if (eventType == ix::CobraConnection_EventType_Subscription_Error)
+            {
+                spdlog::error("Subscriber: Subscription error: {}", errMsg);
+                fatalCobraError = true;
+            }
         });
 
         // Run forever
@@ -251,6 +268,7 @@ namespace ix
                 std::this_thread::sleep_for(duration);
 
                 if (strict && errorSending) break;
+                if (fatalCobraError) break;
             }
         }
         // Run for a duration, used by unittesting now
@@ -262,6 +280,7 @@ namespace ix
                 std::this_thread::sleep_for(duration);
 
                 if (strict && errorSending) break;
+                if (fatalCobraError) break;
             }
         }
 
@@ -272,12 +291,15 @@ namespace ix
         conn.disconnect();
         stop = true;
 
+        // progress thread
         t1.join();
-        if (t2.joinable()) t2.join();
-        spdlog::info("heartbeat thread done");
 
+        // heartbeat thread
+        if (t2.joinable()) t2.join();
+
+        // sentry sender thread
         t3.join();
 
-        return (strict && errorSending) ? -1 : (int) sentCount;
+        return ((strict && errorSending) || fatalCobraError) ? -1 : (int) sentCount;
     }
 } // namespace ix

ixbots/ixbots/IXCobraToStatsdBot.cpp

@@ -77,6 +77,7 @@ namespace ix
         std::atomic<uint64_t> sentCount(0);
         std::atomic<uint64_t> receivedCount(0);
         std::atomic<bool> stop(false);
+        std::atomic<bool> fatalCobraError(false);
 
         QueueManager queueManager(maxQueueSize);
 
@@ -88,16 +89,18 @@ namespace ix
                 auto duration = std::chrono::seconds(1);
                 std::this_thread::sleep_for(duration);
             }
+
+            spdlog::info("timer thread done");
         };
 
         std::thread t1(timer);
 
-        auto heartbeat = [&sentCount, &receivedCount, &enableHeartbeat] {
+        auto heartbeat = [&sentCount, &receivedCount, &stop, &enableHeartbeat] {
             std::string state("na");
 
             if (!enableHeartbeat) return;
 
-            while (true)
+            while (!stop)
             {
                 std::stringstream ss;
                 ss << "messages received " << receivedCount;
@@ -115,6 +118,8 @@ namespace ix
                 auto duration = std::chrono::minutes(1);
                 std::this_thread::sleep_for(duration);
             }
+
+            spdlog::info("heartbeat thread done");
         };
 
         std::thread t2(heartbeat);
@@ -142,7 +147,7 @@ namespace ix
         std::thread t3(statsdSender);
 
         conn.setEventCallback(
-            [&conn, &channel, &filter, &position, &jsonWriter, verbose, &queueManager, &receivedCount](
+            [&conn, &channel, &filter, &position, &jsonWriter, verbose, &queueManager, &receivedCount, &fatalCobraError](
                 ix::CobraConnectionEventType eventType,
                 const std::string& errMsg,
                 const ix::WebSocketHttpHeaders& headers,
@@ -200,6 +205,21 @@ namespace ix
                 {
                     spdlog::info("Received websocket pong");
                 }
+                else if (eventType == ix::CobraConnection_EventType_Handshake_Error)
+                {
+                    spdlog::error("Subscriber: Handshake error: {}", errMsg);
+                    fatalCobraError = true;
+                }
+                else if (eventType == ix::CobraConnection_EventType_Authentication_Error)
+                {
+                    spdlog::error("Subscriber: Authentication error: {}", errMsg);
+                    fatalCobraError = true;
+                }
+                else if (eventType == ix::CobraConnection_EventType_Subscription_Error)
+                {
+                    spdlog::error("Subscriber: Subscription error: {}", errMsg);
+                    fatalCobraError = true;
+                }
             });
 
         // Run forever
@@ -209,6 +229,8 @@ namespace ix
             {
                 auto duration = std::chrono::seconds(1);
                 std::this_thread::sleep_for(duration);
+
+                if (fatalCobraError) break;
             }
         }
         // Run for a duration, used by unittesting now
@@ -218,6 +240,8 @@ namespace ix
             {
                 auto duration = std::chrono::seconds(1);
                 std::this_thread::sleep_for(duration);
+
+                if (fatalCobraError) break;
             }
         }
 
@@ -228,12 +252,15 @@ namespace ix
         conn.disconnect();
         stop = true;
 
+        // progress thread
         t1.join();
-        if (t2.joinable()) t2.join();
-        spdlog::info("heartbeat thread done");
 
+        // heartbeat thread
+        if (t2.joinable()) t2.join();
+
+        // statsd sender thread
         t3.join();
 
-        return (int) sentCount;
+        return fatalCobraError ? -1 : (int) sentCount;
     }
 } // namespace ix

ixcobra/ixcobra/IXCobraConnection.cpp

@@ -166,7 +166,8 @@ namespace ix
                     }
                     else if (action == "auth/handshake/error")
                     {
-                        invokeErrorCallback("Handshake error", msg->str);
+                        invokeEventCallback(ix::CobraConnection_EventType_Handshake_Error,
+                                            msg->str);
                     }
                     else if (action == "auth/authenticate/ok")
                     {
@@ -176,7 +177,8 @@ namespace ix
                     }
                     else if (action == "auth/authenticate/error")
                     {
-                        invokeErrorCallback("Authentication error", msg->str);
+                        invokeEventCallback(ix::CobraConnection_EventType_Authentication_Error,
+                                            msg->str);
                     }
                     else if (action == "rtm/subscription/data")
                     {
@@ -191,7 +193,8 @@ namespace ix
                     }
                     else if (action == "rtm/subscribe/error")
                     {
-                        invokeErrorCallback("Subscription error", msg->str);
+                        invokeEventCallback(ix::CobraConnection_EventType_Subscription_Error,
+                                            msg->str);
                     }
                     else if (action == "rtm/unsubscribe/ok")
                     {

ixcobra/ixcobra/IXCobraConnection.h

@@ -37,7 +37,10 @@ namespace ix
         CobraConnection_EventType_Subscribed = 4,
         CobraConnection_EventType_UnSubscribed = 5,
         CobraConnection_EventType_Published = 6,
-        CobraConnection_EventType_Pong = 7
+        CobraConnection_EventType_Pong = 7,
+        CobraConnection_EventType_Handshake_Error = 8,
+        CobraConnection_EventType_Authentication_Error = 9,
+        CobraConnection_EventType_Subscription_Error = 10
     };
 
     enum CobraConnectionPublishMode

ixwebsocket/IXUserAgent.cpp

@@ -71,7 +71,7 @@ namespace ix
 #elif defined(IXWEBSOCKET_USE_OPEN_SSL)
         ss << " ssl/OpenSSL " << OPENSSL_VERSION_TEXT;
 #elif __APPLE__
-        ss << " ssl/DarwinSSL";
+        ss << " ssl/SecureTransport";
 #endif
 #else
         ss << " nossl";

ixwebsocket/IXWebSocketTransport.cpp

@@ -625,7 +625,7 @@ namespace ix
                     // send back the CLOSE frame
                     sendCloseFrame(code, reason);
 
-                    _socket->wakeUpFromPoll(Socket::kCloseRequest);
+                    wakeUpFromPoll(Socket::kCloseRequest);
 
                     bool remote = true;
                     closeSocketAndSwitchToClosedState(code, reason, _rxbuf.size(), remote);
@@ -845,7 +845,7 @@ namespace ix
         // Request to flush the send buffer on the background thread if it isn't empty
         if (!isSendBufferEmpty())
         {
-            _socket->wakeUpFromPoll(Socket::kSendRequest);
+            wakeUpFromPoll(Socket::kSendRequest);
 
             // FIXME: we should have a timeout when sending large messages: see #131
             if (_blockingSend && !flushSendBuffer())
@@ -1063,6 +1063,12 @@ namespace ix
         _socket->close();
     }
 
+    bool WebSocketTransport::wakeUpFromPoll(uint64_t wakeUpCode)
+    {
+        std::lock_guard<std::mutex> lock(_socketMutex);
+        return _socket->wakeUpFromPoll(wakeUpCode);
+    }
+
     void WebSocketTransport::closeSocketAndSwitchToClosedState(uint16_t code,
                                                                const std::string& reason,
                                                                size_t closeWireSize,
@@ -1110,8 +1116,9 @@ namespace ix
         setReadyState(ReadyState::CLOSING);
 
         sendCloseFrame(code, reason);
+
         // wake up the poll, but do not close yet
-        _socket->wakeUpFromPoll(Socket::kSendRequest);
+        wakeUpFromPoll(Socket::kSendRequest);
     }
 
     size_t WebSocketTransport::bufferedAmount() const

ixwebsocket/IXWebSocketTransport.h

@@ -229,6 +229,8 @@ namespace ix
                                                size_t closeWireSize,
                                                bool remote);
 
+        bool wakeUpFromPoll(uint64_t wakeUpCode);
+
         bool flushSendBuffer();
         bool sendOnSocket();
         bool receiveFromSocket();

ixwebsocket/IXWebSocketVersion.h

@@ -6,4 +6,4 @@
 
 #pragma once
 
-#define IX_WEBSOCKET_VERSION "9.0.3"
+#define IX_WEBSOCKET_VERSION "9.1.2"

makefile

@@ -105,6 +105,16 @@ test_asan:
 	(cd build/test ; ln -sf Debug/ixwebsocket_unittest)
 	(cd test ; python2.7 run.py -r)
 
+test_tsan_openssl:
+	mkdir -p build && (cd build && cmake -GXcode -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_TEST=1 -DUSE_OPEN_SSL=1 .. && xcodebuild -project ixwebsocket.xcodeproj -target ixwebsocket_unittest -enableThreadSanitizer YES)
+	(cd build/test ; ln -sf Debug/ixwebsocket_unittest)
+	(cd test ; python2.7 run.py -r)
+
+test_tsan_mbedtls:
+	mkdir -p build && (cd build && cmake -GXcode -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_TEST=1 -DUSE_MBED_TLS=1 .. && xcodebuild -project ixwebsocket.xcodeproj -target ixwebsocket_unittest -enableThreadSanitizer YES)
+	(cd build/test ; ln -sf Debug/ixwebsocket_unittest)
+	(cd test ; python2.7 run.py -r)
+
 test_openssl:
 	mkdir -p build && (cd build ; cmake -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_OPEN_SSL=1 -DUSE_TEST=1 .. ; make -j 4)
 	(cd test ; python2.7 run.py -r)

ws/ws_cobra_subscribe.cpp

@@ -55,8 +55,10 @@ namespace ix
         std::atomic<int> msgPerSeconds(0);
         std::atomic<int> msgCount(0);
 
-        auto timer = [&msgPerSeconds, &msgCount] {
-            while (true)
+        std::atomic<bool> fatalCobraError(false);
+
+        auto timer = [&msgPerSeconds, &msgCount, &fatalCobraError] {
+            while (!fatalCobraError)
             {
                 spdlog::info("#messages {} msg/s {}", msgCount, msgPerSeconds);
 
@@ -78,11 +80,12 @@ namespace ix
                                &msgCount,
                                &msgPerSeconds,
                                &quiet,
-                               &fluentd](ix::CobraConnectionEventType eventType,
-                                         const std::string& errMsg,
-                                         const ix::WebSocketHttpHeaders& headers,
-                                         const std::string& subscriptionId,
-                                         CobraConnection::MsgId msgId) {
+                               &fluentd,
+                               &fatalCobraError](ix::CobraConnectionEventType eventType,
+                                                 const std::string& errMsg,
+                                                 const ix::WebSocketHttpHeaders& headers,
+                                                 const std::string& subscriptionId,
+                                                 CobraConnection::MsgId msgId) {
             if (eventType == ix::CobraConnection_EventType_Open)
             {
                 spdlog::info("Subscriber connected");
@@ -137,14 +140,32 @@ namespace ix
             {
                 spdlog::info("Received websocket pong");
             }
+            else if (eventType == ix::CobraConnection_EventType_Handshake_Error)
+            {
+                spdlog::error("Subscriber: Handshake error: {}", errMsg);
+                fatalCobraError = true;
+            }
+            else if (eventType == ix::CobraConnection_EventType_Authentication_Error)
+            {
+                spdlog::error("Subscriber: Authentication error: {}", errMsg);
+                fatalCobraError = true;
+            }
+            else if (eventType == ix::CobraConnection_EventType_Subscription_Error)
+            {
+                spdlog::error("Subscriber: Subscription error: {}", errMsg);
+                fatalCobraError = true;
+            }
         });
 
-        while (true)
+        while (!fatalCobraError)
         {
             auto duration = std::chrono::seconds(1);
             std::this_thread::sleep_for(duration);
         }
 
-        return 0;
+        conn.disconnect();
+        t.join();
+
+        return fatalCobraError ? 1 : 0;
     }
 } // namespace ix