shmem: fix osx race condition

Author: Grok Compression

src/lib/codec/shared/Messenger.h

@@ -321,7 +321,11 @@ struct Synch
 };
 struct SharedMemoryManager
 {
-  static bool initShm(const std::string& name, size_t len, grk_handle* hMapFile, char** buffer)
+  // Windows: CreateFileMapping transparently opens or creates the mapping,
+  // so isCreator is unused — stale segments from a previous crash are handled
+  // automatically, unlike POSIX where O_EXCL + shm_unlink recovery is needed.
+  static bool initShm(const std::string& name, size_t len, grk_handle* hMapFile, char** buffer,
+                      [[maybe_unused]] bool isCreator)
   {
     *hMapFile = CreateFileMapping(INVALID_HANDLE_VALUE, // use paging file
                                   NULL, // default security
@@ -447,7 +451,8 @@ struct Synch
 };
 struct SharedMemoryManager
 {
-  static bool initShm(const std::string& name, size_t len, grk_handle* shm_fd, char** buffer)
+  static bool initShm(const std::string& name, size_t len, grk_handle* shm_fd, char** buffer,
+                      bool isCreator)
   {
     if(*shm_fd)
       return true;
@@ -466,38 +471,63 @@ struct SharedMemoryManager
       return false;
     }
 
-    *shm_fd = shm_open(name.c_str(), O_CREAT | O_RDWR, 0666);
-    if(*shm_fd < 0)
+    if(isCreator)
     {
-      getMessengerLogger()->error("Error opening shared memory: %s", strerror(errno));
-      return false;
+      *shm_fd = shm_open(name.c_str(), O_CREAT | O_EXCL | O_RDWR, 0666);
+      if(*shm_fd < 0 && errno == EEXIST)
+      {
+        // stale segment from a previous crash — remove and retry
+        shm_unlink(name.c_str());
+        *shm_fd = shm_open(name.c_str(), O_CREAT | O_EXCL | O_RDWR, 0666);
+      }
+      if(*shm_fd < 0)
+      {
+        getMessengerLogger()->error("Error creating shared memory %s: %s", name.c_str(),
+                                    strerror(errno));
+        *shm_fd = 0;
+        return false;
+      }
+      int rc = ftruncate(*shm_fd, (off_t)len);
+      if(rc)
+      {
+        getMessengerLogger()->error("Error truncating shared memory to %" PRIu64 " bytes: %s",
+                                    (uint64_t)len, strerror(errno));
+        close(*shm_fd);
+        shm_unlink(name.c_str());
+        *shm_fd = 0;
+        return false;
+      }
     }
-    int rc = ftruncate(*shm_fd, (off_t)len);
-    if(rc)
+    else
     {
-      getMessengerLogger()->error("Error truncating shared memory to %" PRIu64 " bytes: %s",
-                                  (uint64_t)len, strerror(errno));
-      rc = close(*shm_fd);
-      if(rc)
-        getMessengerLogger()->error("Error closing shared memory: %s", strerror(errno));
-      rc = shm_unlink(name.c_str());
-      // 2 == No such file or directory
-      if(rc && errno != 2)
-        getMessengerLogger()->error("Error unlinking shared memory: %s", strerror(errno));
-      return false;
+      // Non-creator: open existing segment, retry briefly if not yet created
+      const int maxRetries = 50;
+      const int retryDelayMs = 10;
+      for(int attempt = 0; attempt < maxRetries; ++attempt)
+      {
+        *shm_fd = shm_open(name.c_str(), O_RDWR, 0666);
+        if(*shm_fd >= 0)
+          break;
+        if(errno != ENOENT || attempt == maxRetries - 1)
+        {
+          getMessengerLogger()->error("Error opening shared memory %s: %s", name.c_str(),
+                                      strerror(errno));
+          return false;
+        }
+        std::this_thread::sleep_for(std::chrono::milliseconds(retryDelayMs));
+      }
     }
-    *buffer = static_cast<char*>(mmap(0, len, PROT_WRITE, MAP_SHARED, *shm_fd, 0));
+
+    *buffer = static_cast<char*>(mmap(0, len, PROT_READ | PROT_WRITE, MAP_SHARED, *shm_fd, 0));
     if(*buffer == MAP_FAILED)
     {
       *buffer = nullptr;
-      getMessengerLogger()->error("Error mapping shared memory: %s", strerror(errno));
-      rc = close(*shm_fd);
-      if(rc)
-        getMessengerLogger()->error("Error closing shared memory: %s", strerror(errno));
-      rc = shm_unlink(name.c_str());
-      // 2 == No such file or directory
-      if(rc && errno != 2)
-        getMessengerLogger()->error("Error unlinking shared memory: %s", strerror(errno));
+      getMessengerLogger()->error("Error mapping shared memory %s: %s", name.c_str(),
+                                  strerror(errno));
+      close(*shm_fd);
+      if(isCreator)
+        shm_unlink(name.c_str());
+      *shm_fd = 0;
     }
 
     return *buffer != nullptr;
@@ -674,10 +704,7 @@ static void processorThread(Messenger* messenger, std::function<void(std::string
 
 struct Messenger
 {
-  explicit Messenger(MessengerInit init)
-      : running(true), initialized_(false), shutdown_(false), init_(init), outboundSynch_(nullptr),
-        inboundSynch_(nullptr), uncompressed_buffer_(nullptr), compressed_buffer_(nullptr),
-        uncompressed_fd_(0), compressed_fd_(0)
+  explicit Messenger(MessengerInit init) : init_(init)
   {
     if(!isClient())
     {
@@ -757,15 +784,6 @@ struct Messenger
   }
   bool initBuffers(void)
   {
-#ifndef _WIN32
-    // clean up in case of previous crash
-    if(init_.firstLaunch(init_.isClient_))
-    {
-      // shm_unlink(grokUncompressedBuf.c_str());
-      // shm_unlink(grokCompressedBuf.c_str());
-    }
-#endif
-
     char temp[512];
     sprintf(temp,
             "Initializing shared memory buffers: num frames %zu, "
@@ -775,9 +793,9 @@ struct Messenger
     getMessengerLogger()->info(temp);
     if(init_.uncompressedFrameSize_)
     {
-      bool rc = SharedMemoryManager::initShm(grokUncompressedBuf,
-                                             init_.uncompressedFrameSize_ * init_.numFrames_,
-                                             &uncompressed_fd_, &uncompressed_buffer_);
+      bool rc = SharedMemoryManager::initShm(
+          grokUncompressedBuf, init_.uncompressedFrameSize_ * init_.numFrames_, &uncompressed_fd_,
+          &uncompressed_buffer_, !init_.isClient_);
       if(!rc)
         return false;
 
@@ -787,9 +805,9 @@ struct Messenger
     }
     if(init_.compressedFrameSize_)
     {
-      bool rc = SharedMemoryManager::initShm(grokCompressedBuf,
-                                             init_.compressedFrameSize_ * init_.numFrames_,
-                                             &compressed_fd_, &compressed_buffer_);
+      bool rc = SharedMemoryManager::initShm(
+          grokCompressedBuf, init_.compressedFrameSize_ * init_.numFrames_, &compressed_fd_,
+          &compressed_buffer_, !init_.isClient_);
       if(!rc)
         return false;
 
@@ -939,9 +957,9 @@ struct Messenger
 
     return (uint8_t*)(compressed_buffer_ + frameId * init_.compressedFrameSize_);
   }
-  std::atomic_bool running;
-  bool initialized_;
-  bool shutdown_;
+  std::atomic_bool running{true};
+  bool initialized_ = false;
+  bool shutdown_ = false;
   MessengerBlockingQueue<std::string> sendQueue;
   MessengerBlockingQueue<std::string> receiveQueue;
   MessengerBlockingQueue<BufferSrc> availableBuffers_;
@@ -956,17 +974,17 @@ struct Messenger
 
 private:
   std::thread outbound;
-  Synch* outboundSynch_;
+  Synch* outboundSynch_ = nullptr;
 
   std::thread inbound;
-  Synch* inboundSynch_;
+  Synch* inboundSynch_ = nullptr;
 
   std::vector<std::thread> processors_;
-  char* uncompressed_buffer_;
-  char* compressed_buffer_;
+  char* uncompressed_buffer_ = nullptr;
+  char* compressed_buffer_ = nullptr;
 
-  grk_handle uncompressed_fd_;
-  grk_handle compressed_fd_;
+  grk_handle uncompressed_fd_ = 0;
+  grk_handle compressed_fd_ = 0;
 };
 
 /*************************** I/O Threads *******************************/
@@ -975,7 +993,8 @@ static void outboundThread(Messenger* messenger, const std::string& sendBuf, Syn
   grk_handle shm_fd = 0;
   char* send_buffer = nullptr;
 
-  if(!SharedMemoryManager::initShm(sendBuf, messageBufferLen, &shm_fd, &send_buffer))
+  if(!SharedMemoryManager::initShm(sendBuf, messageBufferLen, &shm_fd, &send_buffer,
+                                   !messenger->isClient()))
     return;
   while(messenger->running)
   {
@@ -998,7 +1017,8 @@ static void inboundThread(Messenger* messenger, const std::string& receiveBuf, S
   grk_handle shm_fd = 0;
   char* receive_buffer = nullptr;
 
-  if(!SharedMemoryManager::initShm(receiveBuf, messageBufferLen, &shm_fd, &receive_buffer))
+  if(!SharedMemoryManager::initShm(receiveBuf, messageBufferLen, &shm_fd, &receive_buffer,
+                                   !messenger->isClient()))
     return;
   while(messenger->running)
   {

tests/GrkMessengerTest.cpp

@@ -365,7 +365,7 @@ static void testSharedMemoryAllocation()
   grk_handle fd = 0;
   char* buf = nullptr;
 
-  bool rc = SharedMemoryManager::initShm("/grk_test_shm_alloc", bufLen, &fd, &buf);
+  bool rc = SharedMemoryManager::initShm("/grk_test_shm_alloc", bufLen, &fd, &buf, true);
   TEST_ASSERT(rc, "initShm should succeed");
   TEST_ASSERT(buf != nullptr, "buffer should be non-null");
 
@@ -380,6 +380,36 @@ static void testSharedMemoryAllocation()
   TEST_PASS("testSharedMemoryAllocation");
 }
 
+#ifndef _WIN32
+// Test that creator handles stale shm from a previous crash (EEXIST → unlink → retry)
+static void testSharedMemoryCrashRecovery()
+{
+  const char* name = "/grk_test_crash_recovery";
+  const size_t bufLen = 4096;
+
+  // simulate stale segment left by a crashed process
+  shm_unlink(name);
+  int stale_fd = shm_open(name, O_CREAT | O_RDWR, 0666);
+  TEST_ASSERT(stale_fd >= 0, "stale shm_open should succeed");
+  ftruncate(stale_fd, (off_t)bufLen);
+  close(stale_fd);
+
+  // creator should recover: unlink stale segment and create fresh
+  grk_handle fd = 0;
+  char* buf = nullptr;
+  bool rc = SharedMemoryManager::initShm(name, bufLen, &fd, &buf, true);
+  TEST_ASSERT(rc, "initShm should recover from stale segment");
+  TEST_ASSERT(buf != nullptr, "buffer should be non-null after recovery");
+
+  memset(buf, 0xCD, bufLen);
+  TEST_ASSERT((uint8_t)buf[0] == 0xCD, "written data should persist after recovery");
+
+  rc = SharedMemoryManager::deinitShm(name, bufLen, fd, &buf);
+  TEST_ASSERT(rc, "deinitShm should succeed");
+  TEST_PASS("testSharedMemoryCrashRecovery");
+}
+#endif
+
 // Test frame accessor methods
 static void testFrameAccessors()
 {
@@ -458,6 +488,18 @@ static void testClientInitBufferQueue()
   const size_t compressedSize = 64;
   const size_t numFrames = 2;
 
+#ifndef _WIN32
+  // Pre-create data buffers (simulates server creating before client opens)
+  shm_unlink(grokUncompressedBuf.c_str());
+  shm_unlink(grokCompressedBuf.c_str());
+  grk_handle pre_uc_fd = 0, pre_c_fd = 0;
+  char *pre_uc = nullptr, *pre_c = nullptr;
+  SharedMemoryManager::initShm(grokUncompressedBuf, uncompressedSize * numFrames, &pre_uc_fd,
+                               &pre_uc, true);
+  SharedMemoryManager::initShm(grokCompressedBuf, compressedSize * numFrames, &pre_c_fd, &pre_c,
+                               true);
+#endif
+
   MessengerInit init(true, "Global\\grk_test3_out", "Global\\grk_test3_out_sent",
                      "Global\\grk_test3_out_ready", "Global\\grk_test3_in",
                      "Global\\grk_test3_in_sent", "Global\\grk_test3_in_ready", proc, 0, 0, 0, 0);
@@ -474,6 +516,14 @@ static void testClientInitBufferQueue()
     TEST_ASSERT(m.availableBuffers_.pop(src), "should get buffer");
     TEST_ASSERT(src.frameId_ == i, "frameId should match");
   }
+
+#ifndef _WIN32
+  munmap(pre_uc, uncompressedSize * numFrames);
+  close(pre_uc_fd);
+  munmap(pre_c, compressedSize * numFrames);
+  close(pre_c_fd);
+#endif
+
   TEST_PASS("testClientInitBufferQueue");
 }
 
@@ -536,6 +586,9 @@ int GrkMessengerTest::main(int argc, char** argv)
   // Full protocol loopback test
   fprintf(stdout, "\n[Shared Memory / Buffers]\n");
   testSharedMemoryAllocation();
+#ifndef _WIN32
+  testSharedMemoryCrashRecovery();
+#endif
   testFrameAccessors();
   testServerBufferQueue();
   testClientInitBufferQueue();