gpu-reco-workflow.cxx

Go to the documentation of this file.

// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
// All rights not expressly granted are reserved.
//
// This software is distributed under the terms of the GNU General Public
// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
//
// 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.
 
 
#include "Framework/WorkflowSpec.h"
#include "Framework/ConfigParamSpec.h"
#include "Framework/CompletionPolicyHelpers.h"
#include "Framework/DispatchPolicy.h"
#include "Framework/ConcreteDataMatcher.h"
#include "TPCReaderWorkflow/TPCSectorCompletionPolicy.h"
#include "Framework/CustomWorkflowTerminationHook.h"
#include "GPUWorkflow/GPUWorkflowSpec.h"
#include "CommonUtils/ConfigurableParam.h"
#include "DetectorsRaw/HBFUtilsInitializer.h"
#include "DetectorsBase/GRPGeomHelper.h"
#include "Framework/CallbacksPolicy.h"
#include "TPCBase/Sector.h"
#include "Algorithm/RangeTokenizer.h"
#include "GlobalTrackingWorkflowHelpers/InputHelper.h"
#include "ReconstructionDataFormats/GlobalTrackID.h"
#include "TPCCalibration/CorrectionMapsLoader.h"
 
#include <unordered_map>
#include <numeric>
 
using namespace o2::framework;
using namespace o2::dataformats;
using namespace o2::gpu;
using CompletionPolicyData = std::vector<InputSpec>;
static CompletionPolicyData gPolicyData;
static constexpr uint64_t gTpcSectorMask = 0xFFFFFFFFF;
static std::function<bool(o2::framework::DataProcessingHeader::StartTime)>* gPolicyOrderCheck;
static std::shared_ptr<GPURecoWorkflowSpec> gTask;
 
void customize(std::vector<o2::framework::CallbacksPolicy>& policies)
{
  o2::raw::HBFUtilsInitializer::addNewTimeSliceCallback(policies);
}
 
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, trd-tracklets, its-clusters"}},
    {"output-type", VariantType::String, "tracks", {"clustersnative, tracks, compressed-clusters-ctf, qa, no-shared-cluster-map, send-clusters-per-sector, trd-tracks, error-qa, tpc-triggers, its-tracks"}},
    {"corrmap-lumi-mode", VariantType::Int, 0, {"scaling mode: (default) 0 = static + scale * full; 1 = full + scale * derivative"}},
    {"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"}},
    {"enableDoublePipeline", VariantType::Bool, false, {"enable GPU double pipeline mode"}},
    {"tpc-deadMap-sources", VariantType::Int, -1, {"Sources to consider for TPC dead channel map creation; -1=all, 0=deactivated"}},
    {"tpc-mc-time-gain", VariantType::Bool, false, {"use time gain calibration for MC (true) or for data (false)"}},
  };
  o2::tpc::CorrectionMapsLoader::addGlobalOptions(options);
  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, &gPolicyOrderCheck)());
}
 
void customize(o2::framework::OnWorkflowTerminationHook& hook)
{
  hook = [](const char* idstring) {
    if (gTask) {
      gTask->deinitialize();
    }
  };
}
 
#include "Framework/runDataProcessing.h" // the main driver
 
using namespace o2::framework;
 
enum struct ioType { Digits,
                     Clusters,
                     ZSRaw,
                     ZSRawOTF,
                     CompClustROOT,
                     CompClustCTF,
                     Tracks,
                     QA,
                     ErrorQA,
                     TRDTracklets,
                     TRDTracks,
                     NoSharedMap,
                     SendClustersPerSector,
                     ITSClusters,
                     ITSTracks,
                     MeanVertex,
                     TPCTriggers };
 
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},
  {"trd-tracklets", ioType::TRDTracklets},
  {"its-clusters", ioType::ITSClusters},
  {"its-mean-vertex", ioType::MeanVertex},
};
 
static const std::unordered_map<std::string, ioType> OutputMap{
  {"clusters", ioType::Clusters},
  {"tracks", ioType::Tracks},
  {"compressed-clusters-ctf", ioType::CompClustCTF},
  {"qa", ioType::QA},
  {"error-qa", ioType::ErrorQA},
  {"no-shared-cluster-map", ioType::NoSharedMap},
  {"send-clusters-per-sector", ioType::SendClustersPerSector},
  {"trd-tracks", ioType::TRDTracks},
  {"its-tracks", ioType::ITSTracks},
  {"tpc-triggers", ioType::TPCTriggers}};
 
