GPU Workflow: Add a 'standalone' GPU workflow, currenly basically identical to the processing part of the tpc-reco-workflow, to be extended...

davidrohr · davidrohr · commit 25aabf361dc3 · 2021-04-25T11:02:45.000+02:00
diff --git a/Detectors/GlobalTrackingWorkflow/src/cosmics-match-workflow.cxx b/Detectors/GlobalTrackingWorkflow/src/cosmics-match-workflow.cxx
@@ -10,7 +10,7 @@
 
 #include "CommonUtils/ConfigurableParam.h"
 #include "Framework/CompletionPolicy.h"
-#include "TPCWorkflow/TPCSectorCompletionPolicy.h"
+#include "GPUWorkflow/TPCSectorCompletionPolicy.h"
 #include "ITSWorkflow/TrackReaderSpec.h"
 #include "ITSMFTWorkflow/ClusterReaderSpec.h"
 #include "TPCWorkflow/TrackReaderSpec.h"
diff --git a/Detectors/GlobalTrackingWorkflow/src/tpcits-match-workflow.cxx b/Detectors/GlobalTrackingWorkflow/src/tpcits-match-workflow.cxx
@@ -11,7 +11,7 @@
 #include "GlobalTrackingWorkflow/MatchTPCITSWorkflow.h"
 #include "CommonUtils/ConfigurableParam.h"
 #include "Framework/CompletionPolicy.h"
-#include "TPCWorkflow/TPCSectorCompletionPolicy.h"
+#include "GPUWorkflow/TPCSectorCompletionPolicy.h"
 #include "DetectorsRaw/HBFUtilsInitializer.h"
 
 using namespace o2::framework;
diff --git a/Detectors/GlobalTrackingWorkflow/tpcinterpolationworkflow/src/tpc-interpolation-workflow.cxx b/Detectors/GlobalTrackingWorkflow/tpcinterpolationworkflow/src/tpc-interpolation-workflow.cxx
@@ -11,7 +11,7 @@
 #include "TPCInterpolationWorkflow/TrackInterpolationWorkflow.h"
 #include "CommonUtils/ConfigurableParam.h"
 #include "Framework/CompletionPolicy.h"
-#include "TPCWorkflow/TPCSectorCompletionPolicy.h"
+#include "GPUWorkflow/TPCSectorCompletionPolicy.h"
 #include "DetectorsRaw/HBFUtilsInitializer.h"
 
 using namespace o2::framework;
