shm: implement first fit for header allocation

Author: ironMann

src/common/MemoryUtils.h

@@ -53,6 +53,11 @@ static constexpr const char *ENV_NOLOCK = "DATADIST_NO_MLOCK";
 static constexpr const char *ENV_SHM_PATH = "DATADIST_SHM_PATH";
 static constexpr const char *ENV_SHM_DELAY = "DATADIST_SHM_DELAY";
 
+enum RegionAllocStrategy {
+  eFindLongest,
+  eFindFirst
+};
+
 template<size_t ALIGN = 64>
 class RegionAllocatorResource
 {
@@ -61,8 +66,10 @@ class RegionAllocatorResource
   RegionAllocatorResource() = delete;
 
   RegionAllocatorResource(std::string pSegmentName, FairMQTransportFactory& pShmTrans,
-                          std::size_t pSize, std::uint64_t pRegionFlags = 0, bool pCanFail = false)
+                          std::size_t pSize, const RegionAllocStrategy pStrategy, std::uint64_t pRegionFlags = 0,
+                          bool pCanFail = false)
   : mSegmentName(pSegmentName),
+    mStrategy(pStrategy),
     mCanFail(pCanFail),
     mTransport(pShmTrans)
   {
@@ -379,31 +386,47 @@ class RegionAllocatorResource
       return false;
     }
 
-    auto lMaxIter = std::max_element(std::begin(mFreeRanges), std::end(mFreeRanges),
-      [](const auto& l, const auto& r) {
-        return (l.first.upper() - l.first.lower()) < (r.first.upper() - r.first.lower());
-      });
+    auto lMaxIter = mFreeRanges.end();
+
+    if (mStrategy == eFindFirst) {
+      for (auto lInt = mFreeRanges.begin(); lInt != mFreeRanges.end(); ++lInt) {
+        if (lInt->first.upper() - lInt->first.lower() >= pSize) {
+          lMaxIter = lInt;
+          break;
+        }
+      }
+    } else { /* eFindLongest */
+      lMaxIter = std::max_element(mFreeRanges.begin(), mFreeRanges.end(),
+        [](const auto& l, const auto& r) {
+          return (l.first.upper() - l.first.lower()) < (r.first.upper() - r.first.lower());
+        }
+      );
+    }
+
+    if (lMaxIter == mFreeRanges.end()) {
+      return false;
+    }
 
     // check if the size is adequate
-    const auto lFoudSize = lMaxIter->first.upper() - lMaxIter->first.lower();
-    if (pSize > lFoudSize) {
+    const auto lFoundSize = lMaxIter->first.upper() - lMaxIter->first.lower();
+    if (pSize > lFoundSize) {
       return false;
     }
 
     if (lMaxIter->second > 1) {
       EDDLOG("RegionAllocator BUG: Overlapping interval found: ptr={:p} length={} overlaps={}",
-        reinterpret_cast<char*>(lMaxIter->first.lower()), lFoudSize, lMaxIter->second);
+        reinterpret_cast<char*>(lMaxIter->first.lower()), lFoundSize, lMaxIter->second);
 
       // erase this segment
-      mFree -= lFoudSize;
+      mFree -= lFoundSize;
       assert (mFree > 0);
       mFreeRanges.erase(lMaxIter);
       return false;
     }
 
     // return the extent
     mStart = reinterpret_cast<char*>(lMaxIter->first.lower());
-    mLength = lFoudSize;
+    mLength = lFoundSize;
     mFreeRanges.erase(lMaxIter);
 
     {
@@ -448,6 +471,7 @@ class RegionAllocatorResource
   std::string mSegmentName;
   std::size_t mSegmentSize;
   std::atomic_bool mRunning = false;
+  RegionAllocStrategy mStrategy;
   bool mCanFail = false;
 
   FairMQTransportFactory &mTransport;

src/common/SubTimeFrameBuilder.cxx

@@ -39,6 +39,7 @@ SubTimeFrameReadoutBuilder::SubTimeFrameReadoutBuilder(MemoryResources &pMemRes,
     "O2HeadersRegion",
     *mMemRes.mShmTransport,
     std::size_t(512) << 20, /* good for 5s 3CRU @ 50Gbps, TODO: make configurable */
+    RegionAllocStrategy::eFindFirst,
     0, /* region flags ? */
     true /* Header alloc can fail with large FLP-DPL backpreassure  */
   );
@@ -305,13 +306,16 @@ SubTimeFrameFileBuilder::SubTimeFrameFileBuilder(MemoryResources &pMemRes,
     "O2HeadersRegion_FileSource",
     *mMemRes.mShmTransport,
     pHdrSegSize,
-    0
+    RegionAllocStrategy::eFindFirst,
+    0, /* GPU flags */
+    false /* cannot fail */
   );
 
   mMemRes.mDataMemRes = std::make_unique<RegionAllocatorResource<>>(
     "O2DataRegion_FileSource",
     *mMemRes.mShmTransport,
     pDataSegSize,
+    RegionAllocStrategy::eFindLongest,
     0 // TODO: GPU flags
   );
 
@@ -385,14 +389,18 @@ void TimeFrameBuilder::allocate_memory(const std::size_t pDataSegSize, const std
     "O2HeadersRegion",
     *mMemRes.mShmTransport,
     pHdrSegSize,
-    0 /* dont need registration flags for headers */
+    RegionAllocStrategy::eFindFirst,
+    0, /* GPU flags */
+    false /* cannot fail */
   );
 
   mMemRes.mDataMemRes = std::make_unique<RegionAllocatorResource<>>(
     "O2DataRegion_TimeFrame",
     *mMemRes.mShmTransport,
     pDataSegSize,
-    0 // TODO: GPU flags
+    RegionAllocStrategy::eFindLongest,
+    0, /* TODO: GPU flags */
+    false /* cannot fail */
   );
 
   mMemRes.start();
@@ -450,6 +458,9 @@ void TimeFrameBuilder::adaptHeaders(SubTimeFrame *pStf)
               o2::framework::DataProcessingHeader{pStf->header().mId}
             );
 
+            WDDLOG_RL(5000, "Reallocation of Header messages is not optimal. orig_size={} new_size={}",
+              lHeader->GetSize(), lStack.size());
+
             lStfDataIter.mHeader = newHeaderMessage(reinterpret_cast<char*>(lStack.data()), lStack.size());
             if (!lStfDataIter.mHeader) {
               return;