WorkflowSpec defineDataProcessing(ConfigContext const& cfgc)
{
  WorkflowSpec specs;
  std::vector<int32_t> 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");
  auto sclOpt = o2::tpc::CorrectionMapsLoader::parseGlobalOptions(cfgc.options());
  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();
  };
 
  GPURecoWorkflowSpec::Config cfg;
  cfg.runTPCTracking = true;
  cfg.lumiScaleType = sclOpt.lumiType;
  cfg.lumiScaleMode = sclOpt.lumiMode;
  cfg.enableMShape = sclOpt.enableMShapeCorrection;
  cfg.enableCTPLumi = sclOpt.requestCTPLumi;
  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.outputErrorQA = isEnabled(outputTypes, ioType::ErrorQA);
  cfg.outputSharedClusterMap = (cfg.outputCAClusters || cfg.caClusterer || isEnabled(inputTypes, ioType::Clusters)) && cfg.outputTracks && !isEnabled(outputTypes, ioType::NoSharedMap);
  cfg.processMC = doMC;
  cfg.sendClustersPerSector = isEnabled(outputTypes, ioType::SendClustersPerSector);
  cfg.askDISTSTF = !cfgc.options().get<bool>("ignore-dist-stf");
  cfg.readTRDtracklets = isEnabled(inputTypes, ioType::TRDTracklets);
  cfg.runTRDTracking = isEnabled(outputTypes, ioType::TRDTracks);
  cfg.tpcTriggerHandling = isEnabled(outputTypes, ioType::TPCTriggers) || cfg.caClusterer;
  cfg.enableDoublePipeline = cfgc.options().get<bool>("enableDoublePipeline");
  cfg.tpcDeadMapSources = cfgc.options().get<int32_t>("tpc-deadMap-sources");
  cfg.tpcUseMCTimeGain = cfgc.options().get<bool>("tpc-mc-time-gain");
  cfg.runITSTracking = isEnabled(outputTypes, ioType::ITSTracks);
  cfg.itsOverrBeamEst = isEnabled(inputTypes, ioType::MeanVertex);
 
  Inputs ggInputs;
  auto ggRequest = std::make_shared<o2::base::GRPGeomRequest>(false, true, false, true, true, o2::base::GRPGeomRequest::Aligned, ggInputs, true);
 
  auto task = std::make_shared<GPURecoWorkflowSpec>(&gPolicyData, cfg, tpcSectors, gTpcSectorMask, ggRequest, &gPolicyOrderCheck);
  Inputs taskInputs = task->inputs();
  std::move(ggInputs.begin(), ggInputs.end(), std::back_inserter(taskInputs));
  gTask = task;
 
  specs.emplace_back(DataProcessorSpec{
    "gpu-reconstruction",
    taskInputs,
    task->outputs(),
    AlgorithmSpec{adoptTask<GPURecoWorkflowSpec>(task)},
    task->options()});
 
  if (cfg.enableDoublePipeline) {
    cfg.enableDoublePipeline = 2;
    Inputs ggInputsPrepare;
    auto ggRequestPrepare = std::make_shared<o2::base::GRPGeomRequest>(false, true, false, false, false, o2::base::GRPGeomRequest::None, ggInputsPrepare, true);
    auto taskPrepare = std::make_shared<GPURecoWorkflowSpec>(&gPolicyData, cfg, tpcSectors, gTpcSectorMask, ggRequestPrepare);
    Inputs taskInputsPrepare = taskPrepare->inputs();
    std::move(ggInputsPrepare.begin(), ggInputsPrepare.end(), std::back_inserter(taskInputsPrepare));
    specs.emplace_back(DataProcessorSpec{
      "gpu-reconstruction-prepare",
      taskInputsPrepare,
      taskPrepare->outputs(),
      AlgorithmSpec{adoptTask<GPURecoWorkflowSpec>(taskPrepare)},
      taskPrepare->options()});
  }
 
  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 specs;
}