Multiple fixes in HMPID Raw->Digit

Add expired messages mechanism, fix input specs, suppress unneeded inputs

Author: shahor02

Detectors/HMPID/workflow/include/HMPIDWorkflow/DataDecoderSpec.h

@@ -45,7 +45,7 @@ class DataDecoderTask : public framework::Task
   ExecutionTimer mExTimer;
 };
 
-o2::framework::DataProcessorSpec getDecodingSpec(std::string inputSpec = "TF:HMP/RAWDATA");
+o2::framework::DataProcessorSpec getDecodingSpec(bool askSTFDist);
 } // end namespace hmpid
 } // end namespace o2
 

Detectors/HMPID/workflow/src/DataDecoderSpec.cxx

@@ -37,6 +37,7 @@
 #include "Framework/Task.h"
 #include "Framework/WorkflowSpec.h"
 #include "Framework/Logger.h"
+#include "Framework/InputRecordWalker.h"
 
 #include "Headers/RAWDataHeader.h"
 #include "DetectorsRaw/RDHUtils.h"
@@ -159,6 +160,21 @@ void DataDecoderTask::decodeTF(framework::ProcessingContext& pc)
 
   // get the input buffer
   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{"HMP", "RAWDATA", 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;
+      }
+    }
+  }
+
   DPLRawParser parser(inputs, o2::framework::select("TF:HMP/RAWDATA"));
   mDeco->mDigits.clear();
   for (auto it = parser.begin(), end = parser.end(); it != end; ++it) {
@@ -253,22 +269,21 @@ void DataDecoderTask::decodeRawFile(framework::ProcessingContext& pc)
 }
 
 //_________________________________________________________________________________________________
-o2::framework::DataProcessorSpec getDecodingSpec(std::string inputSpec)
+o2::framework::DataProcessorSpec getDecodingSpec(bool askDISTSTF)
 {
   std::vector<o2::framework::InputSpec> inputs;
-  inputs.emplace_back("TF", o2::framework::ConcreteDataTypeMatcher{"HMP", "RAWDATA"}, o2::framework::Lifetime::Timeframe);
-  inputs.emplace_back("file", o2::framework::ConcreteDataTypeMatcher{"ROUT", "RAWDATA"}, o2::framework::Lifetime::Timeframe);
-  //  inputs.emplace_back("readout", o2::header::gDataOriginHMP, "RAWDATA", 0, Lifetime::Timeframe);
-  //  inputs.emplace_back("readout", o2::header::gDataOriginHMP, "RAWDATA", 0, Lifetime::Timeframe);
-  //  inputs.emplace_back("rawfile", o2::header::gDataOriginHMP, "RAWDATA", 0, Lifetime::Timeframe);
+  inputs.emplace_back("TF", o2::framework::ConcreteDataTypeMatcher{"HMP", "RAWDATA"}, o2::framework::Lifetime::Optional);
+  if (askDISTSTF) {
+    inputs.emplace_back("stdDist", "FLP", "DISTSUBTIMEFRAME", 0, Lifetime::Timeframe);
+  }
 
   std::vector<o2::framework::OutputSpec> outputs;
   outputs.emplace_back("HMP", "DIGITS", 0, o2::framework::Lifetime::Timeframe);
   outputs.emplace_back("HMP", "INTRECORDS", 0, o2::framework::Lifetime::Timeframe);
 
   return DataProcessorSpec{
     "HMP-RawStreamDecoder",
-    o2::framework::select(inputSpec.c_str()),
+    inputs,
     outputs,
     AlgorithmSpec{adaptFromTask<DataDecoderTask>()},
     Options{{"result-file", VariantType::String, "/tmp/hmpRawDecodeResults", {"Base name of the decoding results files."}},

Detectors/HMPID/workflow/src/raw-to-digits-stream-workflow.cxx

@@ -49,6 +49,7 @@ void customize(std::vector<o2::framework::ConfigParamSpec>& workflowOptions)
 {
   std::string keyvaluehelp("Semicolon separated key=value strings ...");
   workflowOptions.push_back(o2::framework::ConfigParamSpec{"configKeyValues", o2::framework::VariantType::String, "", {keyvaluehelp}});
+  workflowOptions.push_back(o2::framework::ConfigParamSpec{"ignore-dist-stf", o2::framework::VariantType::Bool, false, {"do not subscribe to FLP/DISTSUBTIMEFRAME/0 message (no lost TF recovery)"}});
 }
 
 #include "Framework/runDataProcessing.h"
@@ -61,7 +62,8 @@ WorkflowSpec defineDataProcessing(const ConfigContext& cx)
 {
   WorkflowSpec specs;
   o2::conf::ConfigurableParam::updateFromString(cx.options().get<std::string>("configKeyValues"));
-  DataProcessorSpec producer = o2::hmpid::getDecodingSpec();
+  auto askSTFDist = !cx.options().get<bool>("ignore-dist-stf");
+  DataProcessorSpec producer = o2::hmpid::getDecodingSpec(askSTFDist);
   specs.push_back(producer);
   return specs;
 }