Fixed race conditions related to dispatch queue with shared pointers and reconnects on shutdown

Author: ritchiecarroll

src/lib/Timer.h

@@ -42,7 +42,7 @@ namespace sttp
         TimerElapsedCallback m_callback;
         void* m_userData;
         bool m_autoReset;
-        volatile bool m_running;
+        std::atomic_bool m_running;
 
         void TimerThread()
         {
@@ -142,6 +142,9 @@ namespace sttp
 
         void Stop()
         {
+            if (!m_running)
+                return;
+
             m_running = false;
 
             if (m_timerThread != nullptr)

src/lib/Version.h

@@ -25,7 +25,7 @@
 #define __VERSION_H
 
 #define STTP_TITLE "STTP C++ Library"
-#define STTP_VERSION "1.0.8"
+#define STTP_VERSION "1.0.9"
 #define STTP_UPDATEDON "2019-07-02"
 
 #endif

src/lib/transport/DataPublisher.cpp

@@ -111,7 +111,7 @@ void DataPublisher::AcceptConnection(const SubscriberConnectionPtr& connection,
 
         if (connectionAccepted)
         {
-            DispatchClientConnected(connection.get());
+            DispatchClientConnected(AddDispatchReference(connection));
         }
         else
         {
@@ -132,7 +132,7 @@ void DataPublisher::AcceptConnection(const SubscriberConnectionPtr& connection,
 
 void DataPublisher::ConnectionTerminated(const SubscriberConnectionPtr& connection)
 {
-    DispatchClientDisconnected(connection.get());
+    DispatchClientDisconnected(AddDispatchReference(connection));
 }
 
 void DataPublisher::RemoveConnection(const SubscriberConnectionPtr& connection)
@@ -144,6 +144,26 @@ void DataPublisher::RemoveConnection(const SubscriberConnectionPtr& connection)
     m_subscriberConnections.erase(connection);
 }
 
+SubscriberConnection* DataPublisher::AddDispatchReference(SubscriberConnectionPtr connectionRef)
+{
+    SubscriberConnection* connectionPtr = connectionRef.get();
+
+    // Hold onto subscriber connection shared pointer until it's delivered
+    m_subscriberConnectionDispatchRefs.emplace(connectionRef);
+
+    return connectionPtr;
+}
+
+SubscriberConnectionPtr DataPublisher::ReleaseDispatchReference(SubscriberConnection* connectionPtr)
+{
+    const SubscriberConnectionPtr connectionRef = connectionPtr->GetReference();
+    
+    // Remove used reference to subscriber connection pointer
+    m_subscriberConnectionDispatchRefs.erase(connectionRef);
+
+    return connectionRef;
+}
+
 void DataPublisher::Dispatch(const DispatcherFunction& function)
 {
     Dispatch(function, nullptr, 0, 0);
@@ -252,7 +272,7 @@ void DataPublisher::ClientConnectedDispatcher(DataPublisher* source, const vecto
         const SubscriberConnectionCallback clientConnectedCallback = source->m_clientConnectedCallback;
 
         if (clientConnectedCallback != nullptr)
-            clientConnectedCallback(source, connection->GetReference());
+            clientConnectedCallback(source, source->ReleaseDispatchReference(connection));
     }
 }
 
@@ -263,11 +283,12 @@ void DataPublisher::ClientDisconnectedDispatcher(DataPublisher* source, const st
     if (source != nullptr)
     {
         const SubscriberConnectionCallback clientDisconnectedCallback = source->m_clientDisconnectedCallback;
+        const SubscriberConnectionPtr subscriberConnectionRef = source->ReleaseDispatchReference(connection);
 
         if (clientDisconnectedCallback != nullptr)
-            clientDisconnectedCallback(source, connection->GetReference());
+            clientDisconnectedCallback(source, subscriberConnectionRef);
 
-        source->RemoveConnection(connection->GetReference());
+        source->RemoveConnection(subscriberConnectionRef);
     }
 }
 
@@ -280,7 +301,7 @@ void DataPublisher::ProcessingIntervalChangeRequestedDispatcher(DataPublisher* s
         const SubscriberConnectionCallback temporalProcessingIntervalChangeRequestedCallback = source->m_processingIntervalChangeRequestedCallback;
 
         if (temporalProcessingIntervalChangeRequestedCallback != nullptr)
-            temporalProcessingIntervalChangeRequestedCallback(source, connection->GetReference());
+            temporalProcessingIntervalChangeRequestedCallback(source, source->ReleaseDispatchReference(connection));
     }
 }
 
@@ -293,7 +314,7 @@ void DataPublisher::TemporalSubscriptionRequestedDispatcher(DataPublisher* sourc
         const SubscriberConnectionCallback temporalSubscriptionRequestedCallback = source->m_temporalSubscriptionRequestedCallback;
 
         if (temporalSubscriptionRequestedCallback != nullptr)
-            temporalSubscriptionRequestedCallback(source, connection->GetReference());
+            temporalSubscriptionRequestedCallback(source, source->ReleaseDispatchReference(connection));
     }
 }
 
@@ -306,7 +327,7 @@ void DataPublisher::TemporalSubscriptionCanceledDispatcher(DataPublisher* source
         const SubscriberConnectionCallback temporalSubscriptionCanceledCallback = source->m_temporalSubscriptionCanceledCallback;
 
         if (temporalSubscriptionCanceledCallback != nullptr)
-            temporalSubscriptionCanceledCallback(source, connection->GetReference());
+            temporalSubscriptionCanceledCallback(source, source->ReleaseDispatchReference(connection));
     }
 }
 
@@ -319,7 +340,7 @@ void DataPublisher::UserCommandDispatcher(DataPublisher* source, const std::vect
         const UserCommandCallback userCommandCallback = source->m_userCommandCallback;
 
         if (userCommandCallback != nullptr)
-            userCommandCallback(source, userCommandData->connection->GetReference(), userCommandData->command, userCommandData->data);
+            userCommandCallback(source, source->ReleaseDispatchReference(userCommandData->connection), userCommandData->command, userCommandData->data);
     }
 
     delete userCommandData;
@@ -782,6 +803,9 @@ void DataPublisher::Start(const TcpEndPoint& endpoint)
     if (m_started)
         Stop();
 
+    // Let any pending start operation complete before stop - prevents destruction stop before start is completed
+    ScopeLock lock(m_connectActionMutex);
+
 #if BOOST_LEGACY
     m_commandChannelService.reset();
 #else
@@ -827,6 +851,9 @@ void DataPublisher::Start(const string& networkInterfaceIP, uint16_t port)
 
 void DataPublisher::Stop()
 {
+    // Let any pending start operation complete before stop - prevents destruction stop before start is completed
+    ScopeLock lock(m_connectActionMutex);
+
     // Notify running threads that the
     // publisher is shutting down
     m_shuttingDown = true;

src/lib/transport/DataPublisher.h

@@ -68,6 +68,7 @@ namespace transport
         sttp::data::DataSetPtr m_filteringMetadata;
         RoutingTables m_routingTables;
         std::unordered_set<SubscriberConnectionPtr> m_subscriberConnections;
+        std::unordered_set<SubscriberConnectionPtr> m_subscriberConnectionDispatchRefs;
         sttp::SharedMutex m_subscriberConnectionsLock;
         SecurityMode m_securityMode;
         int32_t m_maximumAllowedConnections;
@@ -77,8 +78,9 @@ namespace transport
         bool m_supportsTemporalSubscriptions;
         bool m_useBaseTimeOffsets;
         uint32_t m_cipherKeyRotationPeriod;
-        volatile bool m_started;
-        volatile bool m_shuttingDown;
+        std::atomic_bool m_started;
+        std::atomic_bool m_shuttingDown;
+        sttp::Mutex m_connectActionMutex;
         void* m_userData;
 
         // Callback queue
@@ -106,6 +108,9 @@ namespace transport
         SubscriberConnectionCallback m_temporalSubscriptionCanceledCallback;
         UserCommandCallback m_userCommandCallback;
 
+        SubscriberConnection* AddDispatchReference(SubscriberConnectionPtr connectionRef);
+        SubscriberConnectionPtr ReleaseDispatchReference(SubscriberConnection* connectionPtr);
+
         // Dispatchers
         void Dispatch(const DispatcherFunction& function);
         void Dispatch(const DispatcherFunction& function, const uint8_t* data, uint32_t offset, uint32_t length);

src/lib/transport/DataSubscriber.cpp

@@ -172,7 +172,12 @@ int SubscriberConnector::Connect(DataSubscriber& subscriber, bool autoReconnecti
         try
         {
             m_connectAttempt++;
+
+            if (subscriber.m_disposing)
+                return ConnectCanceled;
+
             subscriber.Connect(m_hostname, m_port, autoReconnecting);
+            
             connected = true;
             break;
         }
@@ -223,7 +228,7 @@ int SubscriberConnector::Connect(DataSubscriber& subscriber, bool autoReconnecti
         }
     }
 
-    return subscriber.IsConnected() ? ConnectSuccess : ConnectFailed;
+    return subscriber.m_disposing ? ConnectCanceled : subscriber.IsConnected() ? ConnectSuccess : ConnectFailed;
 }
 
 // Cancel all current and
@@ -682,7 +687,7 @@ void DataSubscriber::HandleUpdateSignalIndexCache(uint8_t* data, uint32_t offset
     signalIndexCache->Parse(uncompressedBuffer, m_subscriberID);
     m_signalIndexCache.swap(signalIndexCache);
 
-    DispatchSubscriptionUpdated(m_signalIndexCache.get());
+    DispatchSubscriptionUpdated(AddDispatchReference(m_signalIndexCache));
 }
 
 // Updates base time offsets.
@@ -880,6 +885,26 @@ void DataSubscriber::ParseCompactMeasurements(uint8_t* data, uint32_t offset, ui
     }
 }
 
