Updated dispatch reference operations for subscriber connection references in data publisher to use synchronized reference counting map since instances are persisted through multiple dispatches

Author: ritchiecarroll

src/lib/transport/DataPublisher.cpp

@@ -147,20 +147,36 @@ void DataPublisher::RemoveConnection(const SubscriberConnectionPtr& connection)
 SubscriberConnection* DataPublisher::AddDispatchReference(SubscriberConnectionPtr connectionRef)
 {
     SubscriberConnection* connectionPtr = connectionRef.get();
+    ScopeLock lock(m_subscriberConnectionDispatchRefsLock);
 
-    // Hold onto subscriber connection shared pointer until it's delivered
-    m_subscriberConnectionDispatchRefs.emplace(connectionRef);
+    // Increment reference count for subscriber connection shared pointer until all instances are delivered
+    const auto iterator = m_subscriberConnectionDispatchRefs.find(connectionRef);
+
+    if (iterator != m_subscriberConnectionDispatchRefs.end())
+        iterator->second++;
+    else
+        m_subscriberConnectionDispatchRefs.emplace(connectionRef, uint32_t(1));
 
     return connectionPtr;
 }
 
 SubscriberConnectionPtr DataPublisher::ReleaseDispatchReference(SubscriberConnection* connectionPtr)
 {
     const SubscriberConnectionPtr connectionRef = connectionPtr->GetReference();
+    ScopeLock lock(m_subscriberConnectionDispatchRefsLock);
 
-    // Remove used reference to subscriber connection pointer
-    m_subscriberConnectionDispatchRefs.erase(connectionRef);
+    // Decrement reference count to subscriber connection pointer
+    const auto iterator = m_subscriberConnectionDispatchRefs.find(connectionRef);
+
+    if (iterator != m_subscriberConnectionDispatchRefs.end())
+    {
+        iterator->second--;
 
+        // Remove references when count hits zero
+        if (iterator->second < 1)
+            m_subscriberConnectionDispatchRefs.erase(connectionRef);
+    }
+        
     return connectionRef;
 }
 

src/lib/transport/DataPublisher.h

@@ -68,7 +68,6 @@ 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;
@@ -83,6 +82,11 @@ namespace transport
         sttp::Mutex m_connectActionMutex;
         void* m_userData;
 
+        // Dispatch reference - unordered map needed to manage reference
+        // counts on subscriber connections since these are persisted
+        std::unordered_map<SubscriberConnectionPtr, uint32_t> m_subscriberConnectionDispatchRefs;
+        sttp::Mutex m_subscriberConnectionDispatchRefsLock;
+
         // Callback queue
         Thread m_callbackThread;
         ThreadSafeQueue<CallbackDispatcher> m_callbackQueue;

src/lib/transport/DataSubscriber.cpp

@@ -587,6 +587,7 @@ void DataSubscriber::HandleSucceeded(uint8_t commandCode, uint8_t* data, uint32_
             // still an associated message to be processed.
             m_subscribed = (commandCode == ServerCommand::Subscribe); //-V796
 
+        case ServerCommand::UpdateProcessingInterval:
         case ServerCommand::RotateCipherKeys:
             // Each of these responses come with a message that will
             // be delivered to the user via the status message callback.
@@ -607,7 +608,7 @@ void DataSubscriber::HandleSucceeded(uint8_t commandCode, uint8_t* data, uint32_
             // If we don't know what the message is, we can't interpret
             // the data sent with the packet. Deliver an error message
             // to the user via the error message callback.
-            messageStream << "Received success code in response to unknown server command 0x" << ToHex(commandCode);
+            messageStream << "Received success code in response to unknown server command " << ToHex(commandCode);
             DispatchErrorMessage(messageStream.str());
             break;
     }
@@ -889,6 +890,9 @@ SignalIndexCache* DataSubscriber::AddDispatchReference(SignalIndexCachePtr signa
 {
     SignalIndexCache* signalIndexCachePtr = signalIndexCacheRef.get();
 
+    // Note that no lock is needed for m_signalIndexCacheDispatchRefs set since all calls from
+    // HandleUpdateSignalIndexCache are from a single connection and are processed sequentially
+
     // Hold onto signal index cache shared pointer until it's delivered
     m_signalIndexCacheDispatchRefs.emplace(signalIndexCacheRef);
 

src/lib/transport/DataSubscriber.h

@@ -207,13 +207,16 @@ 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;
         bool m_tsscResetRequested;
         uint16_t m_tsscSequenceNumber;
 
+        // Dispatch reference - unordered set works fine for signal
+        // index cache since each new call will be for a new instance
+        std::unordered_set<SignalIndexCachePtr> m_signalIndexCacheDispatchRefs;
+
         // Callback thread members
         Thread m_callbackThread;
         ThreadSafeQueue<CallbackDispatcher> m_callbackQueue;

src/lib/transport/SubscriberConnection.cpp

@@ -1715,7 +1715,7 @@ string SubscriberConnection::DecodeString(const uint8_t* data, uint32_t offset,
             return string(reinterpret_cast<char*>(const_cast<uint8_t*>(&data[offset])), length / sizeof(char));
         case OperationalEncoding::UTF16BE:
             // UTF16 in C++ is encoded as big-endian
-            swapBytes = !swapBytes;
+            swapBytes = !swapBytes; //-V796
         case OperationalEncoding::UTF16LE:
         {
             wstring value(length / enc_sizeof_wchar, L'\0');
@@ -1755,7 +1755,7 @@ vector<uint8_t> SubscriberConnection::EncodeString(const string& value) const
             break;
         case OperationalEncoding::UTF16BE:
             // UTF16 in C++ is encoded as big-endian
-            swapBytes = !swapBytes;
+            swapBytes = !swapBytes; //-V796
         case OperationalEncoding::UTF16LE:
         {
             const wstring utf16 = ToUTF16(value);