TPC workflow: sending data from publishers unserialized

Providing the data type of the objects to be published in the RootTreeReader
branch configuration to allow the DPL IO to decide upon serialization method.
The unserialized data is extracted as span on the receiver side.

Author: matthiasrichter

Detectors/TPC/workflow/include/TPCWorkflow/PublisherSpec.h

@@ -15,8 +15,12 @@
 
 #include "Framework/DataProcessorSpec.h"
 #include "Framework/OutputSpec.h"
+#include "Framework/DataSpecUtils.h"
+#include "Framework/Output.h"
+#include "DPLUtils/RootTreeReader.h"
 #include <vector>
 #include <string>
+#include <functional>
 
 namespace o2
 {
@@ -44,7 +48,44 @@ struct PublisherConf {
 
 /// create a processor spec
 /// read data from multiple tree branches from ROOT file and publish
-framework::DataProcessorSpec getPublisherSpec(PublisherConf const& config, bool propagateMC = true);
+template <typename T = void>
+framework::DataProcessorSpec getPublisherSpec(PublisherConf const& config, bool propagateMC = true)
+{
+  using Reader = o2::framework::RootTreeReader;
+  using Output = o2::framework::Output;
+  auto dto = o2::framework::DataSpecUtils::asConcreteDataTypeMatcher(config.dataoutput);
+  auto mco = o2::framework::DataSpecUtils::asConcreteDataTypeMatcher(config.mcoutput);
+
+  // a creator callback for the actual reader instance
+  auto creator = [dto, mco, propagateMC](const char* treename, const char* filename, int nofEvents, Reader::PublishingMode publishingMode, o2::header::DataHeader::SubSpecificationType subSpec, const char* branchname, const char* mcbranchname) {
+    constexpr auto persistency = o2::framework::Lifetime::Timeframe;
+    if (propagateMC) {
+      return std::make_shared<Reader>(treename,
+                                      filename,
+                                      nofEvents,
+                                      publishingMode,
+                                      Output{mco.origin, mco.description, subSpec, persistency},
+                                      mcbranchname,
+                                      Reader::BranchDefinition<T>{Output{dto.origin, dto.description, subSpec, persistency}, branchname});
+    } else {
+      return std::make_shared<Reader>(treename,
+                                      filename,
+                                      nofEvents,
+                                      publishingMode,
+                                      Reader::BranchDefinition<T>{Output{dto.origin, dto.description, subSpec, persistency}, branchname});
+    }
+  };
+
+  return createPublisherSpec(config, propagateMC, creator);
+}
+
+namespace workflow_reader
+{
+using Reader = o2::framework::RootTreeReader;
+using Creator = std::function<std::shared_ptr<Reader>(const char*, const char*, int, Reader::PublishingMode, o2::header::DataHeader::SubSpecificationType, const char*, const char*)>;
+} // namespace workflow_reader
+
+framework::DataProcessorSpec createPublisherSpec(PublisherConf const& config, bool propagateMC, workflow_reader::Creator creator);
 
 } // end namespace tpc
 } // end namespace o2

Detectors/TPC/workflow/src/ClustererSpec.cxx

@@ -91,7 +91,7 @@ DataProcessorSpec getClustererSpec(bool sendMC)
       if (!labelKey.empty()) {
         inMCLabels = std::move(pc.inputs().get<const MCLabelContainer*>(labelKey.c_str()));
       }
