trd reading empty frames

Author: bazinski

Detectors/TRD/reconstruction/src/DataReader.cxx

@@ -35,6 +35,7 @@ void customize(std::vector<ConfigParamSpec>& workflowOptions)
     {"trd-datareader-headerverbose", VariantType::Bool, false, {"Enable verbose header info"}},
     {"trd-datareader-dataverbose", VariantType::Bool, false, {"Enable verbose data info"}},
     {"trd-datareader-compresseddata", VariantType::Bool, false, {"The incoming data is compressed or not"}},
+    {"ignore-dist-stf", VariantType::Bool, false, {"do not subscribe to FLP/DISTSUBTIMEFRAME/0 message (no lost TF recovery)"}},
     {"trd-datareader-enablebyteswapdata", VariantType::Bool, false, {"byteswap the incoming data, raw data needs it and simulation does not."}}};
 
   o2::raw::HBFUtilsInitializer::addConfigOption(options);
@@ -62,7 +63,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const& cfgc)
   auto compresseddata = cfgc.options().get<bool>("trd-datareader-compresseddata");
   auto headerverbose = cfgc.options().get<bool>("trd-datareader-headerverbose");
   auto dataverbose = cfgc.options().get<bool>("trd-datareader-dataverbose");
-
+  auto askSTFDist = !cfgc.options().get<bool>("ignore-dist-stf");
   std::vector<OutputSpec> outputs;
   outputs.emplace_back("TRD", "TRACKLETS", 0, Lifetime::Timeframe);
   outputs.emplace_back("TRD", "DIGITS", 0, Lifetime::Timeframe);
@@ -90,9 +91,14 @@ WorkflowSpec defineDataProcessing(ConfigContext const& cfgc)
   int idevice = 0;
   //  LOG(info) << "expected incoming data definition : " << inputspec;
   // this is probably never going to be used but would to nice to know hence here.
+  auto orig = o2::header::gDataOriginTRD;
+  std::vector<InputSpec> inputs{{"stf", ConcreteDataTypeMatcher{orig, "RAWDATA"}, Lifetime::Optional}};
+  if (askSTFDist) {
+    inputs.emplace_back("stdDist", "FLP", "DISTSUBTIMEFRAME", 0, Lifetime::Timeframe);
+  }
   workflow.emplace_back(DataProcessorSpec{
     std::string("trd-datareader"), // left as a string cast incase we append stuff to the string
-    select(std::string("x:TRD/" + inputspec).c_str()),
+    inputs,                        //select(std::string("x:TRD/" + inputspec).c_str()),
     outputs,
     algoSpec,
     Options{}});

Detectors/TRD/reconstruction/src/DataReaderTask.cxx

@@ -15,12 +15,16 @@
 
 #include "TRDReconstruction/DataReaderTask.h"
 #include "TRDReconstruction/CruRawReader.h"
+
 #include "Framework/ControlService.h"
 #include "Framework/ConfigParamRegistry.h"
 #include "Framework/RawDeviceService.h"
 #include "Framework/DeviceSpec.h"
 #include "Framework/DataSpecUtils.h"
+#include "Framework/InputRecordWalker.h"
+
 #include "DataFormatsTRD/Constants.h"
+
 #include <fairmq/FairMQDevice.h>
 
 //using namespace o2::framework;
@@ -62,28 +66,45 @@ void DataReaderTask::run(ProcessingContext& pc)
   auto outputRoutes = pc.services().get<o2::framework::RawDeviceService>().spec().outputs;
   auto fairMQChannel = outputRoutes.at(0).channel;
   int inputcount = 0;
+  std::vector<InputSpec> dummy{InputSpec{"filter", ConcreteDataTypeMatcher{"FLP", "DISTSUBTIMEFRAME"}, Lifetime::Timeframe}};
+  // 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
+
+  for (const auto& ref : InputRecordWalker(pc.inputs(), dummy)) {
+    const auto dh = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
+    if (dh->payloadSize == 16) {
+      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);
+      sendData(pc); //send the empty tf data.
+      return;
+    }
+    LOG(info) << " matched DEADBEEF";
+  }
+  //TODO combine the previous and subsequent loops.
+  int inputcounts = 0;
   /* loop over inputs routes */
   for (auto iit = pc.inputs().begin(), iend = pc.inputs().end(); iit != iend; ++iit) {
+    LOG(info) << " looping over inputs " << inputcounts;
+    inputcounts++;
     if (!iit.isValid()) {
       continue;
     }
     /* loop over input parts */
+    int inputpartscount = 0;
     for (auto const& ref : iit) {
-
+      LOG(info) << " looping over parts " << inputpartscount;
+      if (mVerbose) {
+        const auto dh = DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
+        LOGP(info, "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);
+      }
       const auto* headerIn = DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
       auto payloadIn = ref.payload;
       auto payloadInSize = headerIn->payloadSize;
       if (!mCompressedData) { //we have raw data coming in from flp
         if (mVerbose) {
-          LOG(info) << " parsing non compressed data in the data reader task";
+          LOG(info) << " parsing non compressed data in the data reader task with a payload of " << payloadInSize << " payload size";
         }
-
-        int a = 1;
-        int d = 1;
-        //        while(d==1){
-        //          a=sin(rand());
-        //        }
-
         mReader.setDataBuffer(payloadIn);
         mReader.setDataBufferSize(payloadInSize);
         mReader.configure(mByteSwap, mVerbose, mHeaderVerbose, mDataVerbose);