diff --git a/DataFormats/Detectors/TPC/include/DataFormatsTPC/ClusterNativeHelper.h b/DataFormats/Detectors/TPC/include/DataFormatsTPC/ClusterNativeHelper.h index 73d72ee6ddc2b..f35141f4bf849 100644 --- a/DataFormats/Detectors/TPC/include/DataFormatsTPC/ClusterNativeHelper.h +++ b/DataFormats/Detectors/TPC/include/DataFormatsTPC/ClusterNativeHelper.h @@ -178,17 +178,16 @@ class ClusterNativeHelper static int fillIndex( ClusterNativeAccess& clusterIndex, std::unique_ptr& clusterBuffer, MCLabelContainer& mcBuffer, DataArrayType& inputs, MCArrayType const& mcinputs, - CheckFct checkFct = [](auto&) { return true; }); + CheckFct checkFct = [](auto const&) { return true; }); - template + template > static int fillIndex( ClusterNativeAccess& clusterIndex, std::unique_ptr& clusterBuffer, - DataArrayType& inputs, CheckFct checkFct = [](auto&) { return true; }) + DataArrayType& inputs, CheckFct checkFct = [](auto const&) { return true; }) { - // just use a dummy array of zero-size containers as default with the same extent - // as the data container collection - // TODO: maybe do in one function with conditional template parameter - std::array, std::tuple_size::value> dummy; + // just use a dummy parameter with empty vectors, the ugly double vector will be removed + // soon. TODO: maybe do in one function with conditional template parameter + std::vector> dummy(inputs.size()); // another default, nothing will be added to the container MCLabelContainer mcBuffer; return fillIndex(clusterIndex, clusterBuffer, mcBuffer, inputs, dummy, checkFct); @@ -313,11 +312,13 @@ int ClusterNativeHelper::Reader::fillIndex(ClusterNativeAccess& clusterIndex, std::unique_ptr& clusterBuffer, MCLabelContainer& mcBuffer, DataArrayType& inputs, MCArrayType const& mcinputs, CheckFct checkFct) { - static_assert(std::tuple_size::value == std::tuple_size::value); + if (mcinputs.size() > 0 && mcinputs.size() != inputs.size()) { + std::runtime_error("inconsistent size of MC label array " + std::to_string(mcinputs.size()) + ", expected " + std::to_string(inputs.size())); + } memset(&clusterIndex, 0, sizeof(clusterIndex)); const MCLabelContainer* clustersMCTruth[NSectors][NPadRows] = {}; int result = 0; - for (size_t index = 0; index < NSectors; index++) { + for (size_t index = 0, end = inputs.size(); index < end; index++) { if (!checkFct(index)) { continue; } diff --git a/Detectors/GlobalTrackingWorkflow/include/GlobalTrackingWorkflow/TPCITSMatchingSpec.h b/Detectors/GlobalTrackingWorkflow/include/GlobalTrackingWorkflow/TPCITSMatchingSpec.h index 5b6a84513592e..a33aa93159abe 100644 --- a/Detectors/GlobalTrackingWorkflow/include/GlobalTrackingWorkflow/TPCITSMatchingSpec.h +++ b/Detectors/GlobalTrackingWorkflow/include/GlobalTrackingWorkflow/TPCITSMatchingSpec.h @@ -13,8 +13,6 @@ #ifndef O2_MATCHING_TPCITS_SPEC #define O2_MATCHING_TPCITS_SPEC -#include "TFile.h" - #include "GlobalTracking/MatchTPCITS.h" #include "DataFormatsITSMFT/TopologyDictionary.h" #include "DataFormatsTPC/Constants.h" @@ -34,8 +32,8 @@ namespace globaltracking class TPCITSMatchingDPL : public Task { public: - TPCITSMatchingDPL(bool useMC, const std::vector& tpcClusLanes) - : mUseMC(useMC), mTPCClusLanes(tpcClusLanes) {} + TPCITSMatchingDPL(bool useMC) + : mUseMC(useMC) {} ~TPCITSMatchingDPL() override = default; void init(InitContext& ic) final; void run(ProcessingContext& pc) final; @@ -44,8 +42,6 @@ class TPCITSMatchingDPL : public Task private: o2::globaltracking::MatchTPCITS mMatching; // matching engine o2::itsmft::TopologyDictionary mITSDict; // cluster patterns dictionary - std::vector mTPCClusLanes; - std::array, o2::tpc::Constants::MAXSECTOR> mBufferedTPCClusters; // at the moment not used bool mUseMC = true; TStopwatch mTimer; }; diff --git a/Detectors/GlobalTrackingWorkflow/src/TPCITSMatchingSpec.cxx b/Detectors/GlobalTrackingWorkflow/src/TPCITSMatchingSpec.cxx index 55ab4565c3ee4..b3db0a0bc7f18 100644 --- a/Detectors/GlobalTrackingWorkflow/src/TPCITSMatchingSpec.cxx +++ b/Detectors/GlobalTrackingWorkflow/src/TPCITSMatchingSpec.cxx @@ -12,11 +12,8 @@ #include -#include "TTree.h" -#include - -#include "Framework/ControlService.h" #include "Framework/ConfigParamRegistry.h" +#include "Framework/InputRecordWalker.h" #include "GlobalTrackingWorkflow/TPCITSMatchingSpec.h" #include "ReconstructionDataFormats/TrackTPCITS.h" #include "SimulationDataFormat/MCCompLabel.h" @@ -85,10 +82,10 @@ void TPCITSMatchingDPL::run(ProcessingContext& pc) uint64_t activeSectors = 0; std::bitset validSectors = 0; std::map datarefs; - for (auto const& lane : mTPCClusLanes) { - std::string inputLabel = "clusTPC" + std::to_string(lane); - LOG(INFO) << "Reading lane " << lane << " " << inputLabel; - auto ref = pc.inputs().get(inputLabel); + std::vector filter = { + {"check", ConcreteDataTypeMatcher{"TPC", "CLUSTERNATIVE"}, Lifetime::Timeframe}, + }; + for (auto const& ref : InputRecordWalker(pc.inputs(), filter)) { auto const* sectorHeader = DataRefUtils::getHeader(ref); if (sectorHeader == nullptr) { // FIXME: think about error policy @@ -102,6 +99,7 @@ void TPCITSMatchingDPL::run(ProcessingContext& pc) if (sector >= o2::tpc::TPCSectorHeader::NSectors) { throw std::runtime_error("Expecting data for single sectors"); } + LOG(INFO) << "Reading cluster data for sector " << sector; if (validSectors.test(sector)) { // have already data for this sector, this should not happen in the current // sequential implementation, for parallel path merged at the tracker stage @@ -126,42 +124,16 @@ void TPCITSMatchingDPL::run(ProcessingContext& pc) // not all sectors available // Since we expect complete input, this should not happen (why does the bufferization considered for TPC CA tracker? Ask Matthias) throw std::runtime_error("Did not receive TPC clusters data for all sectors"); - /* - for (auto const& refentry : datarefs) { - auto& sector = refentry.first; - auto& ref = refentry.second; - auto payploadSize = DataRefUtils::getPayloadSize(ref); - mBufferedTPCClusters[sector].resize(payploadSize); - std::copy(ref.payload, ref.payload + payploadSize, mBufferedTPCClusters[sector].begin()); - - printInputLog(ref, "buffering", sector); - } - // not needed to send something, DPL will simply drop this timeslice, whenever the - // data for all sectors is available, the output is sent in that time slice - return; - */ } //------------------------------------------------------------------------------ - std::array, o2::tpc::Constants::MAXSECTOR> mcInputs; // DUMMY - std::array, o2::tpc::Constants::MAXSECTOR> clustersTPC; - auto sectorStatus = validSectors; + std::vector> clustersTPC; for (auto const& refentry : datarefs) { auto& sector = refentry.first; auto& ref = refentry.second; - clustersTPC[sector] = gsl::span(ref.payload, DataRefUtils::getPayloadSize(ref)); - sectorStatus.reset(sector); + clustersTPC.emplace_back(ref.payload, DataRefUtils::getPayloadSize(ref)); printInputLog(ref, "received", sector); } - if (sectorStatus.any()) { - LOG(ERROR) << "Reading bufferized TPC clusters, this should not happen"; - // some of the inputs have been buffered - for (size_t sector = 0; sector < sectorStatus.size(); ++sector) { - if (sectorStatus.test(sector)) { - clustersTPC[sector] = gsl::span(&mBufferedTPCClusters[sector].front(), mBufferedTPCClusters[sector].size()); - } - } - } // Just print TPC clusters status { @@ -203,9 +175,8 @@ void TPCITSMatchingDPL::run(ProcessingContext& pc) o2::tpc::ClusterNativeAccess clusterIndex; std::unique_ptr clusterBuffer; - o2::tpc::MCLabelContainer clusterMCBuffer; memset(&clusterIndex, 0, sizeof(clusterIndex)); - o2::tpc::ClusterNativeHelper::Reader::fillIndex(clusterIndex, clusterBuffer, clusterMCBuffer, clustersTPC, mcInputs, [&validSectors](auto& index) { return validSectors.test(index); }); + o2::tpc::ClusterNativeHelper::Reader::fillIndex(clusterIndex, clusterBuffer, clustersTPC); //----------------------------<< TPC Clusters loading <<------------------------------------------ // @@ -260,12 +231,6 @@ void TPCITSMatchingDPL::run(ProcessingContext& pc) mMatching.run(); - /* // at the moment we don't assume need for bufferization, no nead to clear - for (auto& secClBuf : mBufferedTPCClusters) { - secClBuf.clear(); - } - */ - pc.outputs().snapshot(Output{"GLO", "TPCITS", 0, Lifetime::Timeframe}, mMatching.getMatchedTracks()); if (mUseMC) { pc.outputs().snapshot(Output{"GLO", "TPCITS_ITSMC", 0, Lifetime::Timeframe}, mMatching.getMatchedITSLabels()); @@ -294,10 +259,7 @@ DataProcessorSpec getTPCITSMatchingSpec(bool useMC, const std::vector& tpcC inputs.emplace_back("trackTPC", "TPC", "TRACKS", 0, Lifetime::Timeframe); inputs.emplace_back("trackTPCClRefs", "TPC", "CLUSREFS", 0, Lifetime::Timeframe); - for (auto lane : tpcClusLanes) { - std::string clusBind = "clusTPC" + std::to_string(lane); - inputs.emplace_back(clusBind.c_str(), "TPC", "CLUSTERNATIVE", lane, Lifetime::Timeframe); - } + inputs.emplace_back("clusTPC", ConcreteDataTypeMatcher{"TPC", "CLUSTERNATIVE"}, Lifetime::Timeframe); if (o2::globaltracking::MatchITSTPCParams::Instance().runAfterBurner) { inputs.emplace_back("fitInfo", "FT0", "RECPOINTS", 0, Lifetime::Timeframe); @@ -318,7 +280,7 @@ DataProcessorSpec getTPCITSMatchingSpec(bool useMC, const std::vector& tpcC "itstpc-track-matcher", inputs, outputs, - AlgorithmSpec{adaptFromTask(useMC, tpcClusLanes)}, + AlgorithmSpec{adaptFromTask(useMC)}, Options{{"its-dictionary-path", VariantType::String, "", {"Path of the cluster-topology dictionary file"}}}}; } diff --git a/Detectors/GlobalTrackingWorkflow/src/tpcits-match-workflow.cxx b/Detectors/GlobalTrackingWorkflow/src/tpcits-match-workflow.cxx index 6ca2b4ba17245..9562be20529df 100644 --- a/Detectors/GlobalTrackingWorkflow/src/tpcits-match-workflow.cxx +++ b/Detectors/GlobalTrackingWorkflow/src/tpcits-match-workflow.cxx @@ -10,6 +10,8 @@ #include "GlobalTrackingWorkflow/MatchTPCITSWorkflow.h" #include "CommonUtils/ConfigurableParam.h" +#include "Framework/CompletionPolicy.h" +#include "TPCWorkflow/TPCSectorCompletionPolicy.h" using namespace o2::framework; @@ -28,6 +30,17 @@ void customize(std::vector& workflowOptions) std::swap(workflowOptions, options); } +// the matcher process requires the TPC sector completion to trigger and data on +// all defined routes +void customize(std::vector& policies) +{ + // the TPC sector completion policy checks when the set of TPC/CLUSTERNATIVE data is complete + // in addition we require to have input from all other routes + policies.push_back(o2::tpc::TPCSectorCompletionPolicy("itstpc-track-matcher", + o2::tpc::TPCSectorCompletionPolicy::Config::RequireAll, + InputSpec{"cluster", o2::framework::ConcreteDataTypeMatcher{"TPC", "CLUSTERNATIVE"}})()); +} + // ------------------------------------------------------------------ #include "Framework/runDataProcessing.h" diff --git a/Detectors/GlobalTrackingWorkflow/tpcinterpolationworkflow/include/TPCInterpolationWorkflow/TPCInterpolationSpec.h b/Detectors/GlobalTrackingWorkflow/tpcinterpolationworkflow/include/TPCInterpolationWorkflow/TPCInterpolationSpec.h index bdfab1fac372b..08cae7bfb464c 100644 --- a/Detectors/GlobalTrackingWorkflow/tpcinterpolationworkflow/include/TPCInterpolationWorkflow/TPCInterpolationSpec.h +++ b/Detectors/GlobalTrackingWorkflow/tpcinterpolationworkflow/include/TPCInterpolationWorkflow/TPCInterpolationSpec.h @@ -28,7 +28,7 @@ namespace tpc class TPCInterpolationDPL : public Task { public: - TPCInterpolationDPL(bool useMC, const std::vector& tpcClusLanes) : mUseMC(useMC), mTPCClusLanes(tpcClusLanes) {} + TPCInterpolationDPL(bool useMC) : mUseMC(useMC) {} ~TPCInterpolationDPL() override = default; void init(InitContext& ic) final; void run(ProcessingContext& pc) final; @@ -36,8 +36,6 @@ class TPCInterpolationDPL : public Task private: o2::tpc::TrackInterpolation mInterpolation; // track interpolation engine - std::vector mTPCClusLanes; - std::array, o2::tpc::Constants::MAXSECTOR> mBufferedTPCClusters; bool mUseMC{false}; ///< MC flag TStopwatch mTimer; }; diff --git a/Detectors/GlobalTrackingWorkflow/tpcinterpolationworkflow/src/TPCInterpolationSpec.cxx b/Detectors/GlobalTrackingWorkflow/tpcinterpolationworkflow/src/TPCInterpolationSpec.cxx index 46923f1c5186d..df5816fff1ec7 100644 --- a/Detectors/GlobalTrackingWorkflow/tpcinterpolationworkflow/src/TPCInterpolationSpec.cxx +++ b/Detectors/GlobalTrackingWorkflow/tpcinterpolationworkflow/src/TPCInterpolationSpec.cxx @@ -12,7 +12,7 @@ #include -#include "Framework/ControlService.h" +#include "Framework/InputRecordWalker.h" #include "DataFormatsITS/TrackITS.h" #include "ReconstructionDataFormats/TrackTPCITS.h" #include "DataFormatsTPC/TrackTPC.h" @@ -56,10 +56,10 @@ void TPCInterpolationDPL::run(ProcessingContext& pc) uint64_t activeSectors = 0; std::bitset validSectors = 0; std::map datarefs; - for (auto const& lane : mTPCClusLanes) { - std::string inputLabel = "clusTPC" + std::to_string(lane); - LOG(INFO) << "Reading lane " << lane << " " << inputLabel; - auto ref = pc.inputs().get(inputLabel); + std::vector filter = { + {"check", ConcreteDataTypeMatcher{"TPC", "CLUSTERNATIVE"}, Lifetime::Timeframe}, + }; + for (auto const& ref : InputRecordWalker(pc.inputs(), filter)) { auto const* sectorHeader = DataRefUtils::getHeader(ref); if (sectorHeader == nullptr) { // FIXME: think about error policy @@ -73,6 +73,7 @@ void TPCInterpolationDPL::run(ProcessingContext& pc) if (sector >= o2::tpc::TPCSectorHeader::NSectors) { throw std::runtime_error("Expecting data for single sectors"); } + LOG(INFO) << "Reading cluster data for sector " << sector; if (validSectors.test(sector)) { // have already data for this sector, this should not happen in the current // sequential implementation, for parallel path merged at the tracker stage @@ -97,42 +98,16 @@ void TPCInterpolationDPL::run(ProcessingContext& pc) // not all sectors available // Since we expect complete input, this should not happen (why does the bufferization considered for TPC CA tracker? Ask Matthias) throw std::runtime_error("Did not receive TPC clusters data for all sectors"); - /* - for (auto const& refentry : datarefs) { - auto& sector = refentry.first; - auto& ref = refentry.second; - auto payploadSize = DataRefUtils::getPayloadSize(ref); - mBufferedTPCClusters[sector].resize(payploadSize); - std::copy(ref.payload, ref.payload + payploadSize, mBufferedTPCClusters[sector].begin()); - - printInputLog(ref, "buffering", sector); - } - // not needed to send something, DPL will simply drop this timeslice, whenever the - // data for all sectors is available, the output is sent in that time slice - return; - */ } //------------------------------------------------------------------------------ - std::array, o2::tpc::Constants::MAXSECTOR> mcInputs; // DUMMY - std::array, o2::tpc::Constants::MAXSECTOR> clustersTPC; - auto sectorStatus = validSectors; + std::vector> clustersTPC; for (auto const& refentry : datarefs) { auto& sector = refentry.first; auto& ref = refentry.second; - clustersTPC[sector] = gsl::span(ref.payload, DataRefUtils::getPayloadSize(ref)); - sectorStatus.reset(sector); + clustersTPC.emplace_back(ref.payload, DataRefUtils::getPayloadSize(ref)); printInputLog(ref, "received", sector); } - if (sectorStatus.any()) { - LOG(ERROR) << "Reading bufferized TPC clusters, this should not happen"; - // some of the inputs have been buffered - for (size_t sector = 0; sector < sectorStatus.size(); ++sector) { - if (sectorStatus.test(sector)) { - clustersTPC[sector] = gsl::span(&mBufferedTPCClusters[sector].front(), mBufferedTPCClusters[sector].size()); - } - } - } // Just print TPC clusters status { @@ -174,9 +149,8 @@ void TPCInterpolationDPL::run(ProcessingContext& pc) o2::tpc::ClusterNativeAccess clusterIndex; std::unique_ptr clusterBuffer; - o2::tpc::MCLabelContainer clusterMCBuffer; memset(&clusterIndex, 0, sizeof(clusterIndex)); - o2::tpc::ClusterNativeHelper::Reader::fillIndex(clusterIndex, clusterBuffer, clusterMCBuffer, clustersTPC, mcInputs, [&validSectors](auto& index) { return validSectors.test(index); }); + o2::tpc::ClusterNativeHelper::Reader::fillIndex(clusterIndex, clusterBuffer, clustersTPC); //----------------------------<< TPC Clusters loading <<------------------------------------------ // pass input data to TrackInterpolation object @@ -217,10 +191,7 @@ DataProcessorSpec getTPCInterpolationSpec(bool useMC, const std::vector& tp inputs.emplace_back("trackTPC", "TPC", "TRACKS", 0, Lifetime::Timeframe); inputs.emplace_back("trackTPCClRefs", "TPC", "CLUSREFS", 0, Lifetime::Timeframe); - for (auto lane : tpcClusLanes) { - std::string clusBind = "clusTPC" + std::to_string(lane); - inputs.emplace_back(clusBind.c_str(), "TPC", "CLUSTERNATIVE", lane, Lifetime::Timeframe); - } + inputs.emplace_back("clusTPC", ConcreteDataTypeMatcher{"TPC", "CLUSTERNATIVE"}, Lifetime::Timeframe); inputs.emplace_back("match", "GLO", "TPCITS", 0, Lifetime::Timeframe); inputs.emplace_back("matchTOF", "TOF", "MATCHINFOS", 0, Lifetime::Timeframe); @@ -243,7 +214,7 @@ DataProcessorSpec getTPCInterpolationSpec(bool useMC, const std::vector& tp "tpc-track-interpolation", inputs, outputs, - AlgorithmSpec{adaptFromTask(useMC, tpcClusLanes)}, + AlgorithmSpec{adaptFromTask(useMC)}, Options{}}; } diff --git a/Detectors/GlobalTrackingWorkflow/tpcinterpolationworkflow/src/tpc-interpolation-workflow.cxx b/Detectors/GlobalTrackingWorkflow/tpcinterpolationworkflow/src/tpc-interpolation-workflow.cxx index c4524ad5f2dc6..059eae8df5d8f 100644 --- a/Detectors/GlobalTrackingWorkflow/tpcinterpolationworkflow/src/tpc-interpolation-workflow.cxx +++ b/Detectors/GlobalTrackingWorkflow/tpcinterpolationworkflow/src/tpc-interpolation-workflow.cxx @@ -10,6 +10,8 @@ #include "TPCInterpolationWorkflow/TrackInterpolationWorkflow.h" #include "CommonUtils/ConfigurableParam.h" +#include "Framework/CompletionPolicy.h" +#include "TPCWorkflow/TPCSectorCompletionPolicy.h" using namespace o2::framework; @@ -27,6 +29,17 @@ void customize(std::vector& workflowOptions) workflowOptions.push_back(ConfigParamSpec{"configKeyValues", VariantType::String, "", {keyvaluehelp}}); } +// the matcher process requires the TPC sector completion to trigger and data on +// all defined routes +void customize(std::vector& policies) +{ + // the TPC sector completion policy checks when the set of TPC/CLUSTERNATIVE data is complete + // in addition we require to have input from all other routes + policies.push_back(o2::tpc::TPCSectorCompletionPolicy("tpc-track-interpolation", + o2::tpc::TPCSectorCompletionPolicy::Config::RequireAll, + InputSpec{"cluster", o2::framework::ConcreteDataTypeMatcher{"TPC", "CLUSTERNATIVE"}})()); +} + // ------------------------------------------------------------------ #include "Framework/runDataProcessing.h" diff --git a/Detectors/TPC/workflow/include/TPCWorkflow/TPCSectorCompletionPolicy.h b/Detectors/TPC/workflow/include/TPCWorkflow/TPCSectorCompletionPolicy.h index 05ef0c9113260..a44b334cc33a5 100644 --- a/Detectors/TPC/workflow/include/TPCWorkflow/TPCSectorCompletionPolicy.h +++ b/Detectors/TPC/workflow/include/TPCWorkflow/TPCSectorCompletionPolicy.h @@ -45,17 +45,29 @@ namespace tpc /// stack describing the current sector and defining the active sectors in the setup. The completion /// policy callback will wait until there is data for all active sectors. /// +/// If config flag ConFig::RequireAll is specified in the constructor parameters, data from all inputs +/// will be required in addition to the matching TPC sector policy. With this flag, the policy can be +/// used for processors with TPC input and other inputs, without checking for complex multimessages on +/// the other inputs. +/// /// Parameters: /// processor name rexexp to match a name of the processor for which the policy should be applied /// input matchers provided as an argument pack /// Note: it is important to use ConcreteDataTypeMatcher to define input spec with /// wildcard on subSpecification +/// config param Parameters like Config::RequireAll /// Usage: -/// TPCSectorCompletionPolicy("processor-name-regexp", InputSpec{"", ConcreteDataTypeMatcher{"DET", "RAWDATA"}}, ...)(); +/// TPCSectorCompletionPolicy("processor-name-regexp", +/// TPCSectorCompletionPolicy::Config::RequireAll, +/// InputSpec{"", ConcreteDataTypeMatcher{"DET", "RAWDATA"}}, ...)(); /// class TPCSectorCompletionPolicy { public: + enum struct Config { + // require data on all other inputs in addition to the ones checked for the sector completion + RequireAll, + }; TPCSectorCompletionPolicy() = delete; template TPCSectorCompletionPolicy(const char* processorName, Args&&... args) @@ -72,7 +84,7 @@ class TPCSectorCompletionPolicy return std::regex_match(device.name.begin(), device.name.end(), std::regex(expression.c_str())); }; - auto callback = [inputMatchers = mInputMatchers](CompletionPolicy::InputSet inputs) -> CompletionPolicy::CompletionOp { + auto callback = [inputMatchers = mInputMatchers, bRequireAll = this->mRequireAll](CompletionPolicy::InputSet inputs) -> CompletionPolicy::CompletionOp { auto op = CompletionPolicy::CompletionOp::Wait; std::bitset validSectors = 0; bool haveMatchedInput = false; @@ -123,10 +135,17 @@ class TPCSectorCompletionPolicy } } - if (haveMatchedInput && activeSectors == validSectors.to_ulong()) { + // If the flag Config::RequireAll is set in the constructor arguments we require + // data from all inputs in addition to the sector matching condition + // To be fully correct we would need to require data from all inputs not going + // into the TPC policy, but that is not possible for the moment. That's why there is a possibly + // unhandled case if multiple TPC input routes are defined but a complete data set is coming over + // one of them. Not likely to be a use case, though. + if (haveMatchedInput && activeSectors == validSectors.to_ulong() && + (!bRequireAll || nActiveInputRoutes == inputs.size())) { // we can process if there is input for all sectors, the required sectors are // transported as part of the sector header - op = CompletionPolicy::CompletionOp::Process; + op = CompletionPolicy::CompletionOp::Consume; } else if (activeSectors == 0 && nActiveInputRoutes == inputs.size()) { // no sector header is transmitted, this is the case for e.g. the ZS raw data // we simply require input on all routes, this is also the default of DPL DataRelayer @@ -138,7 +157,7 @@ class TPCSectorCompletionPolicy //if (nMaxPartsPerRoute > 1) { // LOG(WARNING) << "No sector information is provided with the data, data set is complete with data on all input routes. But there are multiple parts on at least one route and this policy might not be complete, no check possible if other parts on some routes are still missing. It is adviced to add a custom policy."; //} - op = CompletionPolicy::CompletionOp::Process; + op = CompletionPolicy::CompletionOp::Consume; } return op; @@ -148,10 +167,21 @@ class TPCSectorCompletionPolicy private: /// recursively init list of input routes from parameter pack - template - void init(InputSpec&& spec, Args&&... args) + template + void init(Arg&& arg, Args&&... args) { - mInputMatchers.emplace_back(std::move(spec)); + using Type = std::decay_t; + if constexpr (std::is_same::value) { + mInputMatchers.emplace_back(std::move(arg)); + } else if constexpr (std::is_same::value) { + switch (arg) { + case Config::RequireAll: + mRequireAll = true; + break; + } + } else { + static_assert(always_static_assert_v); + } if constexpr (sizeof...(args) > 0) { init(std::forward(args)...); } @@ -159,6 +189,7 @@ class TPCSectorCompletionPolicy std::string mProcessorName; std::vector mInputMatchers; + bool mRequireAll = false; }; } // namespace tpc } // namespace o2