diff --git a/Detectors/TPC/workflow/CMakeLists.txt b/Detectors/TPC/workflow/CMakeLists.txt
@@ -19,7 +19,6 @@ o2_add_library(TPCWorkflow
                        src/TrackReaderSpec.cxx
                        src/RawToDigitsSpec.cxx
                        src/LinkZSToDigitsSpec.cxx
-                       src/TPCSectorCompletionPolicy.cxx
                        src/ZSSpec.cxx
                        src/CalibProcessingHelper.cxx
                        src/ClusterSharingMapSpec.cxx
diff --git a/Detectors/TPC/workflow/src/RecoWorkflow.cxx b/Detectors/TPC/workflow/src/RecoWorkflow.cxx
@@ -413,7 +413,7 @@ framework::WorkflowSpec getWorkflow(CompletionPolicyData* policyData, std::vecto
   //
   // selected by output type 'tracks'
   if (runTracker) {
-    o2::gpu::GPUWorkflow::Config cfg;
+    o2::gpu::gpuworkflow::Config cfg;
     cfg.decompressTPC = decompressTPC;
     cfg.decompressTPCFromROOT = decompressTPC && inputType == InputType::CompClusters;
     cfg.caClusterer = caClusterer;
@@ -428,7 +428,7 @@ framework::WorkflowSpec getWorkflow(CompletionPolicyData* policyData, std::vecto
     cfg.processMC = propagateMC;
     cfg.sendClustersPerSector = isEnabled(OutputType::SendClustersPerSector);
     cfg.askDISTSTF = askDISTSTF;
-    specs.emplace_back(o2::gpu::getGPURecoWorkflowSpec(policyData, cfg, tpcSectors, tpcSectorMask));
+    specs.emplace_back(o2::gpu::getGPURecoWorkflowSpec(policyData, cfg, tpcSectors, tpcSectorMask, "tpc-tracker"));
   }
 
   //////////////////////////////////////////////////////////////////////////////////////////////
diff --git a/Detectors/TPC/workflow/src/TPCSectorCompletionPolicy.cxx b/Detectors/TPC/workflow/src/TPCSectorCompletionPolicy.cxx
diff --git a/Detectors/TPC/workflow/src/TrackReaderWorkflow.cxx b/Detectors/TPC/workflow/src/TrackReaderWorkflow.cxx
@@ -18,7 +18,6 @@
 #include "Framework/PartRef.h"
 #include "Framework/ConcreteDataMatcher.h"
 #include "TPCWorkflow/RecoWorkflow.h"
-#include "TPCWorkflow/TPCSectorCompletionPolicy.h"
 #include "DataFormatsTPC/TPCSectorHeader.h"
 #include "Algorithm/RangeTokenizer.h"
 #include "CommonUtils/ConfigurableParam.h"
diff --git a/Detectors/TPC/workflow/src/tpc-reco-workflow.cxx b/Detectors/TPC/workflow/src/tpc-reco-workflow.cxx
@@ -21,7 +21,7 @@
 #include "Framework/PartRef.h"
 #include "Framework/ConcreteDataMatcher.h"
 #include "TPCWorkflow/RecoWorkflow.h"
-#include "TPCWorkflow/TPCSectorCompletionPolicy.h"
+#include "GPUWorkflow/TPCSectorCompletionPolicy.h"
 #include "DataFormatsTPC/TPCSectorHeader.h"
 #include "Algorithm/RangeTokenizer.h"
 #include "CommonUtils/ConfigurableParam.h"
@@ -94,7 +94,7 @@ void customize(std::vector<o2::framework::CompletionPolicy>& policies)
   policies.push_back(CompletionPolicyHelpers::defineByName("tpc-cluster-decoder.*", CompletionPolicy::CompletionOp::Consume));
   policies.push_back(CompletionPolicyHelpers::defineByName("tpc-clusterer.*", CompletionPolicy::CompletionOp::Consume));
   // the custom completion policy for the tracker
-  policies.push_back(o2::tpc::TPCSectorCompletionPolicy("gpu-reconstruction.*", o2::tpc::TPCSectorCompletionPolicy::Config::RequireAll, &gPolicyData, &gTpcSectorMask)());
+  policies.push_back(o2::tpc::TPCSectorCompletionPolicy("tpc-tracker.*", o2::tpc::TPCSectorCompletionPolicy::Config::RequireAll, &gPolicyData, &gTpcSectorMask)());
 }
 
 #include "Framework/runDataProcessing.h" // the main driver
diff --git a/GPU/Workflow/CMakeLists.txt b/GPU/Workflow/CMakeLists.txt
@@ -27,3 +27,10 @@ o2_add_executable(display
                               O2::TPCReconstruction
                               O2::GlobalTrackingWorkflowHelpers)
 target_include_directories(${targetName} PUBLIC "include")
+
+o2_add_executable(reco-workflow
+        COMPONENT_NAME gpu
+        TARGETVARNAME targetName
+        SOURCES src/gpu-reco-workflow.cxx
+        PUBLIC_LINK_LIBRARIES O2::GPUWorkflow
+                              O2::GlobalTrackingWorkflowHelpers)
diff --git a/GPU/Workflow/include/GPUWorkflow/GPUWorkflowSpec.h b/GPU/Workflow/include/GPUWorkflow/GPUWorkflowSpec.h
@@ -17,7 +17,8 @@
 #define O2_GPU_WORKFLOW_SPEC_H
 
 #include "Framework/DataProcessorSpec.h"
-#include <utility> // std::forward
+#include <string>
+#include <vector>
 
 namespace o2
 {
@@ -28,7 +29,7 @@ struct CompletionPolicy;
 
 namespace gpu
 {
-namespace GPUWorkflow
+namespace gpuworkflow
 {
 struct Config {
   bool decompressTPC = false;
@@ -47,7 +48,7 @@ struct Config {
   bool askDISTSTF = true;
 };
 using CompletionPolicyData = std::vector<framework::InputSpec>;
-} // namespace GPUWorkflow
+} // namespace gpuworkflow
 
 /// create a processor spec for the CATracker
 /// The CA tracker is actually much more than the tracker it has evolved to a
@@ -65,7 +66,7 @@ using CompletionPolicyData = std::vector<framework::InputSpec>;
 ///
 /// @param specconfig configuration options for the processor spec
 /// @param tpcsectors list of sector numbers
-framework::DataProcessorSpec getGPURecoWorkflowSpec(GPUWorkflow::CompletionPolicyData* policyData, GPUWorkflow::Config const& specconfig, std::vector<int> const& tpcsectors, unsigned long tpcSectorMask);
+framework::DataProcessorSpec getGPURecoWorkflowSpec(gpuworkflow::CompletionPolicyData* policyData, gpuworkflow::Config const& specconfig, std::vector<int> const& tpcsectors, unsigned long tpcSectorMask, std::string processorName);
 
 o2::framework::CompletionPolicy getCATrackerCompletionPolicy();
 } // end namespace gpu
diff --git a/GPU/Workflow/include/GPUWorkflow/TPCSectorCompletionPolicy.h b/GPU/Workflow/include/GPUWorkflow/TPCSectorCompletionPolicy.h
@@ -199,17 +199,17 @@ class TPCSectorCompletionPolicy
   void init(Arg&& arg, Args&&... args)
   {
     using Type = std::decay_t<Arg>;
-    if constexpr (std::is_same<Type, framework::InputSpec>::value) {
+    if constexpr (std::is_same_v<Type, framework::InputSpec>) {
       mInputMatchers.emplace_back(std::move(arg));
-    } else if constexpr (std::is_same<Type, TPCSectorCompletionPolicy::Config>::value) {
+    } else if constexpr (std::is_same_v<Type, TPCSectorCompletionPolicy::Config>) {
       switch (arg) {
         case Config::RequireAll:
           mRequireAll = true;
           break;
       }
-    } else if constexpr (std::is_same<Type, std::vector<o2::framework::InputSpec>*>::value) {
+    } else if constexpr (std::is_same_v<Type, std::vector<o2::framework::InputSpec>*>) {
       mExternalInputMatchers = arg;
-    } else if constexpr (std::is_same<Type, unsigned long*>::value) {
+    } else if constexpr (std::is_same_v<Type, unsigned long*> || std::is_same_v<Type, const unsigned long*>) {
       mTpcSectorMask = arg;
     } else {
       static_assert(framework::always_static_assert_v<Type>);
@@ -225,7 +225,7 @@ class TPCSectorCompletionPolicy
   // - They are controlled externally and the external entity can modify them, e.g. after parsing command line arguments.
   // - They are all matched independently, it is not sufficient that one of them is present for all sectors
   const std::vector<framework::InputSpec>* mExternalInputMatchers = nullptr;
-  unsigned long* mTpcSectorMask = nullptr;
+  const unsigned long* mTpcSectorMask = nullptr;
   bool mRequireAll = false;
 };
 } // namespace tpc
diff --git a/GPU/Workflow/src/GPUWorkflowSpec.cxx b/GPU/Workflow/src/GPUWorkflowSpec.cxx
@@ -83,7 +83,7 @@ using namespace o2::tpc;
 
 namespace o2::gpu
 {
-DataProcessorSpec getGPURecoWorkflowSpec(GPUWorkflow::CompletionPolicyData* policyData, GPUWorkflow::Config const& specconfig, std::vector<int> const& tpcsectors, unsigned long tpcSectorMask)
+DataProcessorSpec getGPURecoWorkflowSpec(gpuworkflow::CompletionPolicyData* policyData, gpuworkflow::Config const& specconfig, std::vector<int> const& tpcsectors, unsigned long tpcSectorMask, std::string processorName)
 {
   if (specconfig.outputCAClusters && !specconfig.caClusterer && !specconfig.decompressTPC) {
     throw std::runtime_error("inconsistent configuration: cluster output is only possible if CA clusterer is activated");
@@ -791,7 +791,7 @@ DataProcessorSpec getGPURecoWorkflowSpec(GPUWorkflow::CompletionPolicyData* poli
     return std::move(outputSpecs);
   };
 
-  return DataProcessorSpec{"gpu-reconstruction", // process id
+  return DataProcessorSpec{processorName, // process id
                            {createInputSpecs()},
                            {createOutputSpecs()},
                            AlgorithmSpec(initFunction)};
diff --git a/GPU/Workflow/src/gpu-reco-workflow.cxx b/GPU/Workflow/src/gpu-reco-workflow.cxx
@@ -0,0 +1,146 @@
+// Copyright CERN and copyright holders of ALICE O2. This software is
+// distributed under the terms of the GNU General Public License v3 (GPL
+// Version 3), copied verbatim in the file "COPYING".
+//
+// See http://alice-o2.web.cern.ch/license for full licensing information.
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+/// @file   gpu-reco-workflow.cxx
+/// @author David Rohr
+
+#include "Framework/WorkflowSpec.h"
+#include "Framework/ConfigParamSpec.h"
+#include "Framework/CompletionPolicyHelpers.h"
+#include "Framework/DispatchPolicy.h"
+#include "Framework/ConcreteDataMatcher.h"
+#include "GPUWorkflow/TPCSectorCompletionPolicy.h"
+#include "GPUWorkflow/GPUWorkflowSpec.h"
+#include "CommonUtils/ConfigurableParam.h"
+#include "DetectorsRaw/HBFUtilsInitializer.h"
+#include "TPCBase/Sector.h"
+#include "Algorithm/RangeTokenizer.h"
+#include "GlobalTrackingWorkflowHelpers/InputHelper.h"
+#include "ReconstructionDataFormats/GlobalTrackID.h"
+
+#include <unordered_map>
+
+using namespace o2::framework;
+using namespace o2::dataformats;
+using CompletionPolicyData = std::vector<InputSpec>;
+CompletionPolicyData gPolicyData;
+static constexpr unsigned long gTpcSectorMask = 0xFFFFFFFFF;
+
+void customize(std::vector<ConfigParamSpec>& workflowOptions)
+{
+
+  std::vector<ConfigParamSpec> options{
+    {"input-type", VariantType::String, "digits", {"digitizer, digits, zsraw, zsonthefly, clustersnative, compressed-clusters-root, compressed-clusters-ctf"}},
+    {"output-type", VariantType::String, "tracks", {"clustersnative, tracks, compressed-clusters, qa, no-shared-cluster-map"}},
+    {"disable-root-input", VariantType::Bool, true, {"disable root-files input reader"}},
+    {"disable-mc", VariantType::Bool, false, {"disable sending of MC information"}},
+    {"ignore-dist-stf", VariantType::Bool, false, {"do not subscribe to FLP/DISTSUBTIMEFRAME/0 message (no lost TF recovery)"}},
+    {"configKeyValues", VariantType::String, "", {"Semicolon separated key=value strings (e.g.: 'TPCHwClusterer.peakChargeThreshold=4;...')"}},
+    {"configFile", VariantType::String, "", {"configuration file for configurable parameters"}}};
+
+  o2::raw::HBFUtilsInitializer::addConfigOption(options);
+
+  std::swap(workflowOptions, options);
+}
+
+// customize dispatch policy, dispatch immediately what is ready
+void customize(std::vector<DispatchPolicy>& policies)
+{
+  using DispatchOp = DispatchPolicy::DispatchOp;
+  policies.push_back({"prompt-for-gpu-reco", [](auto const& spec) { return true; }, DispatchOp::WhenReady});
+}
+
+void customize(std::vector<CompletionPolicy>& policies)
+{
+  policies.push_back(o2::tpc::TPCSectorCompletionPolicy("gpu-reconstruction.*", o2::tpc::TPCSectorCompletionPolicy::Config::RequireAll, &gPolicyData, &gTpcSectorMask)());
+}
+
+#include "Framework/runDataProcessing.h" // the main driver
+
+using namespace o2::framework;
+
+enum struct ioType { Digits,
+                     Clusters,
+                     ZSRaw,
+                     ZSRawOTF,
+                     CompClustROOT,
+                     CompClustCTF,
+                     Tracks,
+                     QA,
+                     NoSharedMap };
+
+static const std::unordered_map<std::string, ioType> InputMap{
+  {"digits", ioType::Digits},
+  {"clusters", ioType::Clusters},
+  {"zsraw", ioType::ZSRaw},
+  {"zsonthefly", ioType::ZSRawOTF},
+  {"compressed-clusters-root", ioType::CompClustROOT},
+  {"compressed-clusters-ctf", ioType::CompClustCTF}};
+
+static const std::unordered_map<std::string, ioType> OutputMap{
+  {"clusters", ioType::Clusters},
+  {"tracks", ioType::Tracks},
+  {"compressed-clusters", ioType::CompClustCTF},
+  {"qa", ioType::QA},
+  {"no-shared-cluster-map", ioType::NoSharedMap}};
+
+WorkflowSpec defineDataProcessing(ConfigContext const& cfgc)
+{
+  WorkflowSpec specs;
+  std::vector<int> tpcSectors(o2::tpc::Sector::MAXSECTOR);
+  std::iota(tpcSectors.begin(), tpcSectors.end(), 0);
+
+  auto inputType = cfgc.options().get<std::string>("input-type");
+  bool doMC = !cfgc.options().get<bool>("disable-mc");
+
+  o2::conf::ConfigurableParam::updateFromFile(cfgc.options().get<std::string>("configFile"));
+  o2::conf::ConfigurableParam::updateFromString(cfgc.options().get<std::string>("configKeyValues"));
+  o2::conf::ConfigurableParam::writeINI("o2gpurecoworkflow_configuration.ini");
+
+  std::vector<ioType> outputTypes, inputTypes;
+  try {
+    outputTypes = o2::RangeTokenizer::tokenize<ioType>(cfgc.options().get<std::string>("output-type"), [](std::string const& token) { return OutputMap.at(token); });
+    inputTypes = o2::RangeTokenizer::tokenize<ioType>(cfgc.options().get<std::string>("input-type"), [](std::string const& token) { return InputMap.at(token); });
+  } catch (std::out_of_range&) {
+    throw std::invalid_argument("invalid input / output type");
+  }
+
+  auto isEnabled = [](auto& list, ioType type) {
+    return std::find(list.begin(), list.end(), type) != list.end();
+  };
+
+  o2::gpu::gpuworkflow::Config cfg;
+  cfg.decompressTPC = isEnabled(inputTypes, ioType::CompClustCTF);
+  cfg.decompressTPCFromROOT = isEnabled(inputTypes, ioType::CompClustROOT);
+  cfg.zsDecoder = isEnabled(inputTypes, ioType::ZSRaw);
+  cfg.zsOnTheFly = isEnabled(inputTypes, ioType::ZSRawOTF);
+  cfg.caClusterer = cfg.zsDecoder || cfg.zsOnTheFly || isEnabled(inputTypes, ioType::Digits);
+  cfg.outputTracks = isEnabled(outputTypes, ioType::Tracks);
+  cfg.outputCompClusters = isEnabled(outputTypes, ioType::CompClustROOT);
+  cfg.outputCompClustersFlat = isEnabled(outputTypes, ioType::CompClustCTF);
+  cfg.outputCAClusters = isEnabled(outputTypes, ioType::Clusters);
+  cfg.outputQA = isEnabled(outputTypes, ioType::QA);
+  cfg.outputSharedClusterMap = (cfg.outputCAClusters || cfg.caClusterer || isEnabled(inputTypes, ioType::Clusters)) && cfg.outputTracks && !isEnabled(outputTypes, ioType::NoSharedMap);
+  cfg.processMC = doMC;
+  cfg.sendClustersPerSector = false;
+  cfg.askDISTSTF = !cfgc.options().get<bool>("ignore-dist-stf");
+  specs.emplace_back(o2::gpu::getGPURecoWorkflowSpec(&gPolicyData, cfg, tpcSectors, gTpcSectorMask, "gpu-reconstruction"));
+
+  if (!cfgc.options().get<bool>("ignore-dist-stf")) {
+    GlobalTrackID::mask_t srcTrk = GlobalTrackID::getSourcesMask("none");
+    GlobalTrackID::mask_t srcCl = GlobalTrackID::getSourcesMask("TPC");
+    o2::globaltracking::InputHelper::addInputSpecs(cfgc, specs, srcCl, srcTrk, srcTrk, doMC);
+  }
+
+  // configure dpl timer to inject correct firstTFOrbit: start from the 1st orbit of TF containing 1st sampled orbit
+  o2::raw::HBFUtilsInitializer hbfIni(cfgc, specs);
+
+  return std::move(specs);
+}