+SignalIndexCache* DataSubscriber::AddDispatchReference(SignalIndexCachePtr signalIndexCacheRef)
+{
+    SignalIndexCache* signalIndexCachePtr = signalIndexCacheRef.get();
+
+    // Hold onto signal index cache shared pointer until it's delivered
+    m_signalIndexCacheDispatchRefs.emplace(signalIndexCacheRef);
+
+    return signalIndexCachePtr;
+}
+
+SignalIndexCachePtr DataSubscriber::ReleaseDispatchReference(SignalIndexCache* signalIndexCachePtr)
+{
+    const SignalIndexCachePtr signalIndexCacheRef = signalIndexCachePtr->GetReference();
+    
+    // Remove used reference to signal index cache pointer
+    m_signalIndexCacheDispatchRefs.erase(signalIndexCacheRef);
+
+    return signalIndexCacheRef;
+}
+
 // Dispatches the given function to the callback thread.
 void DataSubscriber::Dispatch(const DispatcherFunction& function)
 {
@@ -986,7 +1011,7 @@ void DataSubscriber::SubscriptionUpdatedDispatcher(DataSubscriber* source, const
         const SubscriptionUpdatedCallback subscriptionUpdated = source->m_subscriptionUpdatedCallback;
 
         if (subscriptionUpdated != nullptr)
-            subscriptionUpdated(source, signalIndexCache->GetReference());
+            subscriptionUpdated(source, source->ReleaseDispatchReference(signalIndexCache));
     }
 }
 
