TPC: replace two-copy forwarding by shallow-copy forwarding in IDC/CMV distribute devices

ehellbar · ehellbar · commit e30a9aeace67 · 2026-07-06T17:14:24.000+02:00
diff --git a/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeCMVSpec.h b/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeCMVSpec.h
@@ -182,12 +182,13 @@ class TPCDistributeCMVSpec : public o2::framework::Task
 
     forwardOrbitInfo(pc, currentBuffer, relTF, currentOutLane);
 
-    for (auto& ref : o2::framework::InputRecordWalker(pc.inputs(), mFilter)) {
-      auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
+    auto inputs = o2::framework::InputRecordWalker(pc.inputs(), mFilter);
+    for (auto it = inputs.begin(); it != inputs.end(); ++it) {
+      auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(*it);
       const unsigned int cru = tpcCRUHeader->subSpecification >> 7;
 
       // check if cru is specified in input cru list
-      if (!(std::binary_search(mCRUs.begin(), mCRUs.end(), cru))) {
+      if (!std::binary_search(mCRUs.begin(), mCRUs.end(), cru)) {
         LOGP(debug, "Received data from CRU: {} which was not specified as input. Skipping", cru);
         continue;
       }
@@ -200,7 +201,7 @@ class TPCDistributeCMVSpec : public o2::framework::Task
       // to keep track of processed CRUs
       mProcessedCRUs[currentBuffer][relTF][cru] = true;
 
-      sendOutput(pc, currentOutLane, cru, pc.inputs().get<o2::pmr::vector<uint16_t>>(ref));
+      forwardData(pc, o2::framework::Output{o2::header::gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, cru, it, [&] { sendEmptyCMVOutput(pc, currentOutLane, cru); });
     }
 
     LOGP(detail, "Number of received CRUs for current TF: {} Needed a total number of processed CRUs of: {} Current TF: {}", mProcessedCRU[currentBuffer][relTF], mCRUs.size(), tf);
@@ -247,7 +248,7 @@ class TPCDistributeCMVSpec : public o2::framework::Task
   const unsigned int mTimeFrames{};                                                    ///< number of TFs per aggregation interval
   const int mNTFsBuffer{1};                                                            ///< number of TFs for which the CMVs will be buffered (must match TPCFLPCMVSpec)
   const unsigned int mOutLanes{};                                                      ///< number of parallel aggregate pipelines this distributor feeds
-  std::array<unsigned int, 2> mProcessedTFs{{0, 0}};                                   ///< number of processed timeframes per buffer; triggers sendOutput when it reaches mTimeFrames
+  std::array<unsigned int, 2> mProcessedTFs{{0, 0}};                                   ///< number of processed timeframes per buffer; triggers finishInterval when it reaches mTimeFrames
   std::array<std::vector<unsigned int>, 2> mProcessedCRU{};                            ///< counter of received CRUs per (buffer, relTF); used to detect when a relTF is complete
   std::array<std::vector<std::unordered_map<unsigned int, bool>>, 2> mProcessedCRUs{}; ///< per-CRU received flag ([buffer][relTF][CRU]); prevents double-counting when a CRU re-sends
   std::array<long, 2> mTFStart{};                                                      ///< absolute TF counter of the first TF in each buffer interval
@@ -274,14 +275,26 @@ class TPCDistributeCMVSpec : public o2::framework::Task
   /// Returns the total number of real TFs per buffer interval (= mNTFsBuffer * mTimeFrames)
   unsigned int getNRealTFs() const { return mNTFsBuffer * mTimeFrames; }
 
-  void sendOutput(o2::framework::ProcessingContext& pc, const unsigned int currentOutLane, const unsigned int cru, o2::pmr::vector<uint16_t> cmvs)
+  void sendEmptyCMVOutput(o2::framework::ProcessingContext& pc, const unsigned int currentOutLane, const unsigned int cru)
+  {
+    pc.outputs().adoptContainer(o2::framework::Output{o2::header::gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, o2::pmr::vector<uint16_t>());
+  }
+
+  void sendEmptyOrbitInfo(o2::framework::ProcessingContext& pc, const unsigned int outLane)
   {
-    pc.outputs().adoptContainer(o2::framework::Output{o2::header::gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, std::move(cmvs));
+    pc.outputs().snapshot(o2::framework::Output{o2::header::gDataOriginTPC, mOrbitDescrOut[outLane], header::DataHeader::SubSpecificationType{outLane}}, static_cast<uint64_t>(0));
   }
 
-  void sendOrbitInfo(o2::framework::ProcessingContext& pc, const unsigned int outLane, const uint64_t orbitInfo)
+  template <typename EmptyFn>
+  void forwardData(o2::framework::ProcessingContext& pc, o2::framework::Output outSpec, const unsigned int cru, auto& it, EmptyFn&& sendEmptyData)
   {
-    pc.outputs().snapshot(o2::framework::Output{o2::header::gDataOriginTPC, mOrbitDescrOut[outLane], header::DataHeader::SubSpecificationType{outLane}}, orbitInfo);
+    if (auto* payloadMsg = it.getPayload()) {
+      pc.outputs().forwardPayload(outSpec, *payloadMsg);
+    } else [[unlikely]] {
+      // this should never happen
+      LOGP(warning, "No payload for CRU {}; sending empty {} payload", cru, outSpec.description.as<std::string>());
+      sendEmptyData();
+    }
   }
 
   void forwardOrbitInfo(o2::framework::ProcessingContext& pc, const bool currentBuffer, const unsigned int relTF, const unsigned int currentOutLane)
@@ -290,14 +303,15 @@ class TPCDistributeCMVSpec : public o2::framework::Task
       return;
     }
 
-    for (auto& ref : o2::framework::InputRecordWalker(pc.inputs(), mOrbitFilter)) {
-      auto const* hdr = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
-      const unsigned int cru = hdr->subSpecification >> 7;
+    auto inputs = o2::framework::InputRecordWalker(pc.inputs(), mOrbitFilter);
+    for (auto it = inputs.begin(); it != inputs.end(); ++it) {
+      auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(*it);
+      const unsigned int cru = tpcCRUHeader->subSpecification >> 7;
       if (!std::binary_search(mCRUs.begin(), mCRUs.end(), cru)) {
         continue;
       }
 
-      sendOrbitInfo(pc, currentOutLane, pc.inputs().get<uint64_t>(ref));
+      forwardData(pc, o2::framework::Output{o2::header::gDataOriginTPC, mOrbitDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{currentOutLane}}, cru, it, [&] { sendEmptyOrbitInfo(pc, currentOutLane); });
       mOrbitInfoForwarded[currentBuffer][relTF] = true;
       break;
     }
@@ -319,24 +333,30 @@ class TPCDistributeCMVSpec : public o2::framework::Task
     }
 
     if (!mOrbitInfoForwarded[mBuffer].empty()) {
-      for (auto& ref : o2::framework::InputRecordWalker(pc.inputs(), mOrbitFilter)) {
-        auto const* hdr = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
-        const unsigned int cru = hdr->subSpecification >> 7;
+      auto inputs = o2::framework::InputRecordWalker(pc.inputs(), mOrbitFilter);
+      for (auto it = inputs.begin(); it != inputs.end(); ++it) {
+        auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(*it);
+        const unsigned int cru = tpcCRUHeader->subSpecification >> 7;
         if (!std::binary_search(mCRUs.begin(), mCRUs.end(), cru)) {
           continue;
         }
-        sendOrbitInfo(pc, currentOutLane, pc.inputs().get<uint64_t>(ref));
+
+        forwardData(pc, o2::framework::Output{o2::header::gDataOriginTPC, mOrbitDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{currentOutLane}}, cru, it, [&] { sendEmptyOrbitInfo(pc, currentOutLane); });
         break;
       }
     }
 
-    for (auto& ref : o2::framework::InputRecordWalker(pc.inputs(), mFilter)) {
-      auto const* hdr = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
-      const unsigned int cru = hdr->subSpecification >> 7;
+    auto inputs = o2::framework::InputRecordWalker(pc.inputs(), mFilter);
+    for (auto it = inputs.begin(); it != inputs.end(); ++it) {
+      auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(*it);
+      const unsigned int cru = tpcCRUHeader->subSpecification >> 7;
+
+      // check if cru is specified in input cru list
       if (!std::binary_search(mCRUs.begin(), mCRUs.end(), cru)) {
         continue;
       }
-      sendOutput(pc, currentOutLane, cru, pc.inputs().get<o2::pmr::vector<uint16_t>>(ref));
+
+      forwardData(pc, o2::framework::Output{o2::header::gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, cru, it, [&] { sendEmptyCMVOutput(pc, currentOutLane, cru); });
     }
   }
 
@@ -394,13 +414,13 @@ class TPCDistributeCMVSpec : public o2::framework::Task
         for (auto& it : mProcessedCRUs[currentBuffer][iTF]) {
           if (!it.second) {
             it.second = true;
-            sendOutput(pc, outLane, it.first, o2::pmr::vector<uint16_t>());
+            sendEmptyCMVOutput(pc, outLane, it.first);
           }
         }
 
         // send zero orbit placeholder for missing TF so the aggregate lane can still reconstruct timing
         if (!mOrbitInfoForwarded[currentBuffer][iTF]) {
-          sendOrbitInfo(pc, outLane, 0);
+          sendEmptyOrbitInfo(pc, outLane);
           mOrbitInfoForwarded[currentBuffer][iTF] = true;
         }
       }
diff --git a/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeIDCSpec.h b/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeIDCSpec.h
@@ -169,12 +169,13 @@ class TPCDistributeIDCSpec : public o2::framework::Task
       pc.outputs().snapshot(Output{gDataOriginTPC, getDataDescriptionIDCOrbitReset(), header::DataHeader::SubSpecificationType{currentOutLane}}, dataformats::Pair<long, int>{o2::base::GRPGeomHelper::instance().getOrbitResetTimeMS(), o2::base::GRPGeomHelper::instance().getNHBFPerTF()});
     }
 
-    for (auto& ref : InputRecordWalker(pc.inputs(), mFilter)) {
-      auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
+    auto inputs = o2::framework::InputRecordWalker(pc.inputs(), mFilter);
+    for (auto it = inputs.begin(); it != inputs.end(); ++it) {
+      auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(*it);
       const unsigned int cru = tpcCRUHeader->subSpecification >> 7;
 
       // check if cru is specified in input cru list
-      if (!(std::binary_search(mCRUs.begin(), mCRUs.end(), cru))) {
+      if (!std::binary_search(mCRUs.begin(), mCRUs.end(), cru)) {
         LOGP(debug, "Received data from CRU: {} which was not specified as input. Skipping", cru);
         continue;
       }
@@ -189,8 +190,14 @@ class TPCDistributeIDCSpec : public o2::framework::Task
         mProcessedCRUs[currentBuffer][relTF][cru] = true;
       }
 
-      // sending IDCs
-      sendOutput(pc, currentOutLane, cru, pc.inputs().get<pmr::vector<float>>(ref));
+      // forward payload by shallow copy
+      if (auto* payloadMsg = it.getPayload()) {
+        pc.outputs().forwardPayload(Output{gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, *payloadMsg);
+      } else [[unlikely]] {
+        // this should never happen
+        LOGP(warning, "No IDCGROUP payload for CRU {} (TF {}, relTF {}); sending empty IDCAGG{} payload", cru, tf, relTF, currentOutLane);
+        sendEmptyIDCOutput(pc, currentOutLane, cru);
+      }
     }
 
     LOGP(info, "number of received CRUs for current TF: {}    Needed a total number of processed CRUs of: {}   Current TF: {}", mProcessedCRU[currentBuffer][relTF], mCRUs.size(), tf);
@@ -247,9 +254,9 @@ class TPCDistributeIDCSpec : public o2::framework::Task
   std::vector<InputSpec> mFilter{};                                                    ///< filter for looping over input data
   std::vector<header::DataDescription> mDataDescrOut{};
 
-  void sendOutput(o2::framework::ProcessingContext& pc, const unsigned int currentOutLane, const unsigned int cru, o2::pmr::vector<float> idcs)
+  void sendEmptyIDCOutput(o2::framework::ProcessingContext& pc, const unsigned int currentOutLane, const unsigned int cru)
   {
-    pc.outputs().adoptContainer(Output{gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, std::move(idcs));
+    pc.outputs().adoptContainer(Output{gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, pmr::vector<float>());
   }
 
   /// returns the output lane to which the data will be send
@@ -314,7 +321,7 @@ class TPCDistributeIDCSpec : public o2::framework::Task
         for (auto& it : mProcessedCRUs[currentBuffer][iTF]) {
           if (!it.second) {
             it.second = true;
-            sendOutput(pc, outLane, it.first, pmr::vector<float>());
+            sendEmptyIDCOutput(pc, outLane, it.first);
           }
         }
       }

Original file line number	Diff line number	Diff line change
`@@ -169,12 +169,13 @@ class TPCDistributeIDCSpec : public o2::framework::Task`
`169`	`169`	`pc.outputs().snapshot(Output{gDataOriginTPC, getDataDescriptionIDCOrbitReset(), header::DataHeader::SubSpecificationType{currentOutLane}}, dataformats::Pair<long, int>{o2::base::GRPGeomHelper::instance().getOrbitResetTimeMS(), o2::base::GRPGeomHelper::instance().getNHBFPerTF()});`
`170`	`170`	`}`
`171`	`171`
`172`		`- for (auto& ref : InputRecordWalker(pc.inputs(), mFilter)) {`
`173`		`- auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(ref);`
	`172`	`+ auto inputs = o2::framework::InputRecordWalker(pc.inputs(), mFilter);`
	`173`	`+ for (auto it = inputs.begin(); it != inputs.end(); ++it) {`
	`174`	`+ auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader>(it);`
`174`	`175`	`const unsigned int cru = tpcCRUHeader->subSpecification >> 7;`
`175`	`176`
`176`	`177`	`// check if cru is specified in input cru list`
`177`		`- if (!(std::binary_search(mCRUs.begin(), mCRUs.end(), cru))) {`
	`178`	`+ if (!std::binary_search(mCRUs.begin(), mCRUs.end(), cru)) {`
`178`	`179`	`LOGP(debug, "Received data from CRU: {} which was not specified as input. Skipping", cru);`
`179`	`180`	`continue;`
`180`	`181`	`}`
`@@ -189,8 +190,14 @@ class TPCDistributeIDCSpec : public o2::framework::Task`
`189`	`190`	`mProcessedCRUs[currentBuffer][relTF][cru] = true;`
`190`	`191`	`}`
`191`	`192`
`192`		`- // sending IDCs`
`193`		`- sendOutput(pc, currentOutLane, cru, pc.inputs().get<pmr::vector<float>>(ref));`
	`193`	`+ // forward payload by shallow copy`
	`194`	`+ if (auto* payloadMsg = it.getPayload()) {`
	`195`	`+ pc.outputs().forwardPayload(Output{gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, *payloadMsg);`
	`196`	`+ } else [[unlikely]] {`
	`197`	`+ // this should never happen`
	`198`	`+ LOGP(warning, "No IDCGROUP payload for CRU {} (TF {}, relTF {}); sending empty IDCAGG{} payload", cru, tf, relTF, currentOutLane);`
	`199`	`+ sendEmptyIDCOutput(pc, currentOutLane, cru);`
	`200`	`+ }`
`194`	`201`	`}`
`195`	`202`
`196`	`203`	`LOGP(info, "number of received CRUs for current TF: {} Needed a total number of processed CRUs of: {} Current TF: {}", mProcessedCRU[currentBuffer][relTF], mCRUs.size(), tf);`
`@@ -247,9 +254,9 @@ class TPCDistributeIDCSpec : public o2::framework::Task`
`247`	`254`	`std::vector<InputSpec> mFilter{}; ///< filter for looping over input data`
`248`	`255`	`std::vector<header::DataDescription> mDataDescrOut{};`
`249`	`256`
`250`		`- void sendOutput(o2::framework::ProcessingContext& pc, const unsigned int currentOutLane, const unsigned int cru, o2::pmr::vector<float> idcs)`
	`257`	`+ void sendEmptyIDCOutput(o2::framework::ProcessingContext& pc, const unsigned int currentOutLane, const unsigned int cru)`
`251`	`258`	`{`
`252`		`- pc.outputs().adoptContainer(Output{gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, std::move(idcs));`
	`259`	`+ pc.outputs().adoptContainer(Output{gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, pmr::vector<float>());`
`253`	`260`	`}`
`254`	`261`
`255`	`262`	`/// returns the output lane to which the data will be send`
`@@ -314,7 +321,7 @@ class TPCDistributeIDCSpec : public o2::framework::Task`
`314`	`321`	`for (auto& it : mProcessedCRUs[currentBuffer][iTF]) {`
`315`	`322`	`if (!it.second) {`
`316`	`323`	`it.second = true;`
`317`		`- sendOutput(pc, outLane, it.first, pmr::vector<float>());`
	`324`	`+ sendEmptyIDCOutput(pc, outLane, it.first);`
`318`	`325`	`}`
`319`	`326`	`}`
`320`	`327`	`}`