shmem: use bunch reclaim method of fmq unmanaged region

Author: ironMann

src/ReadoutEmulator/ReadoutDevice.cxx

@@ -61,8 +61,10 @@ void ReadoutDevice::InitTask()
   mDataRegion = NewUnmanagedRegionFor(
     mOutChannelName, 0,
     mDataRegionSize + mSuperpageSize,
-    [this](void* data, size_t size, void* /* hint */) { // callback to be called when message buffers no longer needed by transport
-      mCruMemoryHandler->put_data_buffer(static_cast<char*>(data), size);
+    [this](const std::vector<FairMQRegionBlock>& pBlkVect) { // callback to be called when message buffers no longer needed by transport
+      for (const auto &lBlk : pBlkVect) {
+          mCruMemoryHandler->put_data_buffer(static_cast<char*>(lBlk.ptr), lBlk.size);
+      }
     });
 
   DDLOG(fair::Severity::INFO) << "Memory regions created";

src/common/include/MemoryUtils.h

@@ -97,9 +97,13 @@ class FMQUnsynchronizedPoolMemoryResource : public boost::container::pmr::memory
     mRegion = pChan.Transport()->CreateUnmanagedRegion(
       pSize,
       pRegionFlags,
-      [this](void* pRelData, size_t pRelSize, void* /* hint */) {
+      [this](const std::vector<FairMQRegionBlock>& pBlkVect) {
         // callback to be called when message buffers no longer needed by transport
-        reclaimSHMMessage(pRelData, pRelSize);
+        std::scoped_lock lock(mReclaimLock);
+
+        for (const auto &lBlk : pBlkVect) {
+          reclaimSHMMessage(lBlk.ptr, lBlk.size);
+        }
       },
       lSegmentRoot.c_str(),
       lMapFlags
@@ -217,8 +221,6 @@ class FMQUnsynchronizedPoolMemoryResource : public boost::container::pmr::memory
   {
     (void) pSize;
     assert (pSize == mObjectSize);
-
-    std::scoped_lock lock(mReclaimLock);
     mReclaimedObjects.push_back(pData);
   }
 
@@ -290,9 +292,13 @@ class RegionAllocatorResource : public boost::container::pmr::memory_resource
     mRegion = pChan.Transport()->CreateUnmanagedRegion(
       pSize,
       pRegionFlags,
-      [this](void* pRelData, size_t pRelSize, void* /* hint */) {
-        // callback to be called when message buffers no longer needed by transport
-        reclaimSHMMessage(pRelData, pRelSize);
+      [this](const std::vector<FairMQRegionBlock>& pBlkVect) {
+        // callback to be called when message buffers no longer needed by transports
+        std::scoped_lock lock(mReclaimLock);
+
+        for (const auto &lBlk : pBlkVect) {
+          reclaimSHMMessage(lBlk.ptr, lBlk.size);
+        }
       },
       lSegmentRoot.c_str(),
       lMapFlags
@@ -336,7 +342,7 @@ class RegionAllocatorResource : public boost::container::pmr::memory_resource
 
   virtual void* do_allocate(std::size_t pSize, std::size_t /* pAlign */) override final
   {
-    unsigned long lAllocAttempt= 0;
+    unsigned long lAllocAttempt = 0;
 
     // align up
     pSize = align_size_up(pSize);
@@ -369,6 +375,7 @@ class RegionAllocatorResource : public boost::container::pmr::memory_resource
   virtual void do_deallocate(void *pAddr, std::size_t pSize, std::size_t) override final
   {
     // Objects are only freed through SHM message reclaim callback.
+    std::scoped_lock lock(mReclaimLock);
     reclaimSHMMessage(pAddr, pSize);
   }
 
@@ -394,16 +401,17 @@ class RegionAllocatorResource : public boost::container::pmr::memory_resource
 
   bool try_reclaim(const std::size_t pSize) {
     // First declare any leftover memory as free
+    std::scoped_lock lock(mReclaimLock);
+
     if (mLength > 0) {
       assert(mStart != nullptr);
+      // NOTE: caller must hold mReclaimLock lock
       reclaimSHMMessage(mStart, mLength);
       // invalidate the working extent
       mLength = 0;
       mStart = nullptr;
     }
 
-    std::scoped_lock lock(mReclaimLock);
-
     if (mFrees.empty()) {
       return false;
     }
@@ -432,8 +440,6 @@ class RegionAllocatorResource : public boost::container::pmr::memory_resource
 
     const char *lData = reinterpret_cast<const char*>(pData);
 
-    std::scoped_lock lock(mReclaimLock);
-
     // push object to the free map. Try to merge nodes
     const auto lIter = mFrees.lower_bound(lData);
     bool lInserted = false;