-      auto inDigits = pc.inputs().get<const std::vector<o2::tpc::Digit>>(inputKey.c_str());
+      auto inDigits = pc.inputs().get<gsl::span<o2::tpc::Digit>>(inputKey.c_str());
       if (verbosity > 0 && inMCLabels) {
         LOG(INFO) << "received " << inDigits.size() << " digits, "
                   << inMCLabels->getIndexedSize() << " MC label objects"

Detectors/TPC/workflow/src/PublisherSpec.cxx

@@ -15,10 +15,8 @@
 
 #include "Framework/ConfigParamRegistry.h"
 #include "Framework/ControlService.h"
-#include "Framework/DataSpecUtils.h"
 #include "TPCWorkflow/PublisherSpec.h"
 #include "Headers/DataHeader.h"
-#include "DPLUtils/RootTreeReader.h"
 #include "TPCBase/Sector.h"
 #include "DataFormatsTPC/TPCSectorHeader.h"
 #include <memory> // for make_shared, make_unique, unique_ptr
@@ -39,7 +37,7 @@ namespace tpc
 /// read data from multiple tree branches from ROOT file and publish
 /// data are expected to be stored in separated branches per sector, the default
 /// branch name is configurable, sector number is apended as extension '_n'
-DataProcessorSpec getPublisherSpec(PublisherConf const& config, bool propagateMC)
+DataProcessorSpec createPublisherSpec(PublisherConf const& config, bool propagateMC, workflow_reader::Creator creator)
 {
   if (config.tpcSectors.size() == 0 || config.outputIds.size() == 0) {
     throw std::invalid_argument("need TPC sector and output id configuration");
@@ -54,7 +52,7 @@ DataProcessorSpec getPublisherSpec(PublisherConf const& config, bool propagateMC
     bool finished = false;
   };
 
-  auto initFunction = [config, propagateMC](InitContext& ic) {
+  auto initFunction = [config, propagateMC, creator](InitContext& ic) {
     // get the option from the init context
     auto filename = ic.options().get<std::string>("infile");
     auto treename = ic.options().get<std::string>("treename");
@@ -88,7 +86,6 @@ DataProcessorSpec getPublisherSpec(PublisherConf const& config, bool propagateMC
       // set up the tree interface
       // TODO: parallelism on sectors needs to be implemented as selector in the reader
       // the data is now in parallel branches, as first attempt use an array of readers
-      constexpr auto persistency = Lifetime::Timeframe;
       auto outputId = outputIds.begin();
       for (size_t sector = 0; sector < NSectors; ++sector) {
         if ((activeSectors & ((uint64_t)0x1 << sector)) == 0) {
@@ -103,27 +100,14 @@ DataProcessorSpec getPublisherSpec(PublisherConf const& config, bool propagateMC
         }
         std::string clusterbranchname = clbrName + "_" + std::to_string(sector);
         std::string mcbranchname = mcbrName + "_" + std::to_string(sector);
-        auto dto = DataSpecUtils::asConcreteDataTypeMatcher(config.dataoutput);
-        auto mco = DataSpecUtils::asConcreteDataTypeMatcher(config.mcoutput);
-        if (propagateMC) {
-          readers[sector] = std::make_shared<RootTreeReader>(treename.c_str(),   // tree name
-                                                             sectorfile.c_str(), // input file name
-                                                             nofEvents,          // number of entries to publish
-                                                             publishingMode,
-                                                             Output{mco.origin, mco.description, subSpec, persistency},
-                                                             mcbranchname.c_str(), // name of mc label branch
-                                                             Output{dto.origin, dto.description, subSpec, persistency},
-                                                             clusterbranchname.c_str() // name of cluster branch
-          );
-        } else {
-          readers[sector] = std::make_shared<RootTreeReader>(treename.c_str(),   // tree name
-                                                             sectorfile.c_str(), // input file name
-                                                             nofEvents,          // number of entries to publish
-                                                             publishingMode,
-                                                             Output{dto.origin, dto.description, subSpec, persistency},
-                                                             clusterbranchname.c_str() // name of cluster branch
-          );
-        }
+        readers[sector] = creator(treename.c_str(),   // tree name
+                                  sectorfile.c_str(), // input file name
+                                  nofEvents,          // number of entries to publish
+                                  publishingMode,
+                                  subSpec,
+                                  clusterbranchname.c_str(), // name of data branch
+                                  mcbranchname.c_str()       // name of mc label branch
+        );
         if (++outputId == outputIds.end()) {
           outputId = outputIds.begin();
         }

Detectors/TPC/workflow/src/RecoWorkflow.cxx

@@ -112,17 +112,18 @@ framework::WorkflowSpec getWorkflow(std::vector<int> const& tpcSectors, std::vec
   // needs to be done in accordance. This means, if a new input option is added
   // also the dispatch trigger needs to be updated.
   if (inputType == InputType::Digits) {
-    specs.emplace_back(o2::tpc::getPublisherSpec(PublisherConf{
-                                                   "tpc-digit-reader",
-                                                   "o2sim",
-                                                   {"digitbranch", "TPCDigit", "Digit branch"},
-                                                   {"mcbranch", "TPCDigitMCTruth", "MC label branch"},
-                                                   OutputSpec{"TPC", "DIGITS"},
-                                                   OutputSpec{"TPC", "DIGITSMCTR"},
-                                                   tpcSectors,
-                                                   laneConfiguration,
-                                                 },
-                                                 propagateMC));
+    using Type = std::vector<o2::tpc::Digit>;
+    specs.emplace_back(o2::tpc::getPublisherSpec<Type>(PublisherConf{
+                                                         "tpc-digit-reader",
+                                                         "o2sim",
+                                                         {"digitbranch", "TPCDigit", "Digit branch"},
+                                                         {"mcbranch", "TPCDigitMCTruth", "MC label branch"},
+                                                         OutputSpec{"TPC", "DIGITS"},
+                                                         OutputSpec{"TPC", "DIGITSMCTR"},
+                                                         tpcSectors,
+                                                         laneConfiguration,
+                                                       },
+                                                       propagateMC));
   } else if (inputType == InputType::Raw) {
     specs.emplace_back(o2::tpc::getPublisherSpec(PublisherConf{
                                                    "tpc-raw-cluster-reader",