ensure empty frame is indeed empty

bazinski · bazinski · commit 5288e8faab53 · 2021-06-30T00:38:13.000+02:00
diff --git a/DataFormats/Detectors/TRD/include/DataFormatsTRD/EventRecord.h b/DataFormats/Detectors/TRD/include/DataFormatsTRD/EventRecord.h
@@ -18,6 +18,7 @@
 #include "CommonDataFormat/RangeReference.h"
 #include "FairLogger.h"
 #include "DataFormatsTRD/Tracklet64.h"
+#include "Framework/ProcessingContext.h"
 
 namespace o2::trd
 {
@@ -93,6 +94,9 @@ class EventStorage
   void addTracklets(InteractionRecord& ir, std::vector<Tracklet64>& tracklets);
   void addTracklets(InteractionRecord& ir, std::vector<Tracklet64>::iterator& start, std::vector<Tracklet64>::iterator& end);
   void unpackDataForSending(std::vector<TriggerRecord>& triggers, std::vector<Tracklet64>& tracklets, std::vector<Digit>& digits);
+  void sendData(o2::framework::ProcessingContext& pc);
+  //this could replace by keeing a running total on addition TODO
+  void sumTrackletsDigitsTriggers(uint64_t& tracklets, uint64_t& digits, uint64_t& triggers);
   int sumTracklets();
   int sumDigits();
   std::vector<Tracklet64>& getTracklets(InteractionRecord& ir);
diff --git a/DataFormats/Detectors/TRD/src/EventRecord.cxx b/DataFormats/Detectors/TRD/src/EventRecord.cxx
@@ -21,6 +21,18 @@
 #include "DataFormatsTRD/Digit.h"
 #include "DataFormatsTRD/EventRecord.h"
 #include "DataFormatsTRD/Constants.h"
+
+#include "Framework/Output.h"
+#include "Framework/ProcessingContext.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 <cassert>
 #include <array>
 #include <string>
@@ -142,14 +154,43 @@ void EventStorage::unpackDataForSending(std::vector<TriggerRecord>& triggers, st
 {
   int digitcount = 0;
   int trackletcount = 0;
-  for (auto event : mEventRecords) {
+  for (auto& event : mEventRecords) {
+    tracklets.insert(std::end(tracklets), std::begin(event.getTracklets()), std::end(event.getTracklets()));
+    digits.insert(std::end(digits), std::begin(event.getDigits()), std::end(event.getDigits()));
+    triggers.emplace_back(event.getBCData(), digitcount, event.getDigits().size(), trackletcount, event.getTracklets().size());
+    digitcount += event.getDigits().size();
+    trackletcount += event.getTracklets().size();
+  }
+}
+
+void EventStorage::sendData(o2::framework::ProcessingContext& pc)
+{
+  //at this point we know the total number of tracklets and digits and triggers.
+  //hence we can create the relevant objects inside the message as opposed to creating a local object and snapshotting it(copying) it
+  //into the message.
+  uint64_t trackletcount = 0;
+  uint64_t digitcount = 0;
+  uint64_t triggercount = 0;
+  sumTrackletsDigitsTriggers(trackletcount, digitcount, triggercount);
+  std::vector<Tracklet64> tracklets;
+  tracklets.reserve(trackletcount);
+  std::vector<Digit> digits;
+  digits.reserve(digitcount);
+  std::vector<TriggerRecord> triggers;
+  triggers.reserve(triggercount);
+  for (auto& event : mEventRecords) {
     tracklets.insert(std::end(tracklets), std::begin(event.getTracklets()), std::end(event.getTracklets()));
     digits.insert(std::end(digits), std::begin(event.getDigits()), std::end(event.getDigits()));
     triggers.emplace_back(event.getBCData(), digitcount, event.getDigits().size(), trackletcount, event.getTracklets().size());
     digitcount += event.getDigits().size();
     trackletcount += event.getTracklets().size();
   }
+  LOG(info) << "Sending data onwards with " << digits.size() << " Digits and " << tracklets.size() << " Tracklets and " << triggers.size() << " Triggers";
+  pc.outputs().snapshot(o2::framework::Output{o2::header::gDataOriginTRD, "DIGITS", 0, o2::framework::Lifetime::Timeframe}, digits);
+  pc.outputs().snapshot(o2::framework::Output{o2::header::gDataOriginTRD, "TRACKLETS", 0, o2::framework::Lifetime::Timeframe}, tracklets);
+  pc.outputs().snapshot(o2::framework::Output{o2::header::gDataOriginTRD, "TRKTRGRD", 0, o2::framework::Lifetime::Timeframe}, triggers);
 }
+
 int EventStorage::sumTracklets()
 {
   int sum = 0;
@@ -166,6 +207,18 @@ int EventStorage::sumDigits()
   }
   return sum;
 }
+void EventStorage::sumTrackletsDigitsTriggers(uint64_t& tracklets, uint64_t& digits, uint64_t& triggers)
+{
+  int digitsum = 0;
+  int trackletsum = 0;
+  int triggersum = 0;
+  for (auto event : mEventRecords) {
+    digitsum += event.getDigits().size();
+    trackletsum += event.getTracklets().size();
+    triggersum++;
+  }
+}
+
 std::vector<Tracklet64>& EventStorage::getTracklets(InteractionRecord& ir)
 {
   bool found = false;
diff --git a/Detectors/TRD/reconstruction/include/TRDReconstruction/CruRawReader.h b/Detectors/TRD/reconstruction/include/TRDReconstruction/CruRawReader.h
@@ -92,6 +92,8 @@ class CruRawReader
   std::vector<Digit>& getDigits(InteractionRecord& ir) { return mEventRecords.getDigits(ir); };
   //  std::vector<o2::trd::TriggerRecord> getIR() { return mEventTriggers; }
   void getParsedObjects(std::vector<Tracklet64>& tracklets, std::vector<Digit>& cdigits, std::vector<TriggerRecord>& triggers);
+  void getParsedObjectsandClear(std::vector<Tracklet64>& tracklets, std::vector<Digit>& digits, std::vector<TriggerRecord>& triggers);
+  void buildDPLOutputs(o2::framework::ProcessingContext& outputs);
   int getDigitsFound() { return mTotalDigitsFound; }
   int getTrackletsFound() { return mTotalTrackletsFound; }
   int sumTrackletsFound() { return mEventRecords.sumTracklets(); }
diff --git a/Detectors/TRD/reconstruction/include/TRDReconstruction/DataReaderTask.h b/Detectors/TRD/reconstruction/include/TRDReconstruction/DataReaderTask.h
@@ -57,7 +57,8 @@ class DataReaderTask : public Task
   bool mHeaderVerbose{false};  // verbose output of headers
   bool mCompressedData{false}; // are we dealing with the compressed data from the flp (send via option)
   bool mByteSwap{true};        // whether we are to byteswap the incoming data, mc is not byteswapped, raw data is (too be changed in cru at some point)
-  o2::header::DataDescription mDataSpec; // input spec of th raw incoming data
+                               //  o2::header::DataDescription mDataDesc; // Data description of the incoming data
+  std::string mDataDesc;
 };
 
 } // namespace o2::trd
diff --git a/Detectors/TRD/reconstruction/src/CruRawReader.cxx b/Detectors/TRD/reconstruction/src/CruRawReader.cxx
@@ -23,6 +23,14 @@
 #include "TRDReconstruction/TrackletsParser.h"
 #include "DataFormatsTRD/Constants.h"
 
+#include "Framework/ControlService.h"
+#include "Framework/ConfigParamRegistry.h"
+#include "Framework/RawDeviceService.h"
+#include "Framework/DeviceSpec.h"
+#include "Framework/DataSpecUtils.h"
+#include "Framework/Output.h"
+#include "Framework/InputRecordWalker.h"
+
 #include <cstring>
 #include <string>
 #include <vector>
@@ -415,31 +423,20 @@ void CruRawReader::getParsedObjects(std::vector<Tracklet64>& tracklets, std::vec
   int digitcountsum = 0;
   int trackletcountsum = 0;
   mEventRecords.unpackDataForSending(triggers, tracklets, digits);
-  /*for(auto eventrecord: mEventRecords)//loop over triggers incase they have already been done.
-  {
-  int digitcount=0;
-  int trackletcount=0;
-    int start,end;
-    LOG(info) << __func__ << " " << tracklets.size() << " "
-              << cdigits.size()<< " trackletv size:"<< mEventRecords.getTracklets(ir.getBCData());
-    for(auto trackletv: mEventStores.getTracklets(ir.getBCData())){
-      //loop through the vector of ranges
-      start=trackletv.getFirstEntry();
-      end= start+trackletv.getEntries();
-      LOG(info) << "insert tracklets from " << start<< "  " << end;
-      tracklets.insert(tracklets.end(),mEventTracklets.begin()+start, mEventTracklets.begin()+end);
-      trackletcount+=trackletv.getEntries();
-    }
-    for(auto digitv: mEventStores.getDigits(ir.getBCData())){
-      start=digitv.getFirstEntry();
-      end= start+digitv.getEntries();
-      LOG(info) << "insert digits from " << start<< "  " << end;
-      cdigits.insert(cdigits.end(),mEventCompressedDigits.begin()+start , mEventCompressedDigits.begin()+end);
-      digitcount+=digitv.getEntries();
-    }
-    triggers.emplace_back(ir.getBCData(),digitcountsum,digitcount,trackletcountsum,trackletcount);
-    digitcountsum+=digitcount;
-    trackletcountsum+=trackletcount;
-  }*/
 }
+
+void CruRawReader::getParsedObjectsandClear(std::vector<Tracklet64>& tracklets, std::vector<Digit>& digits, std::vector<TriggerRecord>& triggers)
+{
+  getParsedObjects(tracklets, digits, triggers);
+  clearall();
+}
+
+//write the output data directly to the given DataAllocator from the datareader task.
+void CruRawReader::buildDPLOutputs(o2::framework::ProcessingContext& pc)
+{
+  mEventRecords.sendData(pc);
+  //    pc.outputs().snapshot(Output{o2::header::gDataOriginTRD,"STATS",0,Lifetime::Timerframe},mStats);
+  clearall(); // having now written the messages clear for next.
+}
+
 } // namespace o2::trd
