shared memory: add interface

Author: Grok Compression

INSTALL.md

@@ -221,6 +221,33 @@ The `CUDA_MODULE_LOADING=EAGER` environment variable is set automatically by the
 Tests cover: single/batch compress and decompress, lossy/lossless, bit depths 8/10/12/16,
 mono and RGB, cinema 2K profile, and CPU-only fallback (`-G -2`).
 
+### Shared Memory Batch Tests
+
+The SHM batch tests verify the shared-memory inter-process protocol used for batch
+compression and decompression. Each test spawns the codec binary as a child process and
+communicates via POSIX shared memory and semaphores.
+
+To run SHM batch tests via `ctest`:
+
+    $  cd /PATH/TO/BUILD
+    $  ctest -R grk_shm_batch -V
+
+Individual tests:
+
+    $  ctest -R grk_shm_batch_compress -V
+    $  ctest -R grk_shm_batch_decompress -V
+
+Or run the binaries directly:
+
+    $  ./bin/grk_shm_batch_compress ./bin/grk_compress
+    $  ./bin/grk_shm_batch_decompress ./bin/grk_decompress
+
+Each test compresses/decompresses 4 frames (64x64, 3-channel, 12-bit) through the SHM
+protocol and validates pixel-perfect round-trip fidelity.
+
+Note: these two tests share POSIX named semaphores and SHM segments, so they must not
+run in parallel. The CTest configuration uses `RESOURCE_LOCK` to enforce this automatically.
+
 ## macOS
 
 macOS builds are configured similar to *NIX builds.

src/lib/codec/CMakeLists.txt

@@ -20,6 +20,7 @@ include_directories(
   ${CMAKE_CURRENT_SOURCE_DIR}/formats/fileio
   ${CMAKE_CURRENT_SOURCE_DIR}/common
   ${CMAKE_CURRENT_SOURCE_DIR}/apps
+  ${CMAKE_CURRENT_SOURCE_DIR}/shared
 )
 
 set(GROK_CODEC_SRCS
@@ -68,8 +69,11 @@ if(GROK_HAVE_LIBJPEG)
 	target_link_libraries(${GROK_CODEC_NAME} PRIVATE ${JPEG_LIBNAME})
 endif()
 
-if(UNIX)
-  target_link_libraries(${GROK_CODEC_NAME} PUBLIC ${CMAKE_THREAD_LIBS_INIT})
+find_package(Threads REQUIRED)
+target_link_libraries(${GROK_CODEC_NAME} PUBLIC Threads::Threads)
+# shared memory (shm_open, sem_open) used by Messenger.h
+if(CMAKE_SYSTEM_NAME STREQUAL "Linux")
+  target_link_libraries(${GROK_CODEC_NAME} PRIVATE rt)
 endif()
 set_target_properties(${GROK_CODEC_NAME} PROPERTIES ${GROK_LIBRARY_PROPERTIES})
 target_compile_options(${GROK_CODEC_NAME} PRIVATE ${GROK_COMPILE_OPTIONS})
@@ -96,3 +100,6 @@ install(TARGETS ${INSTALL_LIBS}
 install(FILES grok_codec.h
   DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}/${GROK_INSTALL_SUBDIR} COMPONENT Headers
 )
+install(FILES shared/Messenger.h
+  DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}/${GROK_INSTALL_SUBDIR} COMPONENT Headers
+)

src/lib/codec/apps/GrkCompress.cpp

@@ -62,6 +62,7 @@ using namespace grk;
 #include "grk_string.h"
 #include "spdlog/sinks/basic_file_sink.h"
 #include "GrkCompress.h"
