add DEADBEEF check in TOF decoder spec

noferini · shahor02 · commit ef7c674e9984 · 2021-06-09T17:33:18.000+02:00
diff --git a/Detectors/TOF/workflow/include/TOFWorkflowUtils/CompressedDecodingTask.h b/Detectors/TOF/workflow/include/TOFWorkflowUtils/CompressedDecodingTask.h
@@ -36,15 +36,17 @@ using namespace compressed;
 class CompressedDecodingTask : public DecoderBase, public Task
 {
  public:
-  CompressedDecodingTask(bool conet = false)
+  CompressedDecodingTask(bool conet, o2::header::DataDescription dataDesc)
   {
     mConetMode = conet;
+    mDataDesc = dataDesc;
     setDecoderCONET(conet);
   }
 
   ~CompressedDecodingTask() override = default;
   void init(InitContext& ic) final;
   void run(ProcessingContext& pc) final;
+  void decodeTF(ProcessingContext& pc);
   void endOfStream(EndOfStreamContext& ec) final;
   void postData(ProcessingContext& pc);
 
@@ -60,6 +62,7 @@ class CompressedDecodingTask : public DecoderBase, public Task
 
   o2::tof::compressed::Decoder mDecoder;
   std::vector<std::vector<o2::tof::Digit>> mDigits;
+  o2::header::DataDescription mDataDesc;
   int mNTF = 0;
   int mNCrateOpenTF = 0;
   int mNCrateCloseTF = 0;
diff --git a/Detectors/TOF/workflow/src/CompressedDecodingTask.cxx b/Detectors/TOF/workflow/src/CompressedDecodingTask.cxx
@@ -25,6 +25,7 @@
 #include "Framework/WorkflowSpec.h"
 #include "Framework/Logger.h"
 #include "DetectorsRaw/RDHUtils.h"
+#include "Framework/InputRecordWalker.h"
 
 using namespace o2::framework;
 
@@ -127,23 +128,7 @@ void CompressedDecodingTask::run(ProcessingContext& pc)
     mDecoder.setFirstIR({0, mInitOrbit});
   }
 
-  /** loop over inputs routes **/
-  for (auto iit = pc.inputs().begin(), iend = pc.inputs().end(); iit != iend; ++iit) {
-    if (!iit.isValid()) {
-      continue;
-    }
-
-    /** loop over input parts **/
-    for (auto const& ref : iit) {
-      const auto* headerIn = DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
-      auto payloadIn = ref.payload;
-      auto payloadInSize = headerIn->payloadSize;
-
-      DecoderBase::setDecoderBuffer(payloadIn);
-      DecoderBase::setDecoderBufferSize(payloadInSize);
-      DecoderBase::run();
-    }
-  }
+  decodeTF(pc);
 
   if (!mConetMode) {
     mHasToBePosted = true;
@@ -163,6 +148,43 @@ void CompressedDecodingTask::endOfStream(EndOfStreamContext& ec)
        mTimer.CpuTime(), mTimer.RealTime(), mTimer.Counter() - 1);
 }
 
+void CompressedDecodingTask::decodeTF(ProcessingContext& pc)
+{
+  auto& inputs = pc.inputs();
+
+  // if we see requested data type input with 0xDEADBEEF subspec and 0 payload this means that the "delayed message"
+  // mechanism created it in absence of real data from upstream. Processor should send empty output to not block the workflow
+  {
+    std::vector<InputSpec> dummy{InputSpec{"dummy", ConcreteDataMatcher{"TOF", mDataDesc, 0xDEADBEEF}}};
+    for (const auto& ref : InputRecordWalker(inputs, dummy)) {
+      const auto dh = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
+      if (dh->payloadSize == 0) {
+        LOGP(WARNING, "Found input [{}/{}/{:#x}] TF#{} 1st_orbit:{} Payload {} : assuming no payload for all links in this TF",
+             dh->dataOrigin.str, dh->dataDescription.str, dh->subSpecification, dh->tfCounter, dh->firstTForbit, dh->payloadSize);
+        return;
+      }
+    }
+  }
+
+  /** loop over inputs routes **/
+  for (auto iit = pc.inputs().begin(), iend = pc.inputs().end(); iit != iend; ++iit) {
+    if (!iit.isValid()) {
+      continue;
+    }
+
+    /** loop over input parts **/
+    for (auto const& ref : iit) {
+      const auto* headerIn = DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
+      auto payloadIn = ref.payload;
+      auto payloadInSize = headerIn->payloadSize;
+
+      DecoderBase::setDecoderBuffer(payloadIn);
+      DecoderBase::setDecoderBufferSize(payloadInSize);
+      DecoderBase::run();
+    }
+  }
+}
+
 void CompressedDecodingTask::headerHandler(const CrateHeader_t* crateHeader, const CrateOrbit_t* crateOrbit)
 {
   if (mConetMode) {
@@ -362,6 +384,12 @@ void CompressedDecodingTask::frameHandler(const CrateHeader_t* crateHeader, cons
 
 DataProcessorSpec getCompressedDecodingSpec(const std::string& inputDesc, bool conet)
 {
+  std::vector<InputSpec> inputs;
+  //  inputs.emplace_back(std::string("x:TOF/" + inputDesc).c_str(), 0, Lifetime::Optional);
+  o2::header::DataDescription dataDesc;
+  dataDesc.runtimeInit(inputDesc.c_str());
+  inputs.emplace_back("x", ConcreteDataTypeMatcher{o2::header::gDataOriginTOF, dataDesc}, Lifetime::Optional);
+
   std::vector<OutputSpec> outputs;
   outputs.emplace_back(o2::header::gDataOriginTOF, "DIGITHEADER", 0, Lifetime::Timeframe);
   outputs.emplace_back(o2::header::gDataOriginTOF, "DIGITS", 0, Lifetime::Timeframe);
@@ -371,9 +399,10 @@ DataProcessorSpec getCompressedDecodingSpec(const std::string& inputDesc, bool c
 
   return DataProcessorSpec{
     "tof-compressed-decoder",
-    select(std::string("x:TOF/" + inputDesc).c_str()),
+    inputs,
+    //    select(std::string("x:TOF/" + inputDesc).c_str()),
     outputs,
-    AlgorithmSpec{adaptFromTask<CompressedDecodingTask>(conet)},
+    AlgorithmSpec{adaptFromTask<CompressedDecodingTask>(conet, dataDesc)},
     Options{
       {"row-filter", VariantType::Bool, false, {"Filter empty row"}},
       {"mask-noise", VariantType::Bool, false, {"Flag to mask noisy digits"}},