diff --git a/Detectors/TRD/reconstruction/src/DataReader.cxx b/Detectors/TRD/reconstruction/src/DataReader.cxx
@@ -76,23 +76,11 @@ WorkflowSpec defineDataProcessing(ConfigContext const& cfgc)
 
   WorkflowSpec workflow;
 
-  /*
-   * This is originally replicated from TOF
-     We define at run time the number of devices to be attached
-     to the workflow and the input matching string of the device.
-     This is is done with a configuration string like the following
-     one, where the input matching for each device is provide in
-     comma-separated strings. For instance
-  */
-
-  //  std::stringstream ssconfig(inputspec);
   std::string iconfig;
   std::string inputDescription;
   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;
-  auto inputs = o2::framework::select(inputspec.c_str());
+  auto inputs = o2::framework::select(std::string("x:TRD/" + inputspec).c_str());
   for (auto& inp : inputs) {
     // take care of case where our data is not in the time frame
     inp.lifetime = Lifetime::Optional;
diff --git a/Detectors/TRD/reconstruction/src/DataReaderTask.cxx b/Detectors/TRD/reconstruction/src/DataReaderTask.cxx
@@ -47,14 +47,15 @@ void DataReaderTask::sendData(ProcessingContext& pc, bool blankframe)
 {
   // mReader.getParsedObjects(mTracklets,mDigits,mTriggers);
   if (!blankframe) {
-    mReader.getParsedObjects(mTracklets, mDigits, mTriggers);
+    mReader.buildDPLOutputs(pc); //getParsedObjectsandClear(mTracklets, mDigits, mTriggers);
+  } else {
+    //ensure the objects we are sending back are indeed blank.
+    LOG(info) << "Sending data onwards with " << mDigits.size() << " Digits and " << mTracklets.size() << " Tracklets and " << mTriggers.size() << " Triggers and blankframe:" << blankframe;
+    pc.outputs().snapshot(Output{o2::header::gDataOriginTRD, "DIGITS", 0, Lifetime::Timeframe}, mDigits);
+    pc.outputs().snapshot(Output{o2::header::gDataOriginTRD, "TRACKLETS", 0, Lifetime::Timeframe}, mTracklets);
+    pc.outputs().snapshot(Output{o2::header::gDataOriginTRD, "TRKTRGRD", 0, Lifetime::Timeframe}, mTriggers);
+    //    pc.outputs().snapshot(Output{o2::header::gDataOriginTRD,"STATS",0,Lifetime::Timerframe},mStats);
   }
-
-  LOG(info) << "Sending data onwards with " << mDigits.size() << " Digits and " << mTracklets.size() << " Tracklets and " << mTriggers.size() << " Triggers and blankframe:" << blankframe;
-  pc.outputs().snapshot(Output{o2::header::gDataOriginTRD, "DIGITS", 0, Lifetime::Timeframe}, mDigits);
-  pc.outputs().snapshot(Output{o2::header::gDataOriginTRD, "TRACKLETS", 0, Lifetime::Timeframe}, mTracklets);
-  pc.outputs().snapshot(Output{o2::header::gDataOriginTRD, "TRKTRGRD", 0, Lifetime::Timeframe}, mTriggers);
-  //    pc.outputs().snapshot(Output{o2::header::gDataOriginTRD,"STATS",0,Lifetime::Timerframe},mStats);
 }
 
 void DataReaderTask::run(ProcessingContext& pc)
@@ -67,16 +68,16 @@ void DataReaderTask::run(ProcessingContext& pc)
   auto device = pc.services().get<o2::framework::RawDeviceService>().device();
   auto outputRoutes = pc.services().get<o2::framework::RawDeviceService>().spec().outputs;
   auto fairMQChannel = outputRoutes.at(0).channel;
-  mDataSpec = o2::header::gDataDescriptionRawData;
 
-  std::vector<InputSpec> dummy{InputSpec{"dummy", ConcreteDataMatcher{"TRD", mDataSpec, 0xDEADBEEF}}};
-  // if we see requested data type input with 0xDEADBEEF subspec and 0 payload this means that the "delayed message"
+  std::vector<InputSpec> dummy{InputSpec{"dummy", ConcreteDataMatcher{"TRD", "RAWDATA", 0xDEADBEEF}}};
+  //std::vector<InputSpec> dummy{InputSpec{"dummy", ConcreteDataMatcher{"TRD","RAWDATA"/* mDataDesc.c_str()*/, 0xDEADBEEF}}};
+  // if we see requested data type input with 0xDEADBEEF subspec and 0 payload this mecans 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 || dh->payloadSize == 0) {
-      LOGP(WARNING, "Found input [{}/{}/{:#x}] TF#{} 1st_orbit:{} Payload {} : assuming no payload for all links in this TF",
+    if (dh->payloadSize == 0) { //}|| dh->payloadSize==16) {
+      LOGP(WARNING, "Found blank input input [{}/{}/{:#x}] TF#{} 1st_orbit:{} Payload {} : ",
            dh->dataOrigin.str, dh->dataDescription.str, dh->subSpecification, dh->tfCounter, dh->firstTForbit, dh->payloadSize);
       sendData(pc, true); //send the empty tf data.
       return;
@@ -94,49 +95,50 @@ void DataReaderTask::run(ProcessingContext& pc)
     for (auto const& ref : iit) {
       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",
+        LOGP(info, "Found input [{}/{}/{:#x}] TF#{} 1st_orbit:{} Payload {} : ",
              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 with a payload of " << payloadInSize << " payload size";
-        }
-        mReader.setDataBuffer(payloadIn);
-        mReader.setDataBufferSize(payloadInSize);
-        mReader.configure(mByteSwap, mVerbose, mHeaderVerbose, mDataVerbose);
-        if (mVerbose) {
-          LOG(info) << "%%% about to run " << loopcounter << " %%%";
-        }
-        mReader.run();
-        if (mVerbose) {
-          LOG(info) << "%%% finished running " << loopcounter << " %%%";
-        }
-        loopcounter++;
-        // mTracklets.insert(std::end(mTracklets), std::begin(mReader.getTracklets()), std::end(mReader.getTracklets()));
-        // mCompressedDigits.insert(std::end(mCompressedDigits), std::begin(mReader.getCompressedDigits()), std::end(mReader.getCompressedDigits()));
-        //mReader.clearall();
-        if (mVerbose) {
-          LOG(info) << "from parsing received: " << mTracklets.size() << " tracklets and " << mDigits.size() << " compressed digits";
-          LOG(info) << "relevant vectors to read : " << mReader.sumTrackletsFound() << " tracklets and " << mReader.sumDigitsFound() << " compressed digits";
+      if (std::string(headerIn->dataDescription.str) != std::string("DISTSUBTIMEFRAMEFLP")) {
+        if (!mCompressedData) { //we have raw data coming in from flp
+          if (mVerbose) {
+            LOG(info) << " parsing non compressed data in the data reader task with a payload of " << payloadInSize << " payload size";
+          }
+          mReader.setDataBuffer(payloadIn);
+          mReader.setDataBufferSize(payloadInSize);
+          mReader.configure(mByteSwap, mVerbose, mHeaderVerbose, mDataVerbose);
+          if (mVerbose) {
+            LOG(info) << "%%% about to run " << loopcounter << " %%%";
+          }
+          mReader.run();
+          if (mVerbose) {
+            LOG(info) << "%%% finished running " << loopcounter << " %%%";
+          }
+          loopcounter++;
+          // mTracklets.insert(std::end(mTracklets), std::begin(mReader.getTracklets()), std::end(mReader.getTracklets()));
+          // mCompressedDigits.insert(std::end(mCompressedDigits), std::begin(mReader.getCompressedDigits()), std::end(mReader.getCompressedDigits()));
+          //mReader.clearall();
+          if (mVerbose) {
+            LOG(info) << "from parsing received: " << mTracklets.size() << " tracklets and " << mDigits.size() << " compressed digits";
+            LOG(info) << "relevant vectors to read : " << mReader.sumTrackletsFound() << " tracklets and " << mReader.sumDigitsFound() << " compressed digits";
+          }
+          //  mTriggers = mReader.getIR();
+          //get the payload of trigger and digits out.
+        } else { // we have compressed data coming in from flp.
+          mCompressedReader.setDataBuffer(payloadIn);
+          mCompressedReader.setDataBufferSize(payloadInSize);
+          mCompressedReader.configure(mByteSwap, mVerbose, mHeaderVerbose, mDataVerbose);
+          mCompressedReader.run();
+          //get the payload of trigger and digits out.
         }
-        //  mTriggers = mReader.getIR();
-        //get the payload of trigger and digits out.
-      } else { // we have compressed data coming in from flp.
-        mCompressedReader.setDataBuffer(payloadIn);
-        mCompressedReader.setDataBufferSize(payloadInSize);
-        mCompressedReader.configure(mByteSwap, mVerbose, mHeaderVerbose, mDataVerbose);
-        mCompressedReader.run();
-        mTracklets = mCompressedReader.getTracklets();
-        mDigits = mCompressedReader.getDigits();
-        mTriggers = mCompressedReader.getIR();
-        //get the payload of trigger and digits out.
+        /* output */
+        sendData(pc, false); //TODO do we ever have to not post the data. i.e. can we get here mid event? I dont think so.
+      } else {
+        sendData(pc, true);
       }
     }
-    /* output */
-    sendData(pc, false); //TODO do we ever have to not post the data. i.e. can we get here mid event? I dont think so.
   }
 
   auto dataReadTime = std::chrono::high_resolution_clock::now() - dataReadStart;