+#include "Messenger.h"
 
 void exit_func()
 {
@@ -361,6 +362,13 @@ int GrkCompress::main(int argc, const char** argv, grk_image* in_image, grk_stre
       goto cleanup;
     }
 
+    // CPU shared memory batch compress
+    if(initParams.parameters.shared_memory_interface)
+    {
+      success = shmBatchCompress(&initParams);
+      goto cleanup;
+    }
+
     // MJ2 directory encode: all input images → single .mj2 file
     if(initParams.parameters.cod_format == GRK_FMT_MJ2 && initParams.inputFolder.set_imgdir)
     {
@@ -506,6 +514,195 @@ int GrkCompress::main(int argc, const char** argv, grk_image* in_image, grk_stre
   return success;
 }
 
+int GrkCompress::shmBatchCompress(CompressInitParams* initParams)
+{
+  using namespace grk_plugin;
+
+  setUpSignalHandler();
+
+  // parse "GRK_MSGR_BATCH_IMAGE,width,stride,height,spp,depth" from batch_src
+  std::string batchSrc(initParams->inputFolder.imgdirpath);
+  Msg batchMsg(batchSrc);
+  auto tag = batchMsg.next();
+  if(tag != GRK_MSGR_BATCH_IMAGE)
+  {
+    spdlog::error("shmBatchCompress: expected {} tag, got '{}'", GRK_MSGR_BATCH_IMAGE, tag);
+    return 1;
+  }
+  uint32_t imgWidth = batchMsg.nextUint();
+  uint32_t imgStride = batchMsg.nextUint();
+  (void)imgStride;
+  uint32_t imgHeight = batchMsg.nextUint();
+  uint32_t imgSpp = batchMsg.nextUint();
+  uint32_t imgDepth = batchMsg.nextUint();
+
+  size_t uncompressedFrameSize = Messenger::uncompressedFrameSize(imgWidth, imgHeight, imgSpp);
+  // conservative upper bound for compressed frame size
+  size_t compressedFrameSize = (size_t)imgWidth * imgHeight * imgSpp * ((imgDepth + 7) / 8) * 3 / 2;
+  if(compressedFrameSize < 4096)
+    compressedFrameSize = 4096;
+  const size_t numFrames = 8;
+
+  setMessengerLogger(new MessengerLogger("[SHM-CPU] "));
+
+  auto* parameters = &initParams->parameters;
+
+  // shared state for flush synchronization
+  std::atomic<uint32_t> framesProcessed{0};
+  std::atomic<uint32_t> flushTarget{0};
+  std::atomic<bool> flushRequested{false};
+  std::mutex flushMutex;
+  std::condition_variable flushCv;
+
+  auto processor = [&](std::string message) {
+    Msg msg(message);
+    auto msgTag = msg.next();
+
+    if(msgTag == GRK_MSGR_BATCH_SUBMIT_UNCOMPRESSED)
+    {
+      auto clientFrameId = msg.nextUint();
+      auto uncompressedFrameId = msg.nextUint();
+      uint8_t* srcPixels = messenger_->getUncompressedFrame(uncompressedFrameId);
+      if(!srcPixels)
+      {
+        spdlog::error("shmBatchCompress: null uncompressed frame {}", uncompressedFrameId);
+        return;
+      }
+
+      // create grk_image from raw pixel data
+      auto cmptparms = std::make_unique<grk_image_comp[]>(imgSpp);
+      for(uint32_t c = 0; c < imgSpp; ++c)
+      {
+        memset(&cmptparms[c], 0, sizeof(grk_image_comp));
+        cmptparms[c].dx = 1;
+        cmptparms[c].dy = 1;
+        cmptparms[c].w = imgWidth;
+        cmptparms[c].h = imgHeight;
+        cmptparms[c].prec = (uint8_t)imgDepth;
+        cmptparms[c].sgnd = false;
+      }
+      GRK_COLOR_SPACE clrspc = (imgSpp >= 3) ? GRK_CLRSPC_SRGB : GRK_CLRSPC_GRAY;
+      grk_image* image = grk_image_new((uint16_t)imgSpp, cmptparms.get(), clrspc, true);
+      if(!image)
+      {
+        spdlog::error("shmBatchCompress: failed to create image for frame {}", clientFrameId);
+        messenger_->reclaimUncompressed(uncompressedFrameId);
+        messenger_->send(GRK_MSGR_BATCH_PROCESSED_UNCOMPRESSED, uncompressedFrameId);
+        return;
+      }
+
+      image->x0 = 0;
+      image->y0 = 0;
+      image->x1 = imgWidth;
+      image->y1 = imgHeight;
+
+      // copy pixel data from SHM (interleaved 16-bit) into planar int32_t components
+      auto* src16 = reinterpret_cast<const uint16_t*>(srcPixels);
+      for(uint32_t y = 0; y < imgHeight; ++y)
+      {
+        for(uint32_t x = 0; x < imgWidth; ++x)
+        {
+          for(uint32_t c = 0; c < imgSpp; ++c)
+          {
+            auto* data = static_cast<int32_t*>(image->comps[c].data);
+            data[y * imgWidth + x] = (int32_t)src16[(y * imgWidth + x) * imgSpp + c];
+          }
+        }
+      }
+
+      // done reading uncompressed buffer — release it back to client
+      messenger_->send(GRK_MSGR_BATCH_PROCESSED_UNCOMPRESSED, uncompressedFrameId);
+
+      // compress to memory buffer
+      grk_cparameters localParams = *parameters;
+      if(localParams.mct == 255)
+        localParams.mct = (imgSpp >= 3) ? 1 : 0;
+
+      // get a compressed buffer slot from the server queue
+      BufferSrc compressedSlot;
+      if(!messenger_->availableBuffers_.waitAndPop(compressedSlot))
+      {
+        spdlog::error("shmBatchCompress: no compressed buffer available");
+        grk_object_unref(&image->obj);
+        return;
+      }
+
+      grk_stream_params streamParams{};
+      streamParams.buf = compressedSlot.framePtr_;
+      streamParams.buf_len = compressedFrameSize;
+
+      grk_object* codec = grk_compress_init(&streamParams, &localParams, image);
+      uint64_t compressedLen = 0;
+      if(codec)
+      {
+        compressedLen = grk_compress(codec, nullptr);
+        grk_object_unref(codec);
+      }
+      grk_object_unref(&image->obj);
+
+      if(!compressedLen)
+      {
+        spdlog::error("shmBatchCompress: compression failed for frame {}", clientFrameId);
+        messenger_->availableBuffers_.push(compressedSlot);
+        return;
+      }
+
+      messenger_->send(GRK_MSGR_BATCH_SUBMIT_COMPRESSED, clientFrameId, compressedSlot.frameId_,
+                       compressedLen);
+
+      uint32_t count = ++framesProcessed;
+      if(flushRequested.load() && count >= flushTarget.load())
+      {
+        std::lock_guard<std::mutex> lk(flushMutex);
+        flushCv.notify_all();
+      }
+    }
+    else if(msgTag == GRK_MSGR_BATCH_PROCESSSED_COMPRESSED)
+    {
+      auto compressedFrameId = msg.nextUint();
+      messenger_->reclaimCompressed(compressedFrameId);
+    }
+    else if(msgTag == GRK_MSGR_BATCH_FLUSH)
+    {
+      auto enqueuedCount = msg.nextUint();
+      flushTarget.store(enqueuedCount);
+      flushRequested.store(true);
+      {
+        std::unique_lock<std::mutex> lk(flushMutex);
+        flushCv.wait(lk, [&] { return framesProcessed.load() >= enqueuedCount; });
+      }
+    }
+    else if(msgTag == GRK_MSGR_BATCH_SHUTDOWN)
+    {
+      messenger_->running = false;
+    }
+  };
+
+  uint32_t num_threads = std::thread::hardware_concurrency();
+  if(num_threads == 0)
+    num_threads = 4;
+
+  MessengerInit init(false, // server
+                     grokToClientMessageBuf, grokSentSynch, clientReceiveReadySynch,
+                     clientToGrokMessageBuf, clientSentSynch, grokReceiveReadySynch, processor,
+                     num_threads, uncompressedFrameSize, compressedFrameSize, numFrames);
+
+  messenger_ = new Messenger(init);
+
+  // tell client the buffer layout
+  messenger_->send(GRK_MSGR_BATCH_COMPRESS_INIT, imgWidth, imgStride, imgHeight, imgSpp, imgDepth,
+                   compressedFrameSize, numFrames);
+
+  // wait until shutdown
+  while(messenger_->running)
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+
+  delete messenger_;
+  messenger_ = nullptr;
+
+  return 0;
+}
+
 int GrkCompress::pluginBatchCompress(CompressInitParams* initParams)
 {
   setUpSignalHandler();

src/lib/codec/apps/GrkCompress.h

@@ -20,6 +20,11 @@
 #include "common.h"
 #include "IImageFormat.h"
 
+namespace grk_plugin
+{
+struct Messenger;
+}
+
 namespace grk
 {
 struct CompressInitParams
@@ -41,15 +46,18 @@ struct CompressInitParams
 class GrkCompress
 {
 public:
-  GrkCompress(void) = default;
+  GrkCompress(void) : messenger_(nullptr) {}
   ~GrkCompress(void) = default;
   int main(int argc, const char* argv[], grk_image* in_image, grk_stream_params* out_buffer);
 
 private:
+  int shmBatchCompress(CompressInitParams* initParams);
   int pluginBatchCompress(CompressInitParams* initParams);
   GrkRC pluginMain(int argc, const char* argv[], CompressInitParams* initParams);
   GrkRC parseCommandLine(int argc, const char* argv[], CompressInitParams* initParams);
   int compress(const std::string& inputFile, CompressInitParams* initParams);
+
+  grk_plugin::Messenger* messenger_;
 };
 
 } // namespace grk