shm: use shm transport to construct shm regions and messages

ironMann · ironMann · commit 47b1076a078a · 2020-08-18T18:19:25.000+02:00
diff --git a/src/StfBuilder/StfBuilderDevice.cxx b/src/StfBuilder/StfBuilderDevice.cxx
@@ -65,7 +65,7 @@ void StfBuilderDevice::Init()
 {
   DDLOGF(fair::Severity::DEBUG, "StfBuilderDevice::Init()");
   mI = std::make_unique<StfBuilderInstance>();
-  mMemI = std::make_unique<MemoryResources>();
+  mMemI = std::make_unique<MemoryResources>(this->AddTransport(fair::mq::Transport::SHM));
 
   I().mFileSource = std::make_unique<SubTimeFrameFileSource>(*this, eStfFileSourceOut);
   I().mReadoutInterface = std::make_unique<StfInputInterface>(*this);
@@ -189,31 +189,6 @@ void StfBuilderDevice::InitTask()
     DDLOGF(fair::Severity::WARNING, "Running in standalone mode and with STF file sink disabled. Data will be lost.");
   }
 
-    // channel for FileSource: stf or dpl, or generic one in case of standalone
-  if (isStandalone()) {
-    // create default FMQ shm channel
-    auto lTransportFactory = FairMQTransportFactory::CreateTransportFactory("shmem", "", GetConfig());
-    if (!lTransportFactory) {
-      DDLOGF(fair::Severity::ERROR, "Creating transport factory failed!");
-      exit(-1);
-    }
-
-    I().mStandaloneChannel = std::make_unique<FairMQChannel>(
-      "standalone-chan[0]" ,  // name
-      "pair",                 // type
-      "bind",                 // method
-      "ipc:///tmp/standalone-chan-stfb", // address
-      lTransportFactory
-    );
-
-    // mStandaloneChannel.Init();
-    I().mStandaloneChannel->Init();
-    // mStandaloneChannel->BindEndpoint("ipc:///tmp/standalone-chan");
-    I().mStandaloneChannel->Validate();
-  }
-
-  DDLOGF(fair::Severity::info, "Sending data to channel: {}", getOutputChannel().GetName());
-
   // try to see if channels have been configured
   {
     if (!I().mFileSource->enabled()) {
@@ -241,8 +216,7 @@ void StfBuilderDevice::InitTask()
   I().mFileSink->start();
 
   // start file source
-  // channel for FileSource: stf or dpl, or generic one in case of standalone
-  I().mFileSource->start(getOutputChannel(), MemI(), I().mDplEnabled);
+  I().mFileSource->start(MemI(), I().mDplEnabled);
 
   // start a thread for readout process
   if (!I().mFileSource->enabled()) {
@@ -305,12 +279,11 @@ void StfBuilderDevice::StfOutputThread()
   std::unique_ptr<InterleavedHdrDataSerializer> lStfSerializer;
   std::unique_ptr<StfToDplAdapter> lStfDplAdapter;
 
-  // cannot get the channels in standalone mode
-  auto& lOutputChan = getOutputChannel();
-
-  DDLOGF(fair::Severity::info, "StfOutputThread: sending data to channel: {}", lOutputChan.GetName());
-
   if (!isStandalone()) {
+    // cannot get the channels in standalone mode
+    auto& lOutputChan = getOutputChannel();
+    DDLOGF(fair::Severity::info, "StfOutputThread: sending data to channel: {}", lOutputChan.GetName());
+
     if (!dplEnabled()) {
       lStfSerializer = std::make_unique<InterleavedHdrDataSerializer>(lOutputChan);
     } else {
@@ -329,8 +302,8 @@ void StfBuilderDevice::StfOutputThread()
     // decrement the stf counter
     I().mNumStfs--;
 
-    DDLOGF_RL(2000, fair::Severity::DEBUG, "Sending an STF out. stf_id={} channel={} stf_size={} unique_equipments={}",
-      lStf->header().mId, lOutputChan.GetName(), lStf->getDataSize(), lStf->getEquipmentIdentifiers().size());
+    DDLOGF_RL(2000, fair::Severity::DEBUG, "Sending an STF out. stf_id={} stf_size={} unique_equipments={}",
+      lStf->header().mId, lStf->getDataSize(), lStf->getEquipmentIdentifiers().size());
 
     // get data size sample
     I().mStfSizeSamples.Fill(lStf->getDataSize());
@@ -394,6 +367,7 @@ void StfBuilderDevice::StfOutputThread()
     );
 
     // Send a multipart
+    auto& lOutputChan = getOutputChannel();
     FairMQParts lCompletedMsg;
     auto lNoFree = [](void*, void*) { /* stack */ };
     lCompletedMsg.AddPart(lOutputChan.NewMessage(lDoneStack.data(), lDoneStack.size(), lNoFree));
diff --git a/src/StfBuilder/StfBuilderDevice.h b/src/StfBuilder/StfBuilderDevice.h
@@ -91,10 +91,6 @@ class StfBuilderDevice : public DataDistDevice,
   const std::string& getDplChannelName() const { return I().mDplChannelName; }
 
   auto& getOutputChannel() {
-    if (isStandalone()) {
-      return *I().mStandaloneChannel;
-    }
-
     if (dplEnabled()) {
       return this->GetChannel(I().mDplChannelName);
     }
@@ -218,7 +214,6 @@ class StfBuilderDevice : public DataDistDevice,
     std::unique_ptr<SubTimeFrameFileSink> mFileSink;
 
     /// File source
-    std::unique_ptr<FairMQChannel> mStandaloneChannel;
     std::unique_ptr<SubTimeFrameFileSource> mFileSource;
 
     /// Info thread
@@ -233,10 +228,10 @@ class StfBuilderDevice : public DataDistDevice,
 
   std::unique_ptr<StfBuilderInstance> mI;
   std::unique_ptr<MemoryResources> mMemI;
-
+  const StfBuilderInstance& I() const { return *mI; }
+public:
   StfBuilderInstance& I() { return *mI; }
   MemoryResources& MemI() { return *mMemI; }
-  const StfBuilderInstance& I() const { return *mI; }
 };
 
 }
diff --git a/src/StfBuilder/StfBuilderInput.cxx b/src/StfBuilder/StfBuilderInput.cxx
@@ -39,13 +39,9 @@ void StfInputInterface::start(const std::size_t pNumBuilders)
   mBuilderInputQueues.clear();
   mBuilderInputQueues.resize(mNumBuilders);
 
-  // Reference to the output or DPL channel
-  // const auto &lOutChanName = mDevice.getOutputChannelName();
-  auto& lOutputChan = mDevice.getOutputChannel();
-
   // NOTE: create the mStfBuilders first to avid resizing the vector; then threads
   for (std::size_t i = 0; i < mNumBuilders; i++) {
-    mStfBuilders.emplace_back(lOutputChan, mDevice.dplEnabled());
+    mStfBuilders.emplace_back(mDevice.MemI(), mDevice.dplEnabled());
   }
 
   for (std::size_t i = 0; i < mNumBuilders; i++) {
diff --git a/src/TfBuilder/TfBuilderDevice.cxx b/src/TfBuilder/TfBuilderDevice.cxx
@@ -46,7 +46,7 @@ TfBuilderDevice::~TfBuilderDevice()
 
 void TfBuilderDevice::Init()
 {
-  mMemI = std::make_unique<MemoryResources>();
+  mMemI = std::make_unique<MemoryResources>(this->AddTransport(fair::mq::Transport::SHM));
 }
 
 void TfBuilderDevice::Reset()
@@ -111,27 +111,6 @@ void TfBuilderDevice::InitTask()
       DDLOGF(fair::Severity::INFO, "Not sending to DPL.");
     }
 
-    // channel for FileSource: stf or dpl, or generic one in case of standalone
-    if (mStandalone) {
-      // create default FMQ shm channel
-      auto lTransportFactory = FairMQTransportFactory::CreateTransportFactory("shmem", "", GetConfig());
-      if (!lTransportFactory) {
-        DDLOG(fair::Severity::ERROR) << "Creating transport factory failed!";
-        throw "Transport factory";
-        return;
-      }
-      mStandaloneChannel = std::make_unique<FairMQChannel>(
-        "standalone-chan[0]" ,  // name
-        "pair",                 // type
-        "bind",                 // method
-        "ipc:///tmp/standalone-chan-tfb", // address
-        lTransportFactory
-      );
-
-      mStandaloneChannel->Init();
-      mStandaloneChannel->Validate();
-    }
-
     // start the info thread
     mInfoThread = create_thread_member("tfb_info", &TfBuilderDevice::InfoThread, this);
   }
@@ -157,10 +136,12 @@ bool TfBuilderDevice::start()
 
   // we reached the scheduler instance, initialize everything else
   mRunning = true;
+  auto lShmTransport = this->AddTransport(fair::mq::Transport::SHM);
 
   if (!mStandalone && dplEnabled()) {
     auto& lOutputChan = GetChannel(getDplChannelName(), 0);
-    mTfBuilder = std::make_unique<TimeFrameBuilder>(lOutputChan, mTfBufferSize, dplEnabled());
+
+    mTfBuilder = std::make_unique<TimeFrameBuilder>(MemI(), mTfBufferSize, 512 << 20 /* config */, dplEnabled());
     mTfDplAdapter = std::make_unique<StfToDplAdapter>(lOutputChan);
   }
 
@@ -178,11 +159,7 @@ bool TfBuilderDevice::start()
   }
 
   // start file source
-  if (mStandalone) {
-    mFileSource.start(*mStandaloneChannel, MemI(), false);
-  } else {
-    mFileSource.start(GetChannel(mDplChannelName), MemI(), mDplEnabled);
-  }
+  mFileSource.start(MemI(), mStandalone ? false : mDplEnabled);
 
   return true;
 }
