Eliminate obsolete options of raw-file-reader

shahor02 · shahor02 · commit 52aedc715ff9 · 2020-07-04T13:51:21.000+02:00
No need for options message-per-tf (must be always sent as a multipart of HBFs) and
output-per-link (must be always sent per output_channel)
diff --git a/Detectors/MUON/MCH/Workflow/README.md b/Detectors/MUON/MCH/Workflow/README.md
@@ -18,7 +18,7 @@
 
 ## Example of DPL chain:
 
-`o2-raw-file-reader-workflow --conf file-reader.cfg --loop 0 --message-per-tf -b | o2-mch-raw-to-digits-workflow -b | o2-mch-digits-to-preclusters-workflow -b | o2-mch-preclusters-sink-workflow -b`
+`o2-raw-file-reader-workflow --conf file-reader.cfg --loop 0  -b | o2-mch-raw-to-digits-workflow -b | o2-mch-digits-to-preclusters-workflow -b | o2-mch-preclusters-sink-workflow -b`
 
 where the `file-reader.cfg` looks like this:
 
diff --git a/Detectors/MUON/MID/Workflow/README.md b/Detectors/MUON/MID/Workflow/README.md
@@ -33,7 +33,7 @@ Notice that the executable also generate a configuration file that is needed to
 ### Reconstruction from raw data
 To reconstruct the raw data (either from converted MC digits or real data), run:
 ```bash
-o2-raw-file-reader-workflow --conf mid_raw.cfg --message-per-tf | o2-mid-reco-workflow
+o2-raw-file-reader-workflow --conf mid_raw.cfg | o2-mid-reco-workflow
 ```
 
 ## Timing
@@ -44,4 +44,4 @@ An example output is the following:
 Processing time / 90 ROFs: full: 3.55542 us  tracking: 2.02182 us
 ```
 Two timing values are provided: one is for the full execution of the device (including retrieval and sending of the DPL messages) and one which concerns only the execution of the algorithm (the tracking algorithm in the above example)
-The timing refers to the time needed to process one read-out-frame, i.e. one event.
+The timing refers to the time needed to process one read-out-frame, i.e. one event.
diff --git a/Detectors/Raw/README.md b/Detectors/Raw/README.md
@@ -230,6 +230,8 @@ dataOrigin = ITS
 #optional, if missing then default is used
 dataDescription = RAWDATA     
 filePath = path_and_name_of_the_data_file0
+# for CRU detectors the "readoutCard" record below is optional
+# readoutCard = CRU
 
 [input-1]
 dataOrigin = TPC              
@@ -238,6 +240,14 @@ filePath = path_and_name_of_the_data_file1
 [input-2]
 filePath = path_and_name_of_the_data_file2
 
+[input-1-RORC]
+dataOrigin = EMC
+filePath = path_and_name_of_the_data_fileX
+# for RORC detectors the record below is obligatory
+readoutCard = RORC 
+
+
+
 #...
 # [input-XXX] blocks w/o filePath will be ignoder, XXX is irrelevant
 
@@ -290,8 +300,6 @@ o2-raw-file-reader-workflow
   --loop arg (=1)                       loop N times (infinite for N<0)
   --min-tf arg (=0)                     min TF ID to process
   --max-tf arg (=4294967295)            max TF ID to process
-  --message-per-tf                      send TF of each link as a single FMQ message rather than multipart with message per HB
-  --output-per-link                     send message per Link rather than per FMQ output route
   --delay arg (=0)                      delay in seconds between consecutive TFs sending
   --configKeyValues arg                 semicolon separated key=value strings
 
@@ -315,10 +323,6 @@ If `--loop` argument is provided, data will be re-played in loop. The delay (in
 
 At every invocation of the device `processing` callback a full TimeFrame for every link will be added as N-HBFs parts (one for each HBF in the TF) to the multipart
 relayed by the `FairMQ` channel.
-In case the `--message-per-tf` option is asked, the whole TF is sent as the only part of the `FairMQPart`.
-
-Instead of sending a single output (for multiple links) per output route (which means their data will be received together) one can request sending an output per link
-by using option `--output-per-link`.
 
 The standard use case of this workflow is to provide the input for other worfklows using the piping, e.g.
 ```cpp
