Core: Fixed a bug that may have caused drops in the SHM Layer (#1198)

Prior to this commit, SHM data was dropped, if the subscriber wasn't able to access it in 5ms. Now, the subscriber re-tries, until it got access to it, or new data is available.

ecal/core/src/io/ecal_memfile_pool.cpp

@@ -36,7 +36,7 @@ namespace eCAL
     m_created(false),
     m_do_stop(false),
     m_is_observing(false),
-    m_timeout_read(0)
+    m_time_of_last_life_signal(std::chrono::steady_clock::now())
   {
   }
 
@@ -137,7 +137,7 @@ namespace eCAL
   {
     if (!m_is_observing) return false;
 
-    m_timeout_read = 0;
+    m_time_of_last_life_signal = std::chrono::steady_clock::now();
 
     return true;
   }
@@ -147,21 +147,38 @@ namespace eCAL
     // internal clock sample update checking
     uint64_t last_sample_clock(0);
 
+    // Boolean that tells whether the SHM file has new data that we have NOT already accessed
+    bool has_unprocessed_data = false;
+
     // runs as long as there is no timeout and no external stop request
-    while((m_timeout_read < timeout_) && !m_do_stop)
+    while(std::chrono::steady_clock::now() - std::chrono::steady_clock::time_point(m_time_of_last_life_signal) < std::chrono::milliseconds(timeout_)
+           && !m_do_stop)
     {
-      // loop start in ms
-      auto loop_start = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
+      if (!has_unprocessed_data)
+      {
+        // Only wait for the new-data-event, if we haven't processed the data, yet
+        // check for memory file update event from shm writer (20 ms)
+        has_unprocessed_data = gWaitForEvent(m_event_snd, 20);
+
+        if (has_unprocessed_data)
+        {
+          // We got a signal from the publisher! It is alive! So we reset the time since the last live signal
+          m_time_of_last_life_signal = std::chrono::steady_clock::now();
+        }
+      }
 
-      // check for memory file update event from shm writer (20 ms)
-      if(gWaitForEvent(m_event_snd, 20))
+      // If we have unprocessed data, we try to access (and process!) it
+      if(has_unprocessed_data)
       {
         // last chance to stop ..
         if(m_do_stop) break;
 
         // try to open memory file (timeout 5 ms)
         if(m_memfile.GetReadAccess(5))
         {
+          // We have gotten access! Now the data qualifies as processed, so next loop we will wait for the signal for new data, again.
+          has_unprocessed_data = false;
+
           // read the file header
           SMemFileHeader mfile_hdr;
           ReadFileHeader(mfile_hdr);
@@ -237,14 +254,6 @@ namespace eCAL
             }
           }
         }
-
-        // reset timeout
-        m_timeout_read = 0;
-      }
-      else
-      {
-        // increase timeout in ms
-        m_timeout_read += std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count() - loop_start;
       }
     }
 

ecal/core/src/io/ecal_memfile_pool.h

@@ -66,7 +66,7 @@ namespace eCAL
     std::atomic<bool>       m_do_stop;
     std::atomic<bool>       m_is_observing;
 
-    std::atomic<long long>  m_timeout_read;
+    std::atomic<std::chrono::steady_clock::time_point> m_time_of_last_life_signal;
 
     MemFileDataCallbackT    m_data_callback;