diff --git a/src/TfBuilder/TfBuilderDevice.h b/src/TfBuilder/TfBuilderDevice.h
@@ -144,7 +144,6 @@ class TfBuilderDevice : public DataDistDevice,
   SubTimeFrameFileSink mFileSink;
 
   /// File source
-  std::unique_ptr<FairMQChannel> mStandaloneChannel;
   SubTimeFrameFileSource mFileSource;
 
   /// TF forwarding thread
diff --git a/src/common/SubTimeFrameBuilder.cxx b/src/common/SubTimeFrameBuilder.cxx
@@ -34,13 +34,12 @@ using namespace o2::header;
 /// SubTimeFrameReadoutBuilder
 ////////////////////////////////////////////////////////////////////////////////
 
-SubTimeFrameReadoutBuilder::SubTimeFrameReadoutBuilder(FairMQChannel& pChan, bool pDplEnabled)
-  : mStf(nullptr),
-    mDplEnabled(pDplEnabled)
+SubTimeFrameReadoutBuilder::SubTimeFrameReadoutBuilder(MemoryResources &pMemRes, bool pDplEnabled)
+  : mStf(nullptr), mDplEnabled(pDplEnabled),mMemRes(pMemRes)
 {
-  mHeaderMemRes = std::make_unique<RegionAllocatorResource<alignof(o2::header::DataHeader)>>(
+  mMemRes.mHeaderMemRes = std::make_unique<RegionAllocatorResource<alignof(o2::header::DataHeader)>>(
     "O2HeadersRegion",
-    pChan,
+    *mMemRes.mShmTransport,
     std::size_t(256) << 20, /* make configurable */
     mDplEnabled ?
       sizeof(DataHeader) + sizeof(o2::framework::DataProcessingHeader) :
@@ -190,14 +189,14 @@ void SubTimeFrameReadoutBuilder::addHbFrames(
         o2::framework::DataProcessingHeader{mStf->header().mId}
       );
 
-      lHdrMsg = mHeaderMemRes->NewFairMQMessage(lStack.size());
+      lHdrMsg = mMemRes.newHeaderMessage(lStack.size());
       if (lHdrMsg) {
         std::memcpy(lHdrMsg->GetData(), lStack.data(), lStack.size());
       }
     } else {
       auto lHdrMsgStack = Stack(lDataHdr);
 
-      lHdrMsg = mHeaderMemRes->NewFairMQMessage(lHdrMsgStack.size());
+      lHdrMsg = mMemRes.newHeaderMessage(lHdrMsgStack.size());
       if (lHdrMsg) {
         std::memcpy(lHdrMsg->GetData(), lHdrMsgStack.data(), lHdrMsgStack.size());
       }
@@ -229,20 +228,20 @@ std::unique_ptr<SubTimeFrame> SubTimeFrameReadoutBuilder::getStf()
 /// SubTimeFrameFileBuilder
 ////////////////////////////////////////////////////////////////////////////////
 
-SubTimeFrameFileBuilder::SubTimeFrameFileBuilder(FairMQChannel& pChan, MemoryResources &pMemRes,
+SubTimeFrameFileBuilder::SubTimeFrameFileBuilder(MemoryResources &pMemRes,
   const std::size_t pDataSegSize, const std::size_t pHdrSegSize, bool pDplEnabled)
   : mMemRes(pMemRes), mDplEnabled(pDplEnabled)
 {
   mMemRes.mHeaderMemRes = std::make_unique<RegionAllocatorResource<alignof(o2::header::DataHeader)>>(
     "O2HeadersRegion_FileSource",
-    pChan,
+    *mMemRes.mShmTransport,
     pHdrSegSize,
     0
   );
 
   mMemRes.mDataMemRes = std::make_unique<RegionAllocatorResource<>>(
     "O2DataRegion_FileSource",
-    pChan,
+    *mMemRes.mShmTransport,
     pDataSegSize,
     0 // TODO: GPU flags
   );
@@ -307,20 +306,20 @@ void SubTimeFrameFileBuilder::adaptHeaders(SubTimeFrame *pStf)
 /// TimeFrameBuilder
 ////////////////////////////////////////////////////////////////////////////////
 
-TimeFrameBuilder::TimeFrameBuilder(FairMQChannel& pChan, const std::size_t pDataSegSize, bool pDplEnabled)
-  : mDplEnabled(pDplEnabled),
-    mOutputChan(pChan)
+TimeFrameBuilder::TimeFrameBuilder(MemoryResources &pMemRes,
+  const std::size_t pDataSegSize, const std::size_t pHdrSegSize, bool pDplEnabled)
+  : mDplEnabled(pDplEnabled), mMemRes(pMemRes)
 {
-  mHeaderMemRes = std::make_unique<RegionAllocatorResource<alignof(o2::header::DataHeader)>>(
+  mMemRes.mHeaderMemRes = std::make_unique<RegionAllocatorResource<alignof(o2::header::DataHeader)>>(
     "O2HeadersRegion",
-    pChan,
-    std::size_t(256) << 20, /* make configurable */
+    *mMemRes.mShmTransport,
+    pHdrSegSize,
     0 /* dont need registration flags for headers */
   );
 
-  mDataMemRes = std::make_unique<RegionAllocatorResource<>>(
+  mMemRes.mDataMemRes = std::make_unique<RegionAllocatorResource<>>(
     "O2DataRegion_FileSource",
-    pChan,
+    *mMemRes.mShmTransport,
     pDataSegSize,
     0 // TODO: GPU flags
   );
@@ -332,7 +331,7 @@ void TimeFrameBuilder::adaptHeaders(SubTimeFrame *pStf)
     return;
   }
 
-  const auto lOutChannelType = mOutputChan.GetTransportType();
+  const auto lOutChannelType = fair::mq::Transport::SHM;
 
   // adapt headers for DPL
   for (auto& lDataIdentMapIter : pStf->mData) {
@@ -377,11 +376,7 @@ void TimeFrameBuilder::adaptHeaders(SubTimeFrame *pStf)
               o2::framework::DataProcessingHeader{pStf->header().mId}
             );
 
-            if (lOutChannelType == fair::mq::Transport::SHM) {
-              lStfDataIter.mHeader = newHeaderMessage(lStack.size());
-            } else {
-              lStfDataIter.mHeader = mOutputChan.NewMessage(lStack.size());
-            }
+            lStfDataIter.mHeader = newHeaderMessage(lStack.size());
 
             if (lStfDataIter.mHeader) {
               assert(lStfDataIter.mHeader->GetSize() >= sizeof (DataHeader));
diff --git a/src/common/SubTimeFrameFileSource.cxx b/src/common/SubTimeFrameFileSource.cxx
@@ -40,14 +40,12 @@ using namespace std::chrono_literals;
 /// SubTimeFrameFileSource
 ////////////////////////////////////////////////////////////////////////////////
 
-void SubTimeFrameFileSource::start(FairMQChannel& pDstChan, MemoryResources &pMemRes, const bool pDplEnabled)
+void SubTimeFrameFileSource::start(MemoryResources &pMemRes, const bool pDplEnabled)
 {
   if (enabled()) {
-    mDstChan = &pDstChan;
     mDplEnabled = pDplEnabled;
 
     mFileBuilder = std::make_unique<SubTimeFrameFileBuilder>(
-      pDstChan,
       pMemRes,
       mRegionSizeMB << 20,
       mHdrRegionSizeMB << 20,
diff --git a/src/common/include/MemoryUtils.h b/src/common/include/MemoryUtils.h
diff --git a/src/common/include/SubTimeFrameBuilder.h b/src/common/include/SubTimeFrameBuilder.h
diff --git a/src/common/include/SubTimeFrameFileSource.h b/src/common/include/SubTimeFrameFileSource.h