@@ -1171,6 +1196,8 @@ void DataSubscriber::Connect(const string& hostname, const uint16_t port, bool a
     if (m_connected)
         throw SubscriberException("Subscriber is already connected; disconnect first");
 
+    // Let any pending connect operation complete before disconnect - prevents destruction disconnect before connection is completed
+    ScopeLock lock(m_connectActionMutex);
     DnsResolver resolver(m_commandChannelService);
     const DnsResolver::query query(hostname, to_string(port));
     const DnsResolver::iterator endpointIterator = resolver.resolve(query);
@@ -1218,6 +1245,10 @@ void DataSubscriber::Connect(const string& hostname, const uint16_t port)
 
 void DataSubscriber::Disconnect(bool autoReconnecting)
 {
+    // Let any pending connect operation complete before disconnect - prevents destruction disconnect before connection is completed
+    if (!autoReconnecting)
+        m_connectActionMutex.lock();
+
     ErrorCode error;
 
     // Notify running threads that
@@ -1264,6 +1295,9 @@ void DataSubscriber::Disconnect(bool autoReconnecting)
 
     // Disconnect completed
     m_disconnecting = false;
+
+    if (!autoReconnecting)
+        m_connectActionMutex.unlock();
 }
 
 // Disconnects from the publisher.

src/lib/transport/DataSubscriber.h

@@ -89,7 +89,7 @@ namespace transport
         int32_t m_connectAttempt;
         bool m_connectionRefused;
         bool m_autoReconnect;
-        bool m_cancel;
+        std::atomic_bool m_cancel;
 
         // Auto-reconnect handler.
         static void AutoReconnect(DataSubscriber* subscriber);
@@ -188,16 +188,17 @@ namespace transport
         bool m_compressPayloadData;
         bool m_compressMetadata;
         bool m_compressSignalIndexCache;
-        volatile bool m_disconnecting;
-        volatile bool m_disposing;
+        std::atomic_bool m_disconnecting;
+        std::atomic_bool m_disposing;
+        sttp::Mutex m_connectActionMutex;
         void* m_userData;
 
         // Statistics counters
         uint64_t m_totalCommandChannelBytesReceived;
         uint64_t m_totalDataChannelBytesReceived;
         uint64_t m_totalMeasurementsReceived;
-        volatile bool m_connected;
-        volatile bool m_subscribed;
+        std::atomic_bool m_connected;
+        std::atomic_bool m_subscribed;
 
         // Assembly info
         std::string m_assemblySource;
@@ -206,6 +207,7 @@ namespace transport
 
         // Measurement parsing
         SignalIndexCachePtr m_signalIndexCache;
+        std::unordered_set<SignalIndexCachePtr> m_signalIndexCacheDispatchRefs;
         int32_t m_timeIndex;
         int64_t m_baseTimeOffsets[2];
         tssc::TSSCDecoder m_tsscDecoder;
@@ -264,6 +266,9 @@ namespace transport
         void ParseTSSCMeasurements(uint8_t* data, uint32_t offset, uint32_t length, std::vector<MeasurementPtr>& measurements);
         void ParseCompactMeasurements(uint8_t* data, uint32_t offset, uint32_t length, bool includeTime, bool useMillisecondResolution, int64_t frameLevelTimestamp, std::vector<MeasurementPtr>& measurements);
 
+        SignalIndexCache* AddDispatchReference(SignalIndexCachePtr signalIndexCacheRef);
+        SignalIndexCachePtr ReleaseDispatchReference(SignalIndexCache* signalIndexCachePtr);
+
         // Dispatchers
         void Dispatch(const DispatcherFunction& function);
         void Dispatch(const DispatcherFunction& function, const uint8_t* data, uint32_t offset, uint32_t length);