tfbuilder: synchronize on memory allocations

Author: ironMann

src/TfBuilder/TfBuilderDevice.cxx

@@ -46,7 +46,7 @@ TfBuilderDevice::~TfBuilderDevice()
 
 void TfBuilderDevice::Init()
 {
-  mMemI = std::make_unique<MemoryResources>(this->AddTransport(fair::mq::Transport::SHM));
+  mMemI = std::make_unique<SyncMemoryResources>(this->AddTransport(fair::mq::Transport::SHM));
 }
 
 void TfBuilderDevice::Reset()

src/TfBuilder/TfBuilderDevice.h

@@ -122,7 +122,7 @@ class TfBuilderDevice : public DataDistDevice,
   bool dplEnabled() const noexcept { return mDplEnabled; }
 
   /// Memory region singletons
-  std::unique_ptr<MemoryResources> mMemI;
+  std::unique_ptr<SyncMemoryResources> mMemI;
 
   /// Configuration
   std::string mDplChannelName;

src/common/include/MemoryUtils.h

@@ -477,7 +477,7 @@ class MemoryResources {
   explicit MemoryResources(std::shared_ptr<FairMQTransportFactory> pShmTransport)
   : mShmTransport(pShmTransport) { }
 
-  ~MemoryResources() {
+  virtual ~MemoryResources() {
     // make sure to delete regions before dropping the transport
     mHeaderMemRes.reset();
     mDataMemRes.reset();
@@ -522,6 +522,35 @@ class MemoryResources {
   bool mRunning = true;
 };
 
+
+class SyncMemoryResources : public MemoryResources {
+public:
+  SyncMemoryResources() = delete;
+  explicit SyncMemoryResources(std::shared_ptr<FairMQTransportFactory> pShmTransport)
+  : MemoryResources(pShmTransport) { }
+
+  virtual ~SyncMemoryResources() {}
+
+  inline
+  FairMQMessagePtr newHeaderMessage(const std::size_t pSize) {
+    assert(mHeaderMemRes);
+    std::scoped_lock lock(mHdrLock);
+    return mHeaderMemRes->NewFairMQMessage(pSize);
+  }
+
+  inline
+  FairMQMessagePtr newDataMessage(const std::size_t pSize) {
+    assert(mDataMemRes);
+    std::scoped_lock lock(mDataLock);
+    return mDataMemRes->NewFairMQMessage(pSize);
+  }
+
+
+private:
+  std::mutex mHdrLock;
+  std::mutex mDataLock;
+};
+
 }
 } /* o2::DataDistribution */