Improved lock handling of signal index cache and implemented better thread release management during shutdown.

Author: ritchiecarroll

src/lib/Timer.h

@@ -50,7 +50,15 @@ namespace sttp
 
             do
             {
-                boost::this_thread::sleep(boost::posix_time::milliseconds(m_interval));
+                try
+                {
+                    boost::this_thread::sleep(boost::posix_time::milliseconds(m_interval));
+                }
+                catch (boost::thread_interrupted&)
+                {
+                    m_running = false;
+                    return;
+                }
 
                 if (m_running)
                     m_callback(this, m_userData);

src/lib/transport/DataPublisher.cpp

@@ -851,7 +851,7 @@ void DataPublisher::Start(const TcpEndPoint& endpoint)
     m_commandChannelService.restart();
 #endif
 
-    m_clientAcceptor = TcpAcceptor(m_commandChannelService, endpoint, false);
+    m_clientAcceptor = TcpAcceptor(m_commandChannelService, endpoint, false); //-V601
 
     // Run call-back thread
     m_commandChannelAcceptThread = Thread([&,this]

src/lib/transport/DataSubscriber.cpp

@@ -413,7 +413,7 @@ void DataSubscriber::ReadPayloadHeader(const ErrorCode& error, size_t bytesTrans
     if (error == error::connection_aborted || error == error::connection_reset || error == error::eof)
     {
         // Connection closed by peer; terminate connection
-        Thread(bind(&DataSubscriber::ConnectionTerminatedDispatcher, this));
+        m_connectionTerminationThread = Thread(bind(&DataSubscriber::ConnectionTerminatedDispatcher, this));
         return;
     }
 
@@ -451,7 +451,7 @@ void DataSubscriber::ReadPacket(const ErrorCode& error, size_t bytesTransferred)
     if (error == error::connection_aborted || error == error::connection_reset || error == error::eof)
     {
         // Connection closed by peer; terminate connection
-        Thread(bind(&DataSubscriber::ConnectionTerminatedDispatcher, this));
+        m_connectionTerminationThread = Thread(bind(&DataSubscriber::ConnectionTerminatedDispatcher, this));
         return;
     }
 
@@ -1017,7 +1017,7 @@ void DataSubscriber::SubscriptionUpdatedDispatcher(DataSubscriber* source, const
     if (signalIndexCachePtr != nullptr)
     {
         const SubscriptionUpdatedCallback subscriptionUpdated = source->m_subscriptionUpdatedCallback;
-        const SignalIndexCachePtr signalIndexCacheRef = source->ReleaseDispatchReference(signalIndexCachePtr);
+        const SignalIndexCachePtr signalIndexCacheRef = source->ReleaseDispatchReference(signalIndexCachePtr); //-V821
 
         if (subscriptionUpdated != nullptr)
             subscriptionUpdated(source, signalIndexCacheRef);
@@ -1258,6 +1258,7 @@ void DataSubscriber::Disconnect(bool autoReconnecting)
     if (!autoReconnecting)
     {
         m_connector.Cancel();
+        m_connectionTerminationThread.join();
         m_connectActionMutex.lock();
     }
 
@@ -1504,7 +1505,7 @@ void DataSubscriber::WriteHandler(const ErrorCode& error, size_t bytesTransferre
     if (error == error::connection_aborted || error == error::connection_reset || error == error::eof)
     {
         // Connection closed by peer; terminate connection
-        Thread(bind(&DataSubscriber::ConnectionTerminatedDispatcher, this));
+        m_connectionTerminationThread = Thread(bind(&DataSubscriber::ConnectionTerminatedDispatcher, this));
         return;
     }
 

src/lib/transport/DataSubscriber.h

@@ -220,6 +220,7 @@ namespace transport
         // Callback thread members
         Thread m_callbackThread;
         ThreadSafeQueue<CallbackDispatcher> m_callbackQueue;
+        Thread m_connectionTerminationThread;
 
         // Command channel
         Thread m_commandChannelResponseThread;

src/lib/transport/SubscriberConnection.cpp

@@ -342,17 +342,24 @@ void SubscriberConnection::SetSubscriptionInfo(const string& value)
     m_subscriptionInfo = source + " version " + version + " updated on " + updatedOn;
 }
 
-const SignalIndexCachePtr& SubscriberConnection::GetSignalIndexCache() const
+const SignalIndexCachePtr& SubscriberConnection::GetSignalIndexCache()
 {
+    ReaderLock readLock(m_signalIndexCacheLock);
+
     return m_signalIndexCache;
 }
 
 void SubscriberConnection::SetSignalIndexCache(SignalIndexCachePtr signalIndexCache)
 {
+    WriterLock writeLock(m_signalIndexCacheLock);
+
     m_signalIndexCache = std::move(signalIndexCache);
 
     // Update measurement routes for newly subscribed measurement signal IDs
     m_parent->m_routingTables.UpdateRoutes(shared_from_this(), m_signalIndexCache->GetSignalIDs());
+
+    // Reset TSSC encoder on successful (re)subscription
+    m_tsscResetRequested = true;
 }
 
 uint64_t SubscriberConnection::GetTotalCommandChannelBytesSent() const
@@ -748,15 +755,9 @@ void SubscriberConnection::HandleSubscribe(uint8_t* data, uint32_t length)
                         // Send updated signal index cache to client with validated rights of the selected input measurement keys                        
                         SendResponse(ServerResponse::UpdateSignalIndexCache, ServerCommand::Subscribe, SerializeSignalIndexCache(*signalIndexCache));
                     }
-                   
-                    m_tsscEncoderLock.lock();
 
-                    // Reset TSSC encoder on successful (re)subscription
-                    m_tsscResetRequested = true;
                     SetSignalIndexCache(signalIndexCache);
 
-                    m_tsscEncoderLock.unlock();
-
                     // If using compact measurement format with base time offsets, setup base time rotation timer
                     if (!m_usingPayloadCompression && m_parent->GetUseBaseTimeOffsets() && m_includeTime)
                     {
@@ -1126,7 +1127,8 @@ SignalIndexCachePtr SubscriberConnection::ParseSubscriptionRequest(const string&
 
 void SubscriberConnection::PublishCompactMeasurements(const std::vector<MeasurementPtr>& measurements)
 {
-    CompactMeasurement serializer(m_signalIndexCache, m_baseTimeOffsets, m_includeTime, m_useMillisecondResolution, m_timeIndex);
+    const SignalIndexCachePtr signalIndexCache = GetSignalIndexCache();
+    CompactMeasurement serializer(signalIndexCache, m_baseTimeOffsets, m_includeTime, m_useMillisecondResolution, m_timeIndex);
     vector<uint8_t> packet, buffer;
     int32_t count = 0;
 
@@ -1137,7 +1139,7 @@ void SubscriberConnection::PublishCompactMeasurements(const std::vector<Measurem
     {
         const Measurement& measurement = *measurements[i];
         const int64_t timestamp = measurement.Timestamp;
-        const int32_t runtimeID = m_signalIndexCache->GetSignalIndex(measurement.SignalID);
+        const int32_t runtimeID = signalIndexCache->GetSignalIndex(measurement.SignalID);
 
         if (runtimeID == Int32::MaxValue)
             continue;
@@ -1193,7 +1195,7 @@ void SubscriberConnection::PublishCompactDataPacket(const vector<uint8_t>& packe
 
 void SubscriberConnection::PublishTSSCMeasurements(const std::vector<MeasurementPtr>& measurements)
 {
-    m_tsscEncoderLock.lock();
+    const SignalIndexCachePtr signalIndexCache = GetSignalIndexCache();
 
     if (m_tsscResetRequested)
     {
@@ -1216,7 +1218,7 @@ void SubscriberConnection::PublishTSSCMeasurements(const std::vector<Measurement
 
     for (const auto& measurement : measurements)
     {
-        const int32_t index = m_signalIndexCache->GetSignalIndex(measurement->SignalID);
+        const int32_t index = signalIndexCache->GetSignalIndex(measurement->SignalID);
 
         if (!m_tsscEncoder.TryAddMeasurement(index, measurement->Timestamp, static_cast<uint32_t>(measurement->Flags), static_cast<float32_t>(measurement->AdjustedValue())))
         {
@@ -1231,8 +1233,6 @@ void SubscriberConnection::PublishTSSCMeasurements(const std::vector<Measurement
 
     if (count > 0)
         PublishTSSCDataPacket(count);
-
-    m_tsscEncoderLock.unlock();
 }
 
 void SubscriberConnection::PublishTSSCDataPacket(int32_t count)

src/lib/transport/SubscriberConnection.h

@@ -101,6 +101,7 @@ namespace transport
 
         // Measurement parsing
         SignalIndexCachePtr m_signalIndexCache;
+        sttp::SharedMutex m_signalIndexCacheLock;
         TimerPtr m_baseTimeRotationTimer;
         int32_t m_timeIndex;
         int64_t m_baseTimeOffsets[2];
@@ -110,7 +111,6 @@ namespace transport
         sttp::Mutex m_latestMeasurementsLock;
         TimerPtr m_throttledPublicationTimer;
         tssc::TSSCEncoder m_tsscEncoder;
-        sttp::Mutex m_tsscEncoderLock;
         uint8_t m_tsscWorkingBuffer[TSSCBufferSize];
         bool m_tsscResetRequested;
         uint16_t m_tsscSequenceNumber;
@@ -247,7 +247,7 @@ namespace transport
         void SetSubscriptionInfo(const std::string& value);
 
         // Gets or sets signal index cache for subscriber representing run-time mappings for subscribed points
-        const SignalIndexCachePtr& GetSignalIndexCache() const;
+        const SignalIndexCachePtr& GetSignalIndexCache();
         void SetSignalIndexCache(SignalIndexCachePtr signalIndexCache);
 
         // Statistical functions

src/lib/transport/SubscriberInstance.cpp

@@ -58,7 +58,10 @@ SubscriberInstance::SubscriberInstance() :
 
 SubscriberInstance::~SubscriberInstance()
 {
-    m_subscriber.reset();
+    if (m_subscriber != nullptr)
+        m_subscriber->Disconnect();
+
+    m_connectThread.join();
 }
 
 // public functions
@@ -150,7 +153,7 @@ void SubscriberInstance::SetMetadataFilters(const std::string& metadataFilters)
 
 void SubscriberInstance::ConnectAsync()
 {
-    Thread(bind(&SubscriberInstance::Connect, this));
+    m_connectThread = Thread(bind(&SubscriberInstance::Connect, this));
 }
 
 void SubscriberInstance::Connect()

src/lib/transport/SubscriberInstance.h

@@ -51,6 +51,7 @@ namespace transport
         std::unordered_map<Guid, MeasurementMetadataPtr> m_measurements;
         sttp::StringMap<DeviceMetadataPtr> m_devices;
         sttp::StringMap<ConfigurationFramePtr> m_configurationFrames;
+        sttp::Thread m_connectThread;
 
         Mutex m_configurationUpdateLock;
         void* m_userData;