Fixed race condition in file cache and async observer under load

Author: wsobel

simulator/run_scenario.rb

@@ -30,7 +30,7 @@
   end
 
   opts.on('--heartbeat [interval]', OptionParser::DecimalInteger, "Heartbeat Frequency") do |v|
-    $heartbeat = v.to_f / 1000.0
+    $heartbeat = v.to_i
   end
 
   opts.on('-t', '--[no-]scenario', 'Run scenario or log') do |v|
@@ -79,7 +79,7 @@ def heartbeat(socket)
 
           puts "Received #{r.strip}, responding with pong" if $verbose
           $mutex.synchronize {
-            socket.puts "* PONG #{($heartbeat * 1000).to_i}"
+            socket.puts "* PONG #{$heartbeat}"
             socket.flush
           }
         else

src/mtconnect/observation/change_observer.cpp

@@ -47,7 +47,7 @@ namespace mtconnect::observation {
 
   void ChangeObserver::handler(boost::system::error_code ec)
   {
-    boost::asio::dispatch(m_strand, boost::bind(m_handler, ec));
+    boost::asio::post(m_strand, boost::bind(m_handler, ec));
   }
 
   // Signaler Management

src/mtconnect/sink/rest_sink/file_cache.cpp

@@ -311,43 +311,47 @@ namespace mtconnect::sink::rest_sink {
     {
       CachedFilePtr file;
 
-      auto cached = m_fileCache.find(name);
-      if (cached != m_fileCache.end())
       {
-        auto fp = cached->second;
-        if (!fp->m_cached || fp->m_redirect)
+        std::lock_guard<std::recursive_mutex> lock(m_cacheLock);
+        
+        auto cached = m_fileCache.find(name);
+        if (cached != m_fileCache.end())
         {
-          file = fp;
-        }
-        else
-        {
-          // Cleanup files if they have changed since last cached
-          // Also remove any gzipped content as well
-          auto lastWrite = std::filesystem::last_write_time(fp->m_path);
-          if (lastWrite == fp->m_lastWrite)
+          auto fp = cached->second;
+          if (!fp->m_cached || fp->m_redirect)
+          {
             file = fp;
-          else if (fp->m_pathGz && fs::exists(*fp->m_pathGz))
-            fs::remove(*fp->m_pathGz);
+          }
+          else
+          {
+            // Cleanup files if they have changed since last cached
+            // Also remove any gzipped content as well
+            auto lastWrite = std::filesystem::last_write_time(fp->m_path);
+            if (lastWrite == fp->m_lastWrite)
+              file = fp;
+            else if (fp->m_pathGz && fs::exists(*fp->m_pathGz))
+              fs::remove(*fp->m_pathGz);
+          }
         }
-      }
-
-      if (!file)
-      {
-        auto path = m_fileMap.find(name);
-        if (path != m_fileMap.end())
+        
+        if (!file)
         {
-          auto ext = path->second.extension().string();
-          auto size = fs::file_size(path->second);
-          file = make_shared<CachedFile>(path->second, getMimeType(ext),
-                                         size <= m_maxCachedFileSize, size);
-          m_fileCache.insert_or_assign(name, file);
-        }
-        else
-        {
-          file = findFileInDirectories(name);
+          auto path = m_fileMap.find(name);
+          if (path != m_fileMap.end())
+          {
+            auto ext = path->second.extension().string();
+            auto size = fs::file_size(path->second);
+            file = make_shared<CachedFile>(path->second, getMimeType(ext),
+                                           size <= m_maxCachedFileSize, size);
+            m_fileCache.insert_or_assign(name, file);
+          }
+          else
+          {
+            file = findFileInDirectories(name);
+          }
         }
       }
-
+      
       if (file)
       {
         if (acceptEncoding && acceptEncoding->find("gzip") != string::npos &&

src/mtconnect/sink/rest_sink/file_cache.hpp

@@ -148,5 +148,9 @@ namespace mtconnect::sink::rest_sink {
     std::map<std::string, std::string> m_mimeTypes;
     size_t m_maxCachedFileSize;
     size_t m_minCompressedFileSize;
+    
+    // Access control to the buffer
+    mutable std::recursive_mutex m_cacheLock;
+
   };
 }  // namespace mtconnect::sink::rest_sink