diff --git a/Detectors/Raw/include/DetectorsRaw/RawFileReader.h b/Detectors/Raw/include/DetectorsRaw/RawFileReader.h
@@ -92,7 +92,7 @@ class RawFileReader
   //================================================================================
 
   using RDHAny = header::RDHAny;
-  using RDH = o2::header::RAWDataHeaderV4;
+  using RDH = o2::header::RAWDataHeader;
   using OrigDescCard = std::tuple<o2::header::DataOrigin, o2::header::DataDescription, ReadoutCardType>;
   using InputsMap = std::map<OrigDescCard, std::vector<std::string>>;
 
diff --git a/Detectors/Raw/src/RawFileReaderWorkflow.cxx b/Detectors/Raw/src/RawFileReaderWorkflow.cxx
@@ -42,25 +42,18 @@ using namespace o2::raw;
 namespace o2f = o2::framework;
 namespace o2h = o2::header;
 
-class rawReaderSpecs : public o2f::Task
+class RawReaderSpecs : public o2f::Task
 {
  public:
-  explicit rawReaderSpecs(const std::string& config, bool tfAsMessage = false, bool outPerRoute = true, int loop = 1, uint32_t delay_us = 0,
+  explicit RawReaderSpecs(const std::string& config, int loop = 1, uint32_t delay_us = 0,
                           uint32_t errmap = 0xffffffff, uint32_t minTF = 0, uint32_t maxTF = 0xffffffff, size_t buffSize = 1024L * 1024L)
-    : mLoop(loop < 0 ? INT_MAX : (loop < 1 ? 1 : loop)), mHBFPerMessage(!tfAsMessage), mOutPerRoute(outPerRoute), mDelayUSec(delay_us), mMinTFID(minTF), mMaxTFID(maxTF), mReader(std::make_unique<o2::raw::RawFileReader>(config))
+    : mLoop(loop < 0 ? INT_MAX : (loop < 1 ? 1 : loop)), mDelayUSec(delay_us), mMinTFID(minTF), mMaxTFID(maxTF), mReader(std::make_unique<o2::raw::RawFileReader>(config))
   {
     mReader->setCheckErrors(errmap);
     mReader->setMaxTFToRead(maxTF);
     mReader->setBufferSize(buffSize);
     LOG(INFO) << "Will preprocess files with buffer size of " << buffSize << " bytes";
     LOG(INFO) << "Number of loops over whole data requested: " << mLoop;
-    if (mHBFPerMessage) {
-      LOG(INFO) << "Every link TF will be sent as multipart of HBF messages";
-    } else {
-      LOG(INFO) << "HBF of single TF of each link will be sent as a single message";
-    }
-    LOG(INFO) << "A message per " << (mOutPerRoute ? "route" : "link") << " will be sent";
-    LOG(INFO) << "Delay of " << mDelayUSec << " microseconds will be added between TFs";
   }
 
   void init(o2f::InitContext& ic) final
@@ -103,10 +96,6 @@ class rawReaderSpecs : public o2f::Task
     int nlinks = mReader->getNLinks();
 
     std::unordered_map<std::string, std::unique_ptr<FairMQParts>> messagesPerRoute;
-    std::vector<std::unique_ptr<FairMQParts>> messagesPerLink;
-    if (!mOutPerRoute) {
-      messagesPerLink.resize(nlinks);
-    }
 
     if (tfID > mMaxTFID) {
       if (mReader->getNTimeFrames() && --mLoop) {
@@ -144,18 +133,18 @@ class rawReaderSpecs : public o2f::Task
       o2h::DataHeader hdrTmpl(link.description, link.origin, link.subspec); // template with 0 size
       int nhb = link.getNHBFinTF();
       hdrTmpl.payloadSerializationMethod = o2h::gSerializationMethodNone;
-      hdrTmpl.splitPayloadParts = mHBFPerMessage ? nhb : 1;
+      hdrTmpl.splitPayloadParts = nhb;
 
       while (hdrTmpl.splitPayloadIndex < hdrTmpl.splitPayloadParts) {
 
-        tfSize += hdrTmpl.payloadSize = mHBFPerMessage ? link.getNextHBFSize() : link.getNextTFSize();
+        tfSize += hdrTmpl.payloadSize = link.getNextHBFSize();
         o2::header::Stack headerStack{hdrTmpl, o2::framework::DataProcessingHeader{mTFIDaccum}};
 
         auto hdMessage = device->NewMessage(headerStack.size());
         memcpy(hdMessage->GetData(), headerStack.data(), headerStack.size());
 
         auto plMessage = device->NewMessage(hdrTmpl.payloadSize);
-        auto bread = mHBFPerMessage ? link.readNextHBF(reinterpret_cast<char*>(plMessage->GetData())) : link.readNextTF(reinterpret_cast<char*>(plMessage->GetData()));
+        auto bread = link.readNextHBF(reinterpret_cast<char*>(plMessage->GetData()));
         if (bread != hdrTmpl.payloadSize) {
           LOG(ERROR) << "Link " << il << " read " << bread << " bytes instead of " << hdrTmpl.payloadSize
                      << " expected in TF=" << mTFIDaccum << " part=" << hdrTmpl.splitPayloadIndex;
@@ -175,18 +164,10 @@ class rawReaderSpecs : public o2f::Task
           reinterpret_cast<o2::header::DataHeader*>(hdMessage->GetData())->firstTForbit = hdrTmpl.firstTForbit;
         }
         FairMQParts* parts = nullptr;
-        if (mOutPerRoute) {
-          parts = messagesPerRoute[link.fairMQChannel].get(); // FairMQParts*
-          if (!parts) {
-            messagesPerRoute[link.fairMQChannel] = std::make_unique<FairMQParts>();
-            parts = messagesPerRoute[link.fairMQChannel].get();
-          }
-        } else {                             // message per link
-          parts = messagesPerLink[il].get(); // FairMQParts*
-          if (!parts) {
-            messagesPerLink[il] = std::make_unique<FairMQParts>();
-            parts = messagesPerLink[il].get();
-          }
+        parts = messagesPerRoute[link.fairMQChannel].get(); // FairMQParts*
+        if (!parts) {
+          messagesPerRoute[link.fairMQChannel] = std::make_unique<FairMQParts>();
+          parts = messagesPerRoute[link.fairMQChannel].get();
         }
         parts->AddPart(std::move(hdMessage));
         parts->AddPart(std::move(plMessage));
@@ -201,24 +182,13 @@ class rawReaderSpecs : public o2f::Task
       usleep(mDelayUSec);
     }
 
-    if (mOutPerRoute) {
-      for (auto& msgIt : messagesPerRoute) {
-        LOG(INFO) << "Sending " << msgIt.second->Size() / 2 << " parts to channel " << msgIt.first;
-        device->Send(*msgIt.second.get(), msgIt.first);
-      }
-    } else {
-      for (int il = 0; il < nlinks; il++) {
-        auto& link = mReader->getLink(il);
-        auto* parts = messagesPerLink[il].get(); // FairMQParts*
-        if (parts) {
-          LOG(INFO) << "Sending " << parts->Size() / 2 << " parts to channel " << link.fairMQChannel << " for " << link.describe();
-          device->Send(*parts, link.fairMQChannel);
-        }
-      }
+    for (auto& msgIt : messagesPerRoute) {
+      LOG(INFO) << "Sending " << msgIt.second->Size() / 2 << " parts to channel " << msgIt.first;
+      device->Send(*msgIt.second.get(), msgIt.first);
     }
 
     LOGF(INFO, "Sent payload of %zu bytes in %zu parts in %zu messages for TF %d", tfSize, tfNParts,
-         (mOutPerRoute ? messagesPerRoute.size() : messagesPerLink.size()), mTFIDaccum);
+         messagesPerRoute.size(), mTFIDaccum);
     sentSize += tfSize;
     sentMessages += tfNParts;
 
@@ -240,38 +210,34 @@ class rawReaderSpecs : public o2f::Task
   uint32_t mDelayUSec = 0;                         // Delay in microseconds between TFs
   uint32_t mMinTFID = 0;                           // 1st TF to extract
   uint32_t mMaxTFID = 0xffffffff;                  // last TF to extrct
-  bool mHBFPerMessage = true;                      // true: send TF as multipart of HBFs, false: single message per TF
-  bool mOutPerRoute = true;                        // true: send 1 large output route, otherwise 1 outpur per link
   bool mDone = false;                              // processing is over or not
   std::unique_ptr<o2::raw::RawFileReader> mReader; // matching engine
 };
 
-o2f::DataProcessorSpec getReaderSpec(std::string config, bool tfAsMessage, bool outPerRoute, int loop, uint32_t delay_us, uint32_t errmap,
+o2f::DataProcessorSpec getReaderSpec(std::string config, int loop, uint32_t delay_us, uint32_t errmap,
                                      uint32_t minTF, uint32_t maxTF, size_t buffSize)
 {
   // check which inputs are present in files to read
-  o2f::Outputs outputs;
+  o2f::DataProcessorSpec spec;
+  spec.name = "raw-file-reader";
+
   if (!config.empty()) {
     auto conf = o2::raw::RawFileReader::parseInput(config);
     for (const auto& entry : conf) {
       const auto& ordescard = entry.first;
       if (!entry.second.empty()) { // origin and decription for files to process
-        outputs.emplace_back(o2f::OutputSpec(o2f::ConcreteDataTypeMatcher{std::get<0>(ordescard), std::get<1>(ordescard)}));
+        spec.outputs.emplace_back(o2f::OutputSpec(o2f::ConcreteDataTypeMatcher{std::get<0>(ordescard), std::get<1>(ordescard)}));
       }
     }
   }
-  return o2f::DataProcessorSpec{
-    "raw-file-reader",
-    o2f::Inputs{},
-    outputs,
-    o2f::AlgorithmSpec{o2f::adaptFromTask<rawReaderSpecs>(config, tfAsMessage, outPerRoute, loop, delay_us, errmap, minTF, maxTF, buffSize)},
-    o2f::Options{}};
+  spec.algorithm = o2f::adaptFromTask<RawReaderSpecs>(config, loop, delay_us, errmap, minTF, maxTF, buffSize);
+
+  return spec;
 }
 
-o2f::WorkflowSpec o2::raw::getRawFileReaderWorkflow(std::string inifile, bool tfAsMessage, bool outPerRoute,
-                                                    int loop, uint32_t delay_us, uint32_t errmap, uint32_t minTF, uint32_t maxTF, size_t buffSize)
+o2f::WorkflowSpec o2::raw::getRawFileReaderWorkflow(std::string inifile, int loop, uint32_t delay_us, uint32_t errmap, uint32_t minTF, uint32_t maxTF, size_t buffSize)
 {
   o2f::WorkflowSpec specs;
-  specs.emplace_back(getReaderSpec(inifile, tfAsMessage, outPerRoute, loop, delay_us, errmap, minTF, maxTF, buffSize));
+  specs.emplace_back(getReaderSpec(inifile, loop, delay_us, errmap, minTF, maxTF, buffSize));
   return specs;
 }
diff --git a/Detectors/Raw/src/RawFileReaderWorkflow.h b/Detectors/Raw/src/RawFileReaderWorkflow.h
@@ -20,7 +20,7 @@ namespace o2
 namespace raw
 {
 
-framework::WorkflowSpec getRawFileReaderWorkflow(std::string inifile, bool tfAsMessage = false, bool outPerRoute = true,
+framework::WorkflowSpec getRawFileReaderWorkflow(std::string inifile,
                                                  int loop = 1, uint32_t delay_us = 0, uint32_t errMap = 0xffffffff,
                                                  uint32_t minTF = 0, uint32_t maxTF = 0xffffffff, size_t bufferSize = 1024L * 1024L);
 
diff --git a/Detectors/Raw/src/rawfile-reader-workflow.cxx b/Detectors/Raw/src/rawfile-reader-workflow.cxx
@@ -27,8 +27,6 @@ void customize(std::vector<o2::framework::ConfigParamSpec>& workflowOptions)
   options.push_back(ConfigParamSpec{"min-tf", VariantType::Int64, 0L, {"min TF ID to process"}});
   options.push_back(ConfigParamSpec{"max-tf", VariantType::Int64, 0xffffffffL, {"max TF ID to process"}});
   options.push_back(ConfigParamSpec{"loop", VariantType::Int, 1, {"loop N times (infinite for N<0)"}});
-  options.push_back(ConfigParamSpec{"message-per-tf", VariantType::Bool, false, {"send TF of each link as a single FMQ message rather than multipart with message per HB"}});
-  options.push_back(ConfigParamSpec{"output-per-link", VariantType::Bool, false, {"send message per Link rather than per FMQ output route"}});
   options.push_back(ConfigParamSpec{"delay", VariantType::Float, 0.f, {"delay in seconds between consecutive TFs sending"}});
   options.push_back(ConfigParamSpec{"buffer-size", VariantType::Int64, 1024L * 1024L, {"buffer size for files preprocessing"}});
   options.push_back(ConfigParamSpec{"configKeyValues", VariantType::String, "", {"semicolon separated key=value strings"}});
@@ -52,8 +50,6 @@ WorkflowSpec defineDataProcessing(ConfigContext const& configcontext)
   uint32_t maxTF = uint32_t(configcontext.options().get<int64_t>("max-tf"));
   uint32_t minTF = uint32_t(configcontext.options().get<int64_t>("min-tf"));
   uint64_t buffSize = uint64_t(configcontext.options().get<int64_t>("buffer-size"));
-  auto tfAsMessage = configcontext.options().get<bool>("message-per-tf");
-  auto outPerRoute = !configcontext.options().get<bool>("output-per-link");
 
   uint32_t errmap = 0;
   for (int i = RawFileReader::NErrorsDefined; i--;) {
@@ -66,5 +62,5 @@ WorkflowSpec defineDataProcessing(ConfigContext const& configcontext)
   o2::conf::ConfigurableParam::updateFromString(configcontext.options().get<std::string>("configKeyValues"));
   uint32_t delay_us = uint32_t(1e6 * configcontext.options().get<float>("delay")); // delay in microseconds
 
-  return std::move(o2::raw::getRawFileReaderWorkflow(inifile, tfAsMessage, outPerRoute, loop, delay_us, errmap, minTF, maxTF, buffSize));
+  return std::move(o2::raw::getRawFileReaderWorkflow(inifile, loop, delay_us, errmap, minTF, maxTF, buffSize));
 }

Original file line number	Diff line number	Diff line change
`@@ -20,7 +20,7 @@ namespace o2`
`20`	`20`	`namespace raw`
`21`	`21`	`{`
`22`	`22`
`23`		`-framework::WorkflowSpec getRawFileReaderWorkflow(std::string inifile, bool tfAsMessage = false, bool outPerRoute = true,`
	`23`	`+framework::WorkflowSpec getRawFileReaderWorkflow(std::string inifile,`
`24`	`24`	`int loop = 1, uint32_t delay_us = 0, uint32_t errMap = 0xffffffff,`
`25`	`25`	`uint32_t minTF = 0, uint32_t maxTF = 0xffffffff, size_t bufferSize = 1024L * 1024L);`
`26`	`26`