dpl: insepect the o2 protocol validity on the output channel

Author: ironMann

script/README.md

@@ -53,3 +53,5 @@ Make sure that only a single instance of the proxy is started.
   - `DATADIST_FEE_MASK=0xffff`  Apply the mask if StfBuilder is configured to use the FeeID field as a O2::Subspecification (O2::Subspec = (RDH::FeeID & DATADIST_FEE_MASK))
 
   - `DATADIST_FILE_READ_COUNT=N`     Terminate after injecting set number of TF files. Data set will be repeated if necessary. Number of TimeFrames will be `DATADIST_FILE_READ_COUNT x Number of TFs per file`.
+
+  - `DATADIST_DEBUG_DPL_CHAN` When defined, data sent to DPL will be checked for consistency with the O2 data model. Note: will be slow with larger TimeFrames.

src/common/SubTimeFrameBuilder.cxx

@@ -199,9 +199,7 @@ void SubTimeFrameReadoutBuilder::addHbFrames(
 
       lHdrMsg = mMemRes.newHeaderMessage(reinterpret_cast<char*>(lStack.data()), lStack.size());
     } else {
-      auto lHdrMsgStack = Stack(lDataHdr);
-
-      lHdrMsg = mMemRes.newHeaderMessage(reinterpret_cast<char*>(lHdrMsgStack.data()), lHdrMsgStack.size());
+      lHdrMsg = mMemRes.newHeaderMessage(reinterpret_cast<char*>(&lDataHdr), sizeof(lDataHdr));
     }
 
     if (!lHdrMsg) {
@@ -213,7 +211,6 @@ void SubTimeFrameReadoutBuilder::addHbFrames(
       SubTimeFrame::StfData{ std::move(lHdrMsg), std::move(pHbFramesBegin[i]) }
     );
   }
-
 }
 
 

src/common/SubTimeFrameDPL.cxx

@@ -89,30 +89,9 @@ void StfToDplAdapter::visit(SubTimeFrame& pStf)
   }
 }
 
-void StfToDplAdapter::sendToDpl(std::unique_ptr<SubTimeFrame>&& pStf)
+void StfToDplAdapter::inspect() const
 {
-  if (!mRunning) {
-    return;
-  }
-
-  mMessages.clear();
-  pStf->accept(*this);
-
-#if 0
-  DDDLOG("Content of the Stf:");
-  uint64_t lMsgIdx = 0;
-  for (auto lM = mMessages.cbegin(); lM != mMessages.cend(); ) {
-    DDDLOG("  o2: message #{}", lMsgIdx++);
-    o2::header::hexDump("o2 header", (*lM)->GetData(), (*lM)->GetSize());
-    lM++;
-    o2::header::hexDump("o2 payload", (*lM)->GetData(), std::clamp((*lM)->GetSize(), std::size_t(0), std::size_t(256)) );
-    lM++;
-  }
-#endif
-
-#if !defined(NDEBUG)
   // check validity of the TF message sequence
-
   const auto cMsgSize = mMessages.size() / 2;
 
   DataHeader::TForbitType lFirstTForbit = ~  DataHeader::TForbitType{0};
@@ -207,9 +186,33 @@ void StfToDplAdapter::sendToDpl(std::unique_ptr<SubTimeFrame>&& pStf)
     // TODO: record the origin
     lIdx += lDh->splitPayloadParts;
   }
+}
 
+void StfToDplAdapter::sendToDpl(std::unique_ptr<SubTimeFrame>&& pStf)
+{
+  if (!mRunning) {
+    return;
+  }
+
+  mMessages.clear();
+  pStf->accept(*this);
+
+#if 0
+  DDDLOG("Content of the Stf:");
+  uint64_t lMsgIdx = 0;
+  for (auto lM = mMessages.cbegin(); lM != mMessages.cend(); ) {
+    DDDLOG("  o2: message #{}", lMsgIdx++);
+    o2::header::hexDump("o2 header", (*lM)->GetData(), (*lM)->GetSize());
+    lM++;
+    o2::header::hexDump("o2 payload", (*lM)->GetData(), std::clamp((*lM)->GetSize(), std::size_t(0), std::size_t(256)) );
+    lM++;
+  }
 #endif
 
+  if (mInspectChannel) {
+    inspect();
+  }
+
   mChan.Send(mMessages);
 
   // make sure headers and chunk pointers don't linger

src/common/SubTimeFrameDPL.h

@@ -34,6 +34,9 @@ class StfToDplAdapter : public ISubTimeFrameVisitor
     : mChan(pDplBridgeChan)
   {
     mMessages.reserve(1024);
+    if (getenv("DATADIST_DEBUG_DPL_CHAN")) {
+      mInspectChannel = true;
+    }
   }
 
   virtual ~StfToDplAdapter() = default;
@@ -45,8 +48,12 @@ class StfToDplAdapter : public ISubTimeFrameVisitor
  protected:
   void visit(SubTimeFrame& pStf) override;
 
+  void inspect() const;
+
  private:
   std::atomic_bool mRunning = true;
+  bool mInspectChannel = false;
+
   std::vector<FairMQMessagePtr> mMessages;
   FairMQChannel& mChan;
 };