From f6bd2260b1faea5b32a4ec9f85ced4acf9ac7fce Mon Sep 17 00:00:00 2001 From: Giulio Eulisse <10544+ktf@users.noreply.github.com> Date: Wed, 24 Jun 2026 15:04:09 +0200 Subject: [PATCH 1/6] DPL: add support for writing general Metadata to AO2D --- .../AnalysisSupport/src/AODWriterHelpers.cxx | 47 ++++++++++++++++ .../AnalysisSupport/src/AODWriterHelpers.h | 1 + Framework/AnalysisSupport/src/Plugin.cxx | 8 +++ .../Framework/AnalysisSupportHelpers.h | 3 + Framework/Core/src/AnalysisSupportHelpers.cxx | 19 +++++++ Framework/Core/src/CompletionPolicy.cxx | 1 + Framework/Core/src/WorkflowHelpers.cxx | 9 +++ .../TestWorkflows/src/o2TestHistograms.cxx | 55 +++++++++++++------ 8 files changed, 125 insertions(+), 18 deletions(-) diff --git a/Framework/AnalysisSupport/src/AODWriterHelpers.cxx b/Framework/AnalysisSupport/src/AODWriterHelpers.cxx index 5b5829d96a1de..cd7aaf55abf76 100644 --- a/Framework/AnalysisSupport/src/AODWriterHelpers.cxx +++ b/Framework/AnalysisSupport/src/AODWriterHelpers.cxx @@ -16,6 +16,7 @@ #include "Framework/EndOfStreamContext.h" #include "Framework/ProcessingContext.h" #include "Framework/InitContext.h" +#include "Framework/Output.h" #include "Framework/CallbackService.h" #include "Framework/AnalysisSupportHelpers.h" #include "Framework/TableConsumer.h" @@ -31,6 +32,10 @@ #include #include #include +#include +#include +#include +#include O2_DECLARE_DYNAMIC_LOG(histogram_registry); @@ -477,4 +482,46 @@ AlgorithmSpec AODWriterHelpers::getOutputObjHistWriter(ConfigContext const& /*ct }; }}; } + +AlgorithmSpec AODWriterHelpers::getMetadataCollector(ConfigContext const& /*ctx*/) +{ + return AlgorithmSpec{[](InitContext&) -> std::function { + // Accumulated metadata, last writer wins per key. + auto merged = std::make_shared>>(); + return [merged](ProcessingContext& pc) -> void { + auto nParts = pc.inputs().getNofParts(0); + for (auto pi = 0U; pi < nParts; ++pi) { + auto part = pc.inputs().get("meta", pi); + if (!part) { + continue; + } + TIter next(part.get()); + while (TObject* key = next()) { + TObject* value = part->GetValue(key); + std::string k = key->GetName(); + std::string v = value != nullptr ? value->GetName() : ""; + auto it = std::find_if(merged->begin(), merged->end(), + [&k](auto const& e) { return e.first == k; }); + if (it != merged->end()) { + it->second = std::move(v); + } else { + merged->emplace_back(std::move(k), std::move(v)); + } + } + } + // Emit the keys/vals vectors the AOD writer already turns into the AO2D + // metaData TMap, so no special handling is needed there. + std::vector keys, vals; + keys.reserve(merged->size()); + vals.reserve(merged->size()); + for (auto const& [k, v] : *merged) { + keys.emplace_back(k); + vals.emplace_back(v); + } + LOG(debug) << "metadata-collector: emitting " << keys.size() << " aggregated metadata entries"; + pc.outputs().snapshot(Output{"AMD", "AODMetadataKeys", 0}, keys); + pc.outputs().snapshot(Output{"AMD", "AODMetadataVals", 0}, vals); + }; + }}; +} } // namespace o2::framework::writers diff --git a/Framework/AnalysisSupport/src/AODWriterHelpers.h b/Framework/AnalysisSupport/src/AODWriterHelpers.h index 7ae59a5cf3b01..65b068cb630a7 100644 --- a/Framework/AnalysisSupport/src/AODWriterHelpers.h +++ b/Framework/AnalysisSupport/src/AODWriterHelpers.h @@ -21,6 +21,7 @@ namespace o2::framework::writers struct AODWriterHelpers { static AlgorithmSpec getOutputObjHistWriter(ConfigContext const& context); static AlgorithmSpec getOutputTTreeWriter(ConfigContext const& context); + static AlgorithmSpec getMetadataCollector(ConfigContext const& context); }; } // namespace o2::framework::writers diff --git a/Framework/AnalysisSupport/src/Plugin.cxx b/Framework/AnalysisSupport/src/Plugin.cxx index 5f61a236cbd58..f20475a1a2bd2 100644 --- a/Framework/AnalysisSupport/src/Plugin.cxx +++ b/Framework/AnalysisSupport/src/Plugin.cxx @@ -54,6 +54,13 @@ struct ROOTTTreeWriter : o2::framework::AlgorithmPlugin { } }; +struct ROOTMetadataCollector : o2::framework::AlgorithmPlugin { + o2::framework::AlgorithmSpec create(o2::framework::ConfigContext const& config) override + { + return o2::framework::writers::AODWriterHelpers::getMetadataCollector(config); + } +}; + using namespace o2::framework; struct RunSummary : o2::framework::ServicePlugin { o2::framework::ServiceSpec* create() final @@ -286,6 +293,7 @@ DEFINE_DPL_PLUGINS_BEGIN DEFINE_DPL_PLUGIN_INSTANCE(ROOTFileReader, CustomAlgorithm); DEFINE_DPL_PLUGIN_INSTANCE(ROOTObjWriter, CustomAlgorithm); DEFINE_DPL_PLUGIN_INSTANCE(ROOTTTreeWriter, CustomAlgorithm); +DEFINE_DPL_PLUGIN_INSTANCE(ROOTMetadataCollector, CustomAlgorithm); DEFINE_DPL_PLUGIN_INSTANCE(RunSummary, CustomService); DEFINE_DPL_PLUGIN_INSTANCE(DiscoverMetadataInAOD, ConfigDiscovery); DEFINE_DPL_PLUGINS_END diff --git a/Framework/Core/include/Framework/AnalysisSupportHelpers.h b/Framework/Core/include/Framework/AnalysisSupportHelpers.h index c1968123e765d..8eaaa4f1a61a6 100644 --- a/Framework/Core/include/Framework/AnalysisSupportHelpers.h +++ b/Framework/Core/include/Framework/AnalysisSupportHelpers.h @@ -48,6 +48,9 @@ struct AnalysisSupportHelpers { static DataProcessorSpec getOutputObjHistSink(ConfigContext const&); /// writes inputs of kind AOD to file static DataProcessorSpec getGlobalAODSink(ConfigContext const&); + /// Match all inputs of kind META, merge them and republish as the AOD + /// metadata keys/vals consumed by the AOD writer. + static DataProcessorSpec getMetadataCollectorSink(ConfigContext const&); /// Get the data director static std::shared_ptr getDataOutputDirector(ConfigContext const& ctx); }; diff --git a/Framework/Core/src/AnalysisSupportHelpers.cxx b/Framework/Core/src/AnalysisSupportHelpers.cxx index c16a1da61ae8a..ba6cb23e8652f 100644 --- a/Framework/Core/src/AnalysisSupportHelpers.cxx +++ b/Framework/Core/src/AnalysisSupportHelpers.cxx @@ -220,4 +220,23 @@ DataProcessorSpec return spec; } + +DataProcessorSpec + AnalysisSupportHelpers::getMetadataCollectorSink(ConfigContext const& ctx) +{ + // Lifetime is sporadic because META messages are not produced every + // timeframe. The oldest-possible-timeframe completion policy (registered + // in CompletionPolicy::createDefaultPolicies) decides when the collected + // parts are merged and republished as the AOD metadata keys/vals that the + // AOD writer turns into the AO2D metaData object. + DataProcessorSpec spec{ + .name = "internal-dpl-metadata-collector", + .inputs = {InputSpec("meta", DataSpecUtils::dataDescriptorMatcherFrom(header::DataOrigin{"META"}), Lifetime::Sporadic)}, + .outputs = {OutputSpec{OutputLabel{"keys"}, header::DataOrigin{"AMD"}, header::DataDescription{"AODMetadataKeys"}, 0, Lifetime::Sporadic}, + OutputSpec{OutputLabel{"vals"}, header::DataOrigin{"AMD"}, header::DataDescription{"AODMetadataVals"}, 0, Lifetime::Sporadic}}, + .algorithm = PluginManager::loadAlgorithmFromPlugin("O2FrameworkAnalysisSupport", "ROOTMetadataCollector", ctx), + }; + + return spec; +} } // namespace o2::framework diff --git a/Framework/Core/src/CompletionPolicy.cxx b/Framework/Core/src/CompletionPolicy.cxx index a09028b9249f3..27b5485207978 100644 --- a/Framework/Core/src/CompletionPolicy.cxx +++ b/Framework/Core/src/CompletionPolicy.cxx @@ -27,6 +27,7 @@ std::vector return { CompletionPolicyHelpers::consumeWhenAllOrdered("internal-dpl-aod-writer"), CompletionPolicyHelpers::consumeWhenAnyZeroCount("internal-dpl-injected-dummy-sink", [](DeviceSpec const& s) { return s.name.find("internal-dpl-injected-dummy-sink") != std::string::npos; }), + CompletionPolicyHelpers::consumeWhenPastOldestPossibleTimeframe("internal-dpl-metadata-collector", [](DeviceSpec const& s) { return s.name == "internal-dpl-metadata-collector"; }), CompletionPolicyHelpers::consumeWhenAll()}; } diff --git a/Framework/Core/src/WorkflowHelpers.cxx b/Framework/Core/src/WorkflowHelpers.cxx index 188b6653c6a43..df689d961245a 100644 --- a/Framework/Core/src/WorkflowHelpers.cxx +++ b/Framework/Core/src/WorkflowHelpers.cxx @@ -236,6 +236,7 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext std::vector requestedCCDBs; std::vector providedCCDBs; + bool hasMetaOutput = false; for (size_t wi = 0; wi < workflow.size(); ++wi) { auto& processor = workflow[wi]; @@ -392,6 +393,8 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext } else { it->bindings.push_back(output.binding.value); } + } else if (DataSpecUtils::partialMatch(output, header::DataOrigin{"META"})) { + hasMetaOutput = true; } if (output.lifetime == Lifetime::Condition) { @@ -583,6 +586,12 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext extraSpecs.push_back(rootSink); } + // Inject a collector which merges all META messages and republishes them as + // the AOD metadata keys/vals the AOD writer writes into the AO2D file. + if (hasMetaOutput) { + extraSpecs.push_back(AnalysisSupportHelpers::getMetadataCollectorSink(ctx)); + } + workflow.insert(workflow.end(), extraSpecs.begin(), extraSpecs.end()); extraSpecs.clear(); diff --git a/Framework/TestWorkflows/src/o2TestHistograms.cxx b/Framework/TestWorkflows/src/o2TestHistograms.cxx index 9c2cba35b9156..e4e0549f866c3 100644 --- a/Framework/TestWorkflows/src/o2TestHistograms.cxx +++ b/Framework/TestWorkflows/src/o2TestHistograms.cxx @@ -16,11 +16,18 @@ #include "Framework/runDataProcessing.h" #include "Framework/AnalysisTask.h" #include +#include +#include +#include using namespace o2; using namespace o2::framework; using namespace o2::framework::expressions; +// pT cut applied when producing the skimmed derived dataset; the same value is +// published as run metadata so it lands in the derived AO2D's metaData object. +static constexpr float kSkimPtCut = 1.5f; + namespace o2::aod { O2ORIGIN("EMB"); @@ -38,8 +45,8 @@ DECLARE_SOA_TABLE(SkimmedExampleTrack, "AOD", "SKIMEXTRK", //! struct EtaAndClsHistogramsSimple { OutputObj etaClsH{TH2F("eta_vs_pt", "#eta vs pT", 102, -2.01, 2.01, 100, 0, 10)}; Produces skimEx; - Configurable trackFilterString{"track-filter", "o2::aod::track::pt < 10.f", "Track filter string"}; - Filter trackFilter = o2::aod::track::pt < 10.f; + Configurable trackFilterString{"track-filter", "", "Track filter string (overrides the pT cut when set)"}; + Filter trackFilter = o2::aod::track::pt < kSkimPtCut; HistogramRegistry registry{ "registry", @@ -88,8 +95,8 @@ struct EtaAndClsHistogramsSimple { struct EtaAndClsHistogramsIUSimple { OutputObj etaClsH{TH2F("eta_vs_pt", "#eta vs pT", 102, -2.01, 2.01, 100, 0, 10)}; Produces skimEx; - Configurable trackFilterString{"track-filter", "o2::aod::track::pt < 10.f", "Track filter string"}; - Filter trackFilter = o2::aod::track::pt < 10.f; + Configurable trackFilterString{"track-filter", "", "Track filter string (overrides the pT cut when set)"}; + Filter trackFilter = o2::aod::track::pt < kSkimPtCut; HistogramRegistry registry{ "registry", @@ -136,8 +143,8 @@ struct EtaAndClsHistogramsFull { } // }; - Configurable trackFilterString{"track-filter", "o2::aod::track::pt < 10.f", "Track filter string"}; - Filter trackFilter = o2::aod::track::pt < 10.f; + Configurable trackFilterString{"track-filter", "", "Track filter string (overrides the pT cut when set)"}; + Filter trackFilter = o2::aod::track::pt < kSkimPtCut; void init(InitContext&) { @@ -183,25 +190,37 @@ WorkflowSpec defineDataProcessing(ConfigContext const& cfgc) LOGP(info, "- {} present.", table); } // Notice it's important for the tasks to use the same name, otherwise topology generation will be confused. + WorkflowSpec specs; if (runType == "2" || !hasTrackCov) { LOGP(info, "Using only tracks {}", runType); if (hasTrackIU) { - return WorkflowSpec{ - adaptAnalysisTask(cfgc, TaskName{"simple-histos"}), - }; + specs = WorkflowSpec{adaptAnalysisTask(cfgc, TaskName{"simple-histos"})}; + } else { + specs = WorkflowSpec{adaptAnalysisTask(cfgc, TaskName{"simple-histos"})}; } - return WorkflowSpec{ - adaptAnalysisTask(cfgc, TaskName{"simple-histos"}), - }; } else { LOGP(info, "Using tracks extra {}", runType); if (hasTrackIU) { - return WorkflowSpec{ - adaptAnalysisTask(cfgc, TaskName{"simple-histos"}), - }; + specs = WorkflowSpec{adaptAnalysisTask(cfgc, TaskName{"simple-histos"})}; + } else { + specs = WorkflowSpec{adaptAnalysisTask(cfgc, TaskName{"simple-histos"})}; } - return WorkflowSpec{ - adaptAnalysisTask(cfgc, TaskName{"simple-histos"}), - }; } + + // Publish the skimming pT cut as run metadata, once per data frame so it is + // aligned with the derived tables. The auto-injected metadata collector merges + // all META messages (oldest-possible completion) and the AOD writer stores the + // result as the metaData object of the derived AO2D file. + specs.push_back(DataProcessorSpec{ + .name = "skim-metadata", + .inputs = {InputSpec{"tfn", "TFN", "TFNumber"}}, + .outputs = {OutputSpec{{"meta"}, "META", "SKIMINFO", 0, Lifetime::Sporadic}}, + .algorithm = adaptStateless([](ProcessingContext& pc) { + TMap m; + m.SetOwnerKeyValue(); + m.Add(new TObjString("SkimTrackPtCut"), new TObjString(TString::Format("%g", kSkimPtCut))); + pc.outputs().snapshot(Output{"META", "SKIMINFO", 0}, m); + }), + }); + return specs; } From 8ee74def0b1b0f66d0e72ba3fcbb7afe836cf824 Mon Sep 17 00:00:00 2001 From: Giulio Eulisse <10544+ktf@users.noreply.github.com> Date: Mon, 29 Jun 2026 13:49:43 +0200 Subject: [PATCH 2/6] Update Framework/AnalysisSupport/src/AODWriterHelpers.cxx Co-authored-by: Anton Alkin --- Framework/AnalysisSupport/src/AODWriterHelpers.cxx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Framework/AnalysisSupport/src/AODWriterHelpers.cxx b/Framework/AnalysisSupport/src/AODWriterHelpers.cxx index cd7aaf55abf76..60ab89c3df465 100644 --- a/Framework/AnalysisSupport/src/AODWriterHelpers.cxx +++ b/Framework/AnalysisSupport/src/AODWriterHelpers.cxx @@ -500,7 +500,7 @@ AlgorithmSpec AODWriterHelpers::getMetadataCollector(ConfigContext const& /*ctx* TObject* value = part->GetValue(key); std::string k = key->GetName(); std::string v = value != nullptr ? value->GetName() : ""; - auto it = std::find_if(merged->begin(), merged->end(), + auto it = std::ranges::find_if(merged, [&k](auto const& e) { return e.first == k; }); if (it != merged->end()) { it->second = std::move(v); From 62fd72e52cf402af00da32cd2e1334790f034b8b Mon Sep 17 00:00:00 2001 From: Giulio Eulisse <10544+ktf@users.noreply.github.com> Date: Mon, 29 Jun 2026 15:57:43 +0200 Subject: [PATCH 3/6] Update Framework/AnalysisSupport/src/AODWriterHelpers.cxx Co-authored-by: Anton Alkin --- Framework/AnalysisSupport/src/AODWriterHelpers.cxx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Framework/AnalysisSupport/src/AODWriterHelpers.cxx b/Framework/AnalysisSupport/src/AODWriterHelpers.cxx index 60ab89c3df465..86a8aa1694a99 100644 --- a/Framework/AnalysisSupport/src/AODWriterHelpers.cxx +++ b/Framework/AnalysisSupport/src/AODWriterHelpers.cxx @@ -500,7 +500,7 @@ AlgorithmSpec AODWriterHelpers::getMetadataCollector(ConfigContext const& /*ctx* TObject* value = part->GetValue(key); std::string k = key->GetName(); std::string v = value != nullptr ? value->GetName() : ""; - auto it = std::ranges::find_if(merged, + auto it = std::ranges::find_if(*merged, [&k](auto const& e) { return e.first == k; }); if (it != merged->end()) { it->second = std::move(v); From 70f5cd8ab822c7e3feb66140ea8ac3449c50a663 Mon Sep 17 00:00:00 2001 From: shahoian Date: Tue, 7 Jul 2026 18:15:05 +0200 Subject: [PATCH 4/6] Method to extract ConfigurableParam as JSON string and update from such string --- .../include/CommonUtils/ConfigurableParam.h | 5 + Common/Utils/src/ConfigurableParam.cxx | 98 ++++++++++++++----- Common/Utils/test/testConfigurableParam.cxx | 36 +++++++ 3 files changed, 117 insertions(+), 22 deletions(-) diff --git a/Common/Utils/include/CommonUtils/ConfigurableParam.h b/Common/Utils/include/CommonUtils/ConfigurableParam.h index 0a6da30aa39f5..fa4acd2744703 100644 --- a/Common/Utils/include/CommonUtils/ConfigurableParam.h +++ b/Common/Utils/include/CommonUtils/ConfigurableParam.h @@ -407,6 +407,8 @@ class ConfigurableParam // writes a human readable INI or JSON file depending on the extension static void write(std::string const& filename, std::string const& keyOnly = ""); + static std::string asJSON(std::string const& keyOnly = ""); + // can be used instead of using API on concrete child classes template static T getValueAs(std::string key) @@ -494,6 +496,9 @@ class ConfigurableParam // be updated, absence of data for any of requested params will lead to fatal static void updateFromFile(std::string const&, std::string const& paramsList = "", bool unchangedOnly = false); + // update from a JSON string with the same filtering semantics as updateFromFile + static void updateFromJSONString(std::string const&, std::string const& paramsList = "", bool unchangedOnly = false); + // interface for use from the CCDB API; allows to sync objects read from CCDB with the information // stored in the registry; modifies given object as well as registry virtual void syncCCDBandRegistry(void* obj) = 0; diff --git a/Common/Utils/src/ConfigurableParam.cxx b/Common/Utils/src/ConfigurableParam.cxx index 202433b04e4c4..4de478a6d0e9f 100644 --- a/Common/Utils/src/ConfigurableParam.cxx +++ b/Common/Utils/src/ConfigurableParam.cxx @@ -37,6 +37,7 @@ #endif #include #include +#include #include #include #include @@ -754,6 +755,29 @@ void ConfigurableParam::writeJSON(std::string const& filename, std::string const // ------------------------------------------------------------------ +std::string ConfigurableParam::asJSON(std::string const& keyOnly) +{ + initPropertyTree(); // update the boost tree before writing + std::ostringstream os; + if (!keyOnly.empty()) { // write ini for selected key only + try { + boost::property_tree::ptree kTree; + auto keys = o2::utils::Str::tokenize(keyOnly, " ,;", true, true); + for (const auto& k : keys) { + kTree.add_child(k, sPtree->get_child(k)); + } + boost::property_tree::write_json(os, kTree); + } catch (const boost::property_tree::ptree_bad_path& err) { + LOG(fatal) << "non-existing key " << keyOnly << " provided to writeJSON"; + } + } else { + boost::property_tree::write_json(os, *sPtree); + } + return os.str(); +} + +// ------------------------------------------------------------------ + void ConfigurableParam::initPropertyTree() { sPtree->clear(); @@ -870,26 +894,10 @@ void ConfigurableParam::printAllRegisteredParamNames() // ------------------------------------------------------------------ -// Update the storage map of params from the given configuration file. -// It can be in JSON or INI format. -// If nonempty comma-separated paramsList is provided, only those params will -// be updated, absence of data for any of requested params will lead to fatal -// If unchangedOnly is true, then only those parameters whose provenance is kCODE will be updated -// (to allow preference of run-time settings) -void ConfigurableParam::updateFromFile(std::string const& configFile, std::string const& paramsList, bool unchangedOnly) +namespace +{ +void updateFromPropertyTree(boost::property_tree::ptree const& pt, std::string const& source, std::string const& paramsList, bool unchangedOnly) { - if (!sIsFullyInitialized) { - initialize(); - } - - auto cfgfile = o2::utils::Str::trim_copy(configFile); - - if (cfgfile.length() == 0) { - return; - } - - boost::property_tree::ptree pt = ConfigurableParamReaders::readConfigFile(cfgfile); - std::vector> keyValPairs; auto request = o2::utils::Str::tokenize(paramsList, ',', true); std::unordered_map requestMap; @@ -913,7 +921,7 @@ void ConfigurableParam::updateFromFile(std::string const& configFile, std::strin auto name = subKey.first; auto value = subKey.second.get_value(); std::string key = mainKey + "." + name; - if (!unchangedOnly || getProvenance(key) == kCODE) { + if (!unchangedOnly || ConfigurableParam::getProvenance(key) == ConfigurableParam::kCODE) { std::pair pair = std::make_pair(key, o2::utils::Str::trim_copy(value)); keyValPairs.push_back(pair); } @@ -928,16 +936,62 @@ void ConfigurableParam::updateFromFile(std::string const& configFile, std::strin // make sure all requested params were retrieved for (const auto& req : requestMap) { if (req.second == 0) { - throw std::runtime_error(fmt::format("Param {:s} was not found in {:s}", req.first, configFile)); + throw std::runtime_error(fmt::format("Param {:s} was not found in {:s}", req.first, source)); } } try { - setValues(keyValPairs); + ConfigurableParam::setValues(keyValPairs); } catch (std::exception const& error) { LOG(error) << "Error while setting values " << error.what(); } } +} // namespace + +// Update the storage map of params from the given configuration file. +// It can be in JSON or INI format. +// If nonempty comma-separated paramsList is provided, only those params will +// be updated, absence of data for any of requested params will lead to fatal +// If unchangedOnly is true, then only those parameters whose provenance is kCODE will be updated +// (to allow preference of run-time settings) +void ConfigurableParam::updateFromFile(std::string const& configFile, std::string const& paramsList, bool unchangedOnly) +{ + if (!sIsFullyInitialized) { + initialize(); + } + + auto cfgfile = o2::utils::Str::trim_copy(configFile); + + if (cfgfile.length() == 0) { + return; + } + + updateFromPropertyTree(ConfigurableParamReaders::readConfigFile(cfgfile), configFile, paramsList, unchangedOnly); +} + +// ------------------------------------------------------------------ + +void ConfigurableParam::updateFromJSONString(std::string const& configJSON, std::string const& paramsList, bool unchangedOnly) +{ + if (!sIsFullyInitialized) { + initialize(); + } + + auto json = o2::utils::Str::trim_copy(configJSON); + if (json.length() == 0) { + return; + } + + boost::property_tree::ptree pt; + std::istringstream input(json); + try { + boost::property_tree::read_json(input, pt); + } catch (const boost::property_tree::ptree_error& e) { + LOG(fatal) << "Failed to read JSON config string (" << e.what() << ")"; + } + + updateFromPropertyTree(pt, "provided JSON string", paramsList, unchangedOnly); +} // ------------------------------------------------------------------ // ------------------------------------------------------------------ diff --git a/Common/Utils/test/testConfigurableParam.cxx b/Common/Utils/test/testConfigurableParam.cxx index f0c3c59c79c35..6fd0344cdd1be 100644 --- a/Common/Utils/test/testConfigurableParam.cxx +++ b/Common/Utils/test/testConfigurableParam.cxx @@ -151,6 +151,42 @@ BOOST_AUTO_TEST_CASE(ConfigurableParam_FileIO_Json) std::remove(testFileName.c_str()); } +BOOST_AUTO_TEST_CASE(ConfigurableParam_JSONString_UnchangedOnly) +{ + ConfigurableParam::setValue("TestParam.ulValue", "2"); + ConfigurableParam::setProvenance("TestParam", "lValue", ConfigurableParam::kCODE); + ConfigurableParam::setProvenance("TestParam", "ulValue", ConfigurableParam::kRT); + ConfigurableParam::updateFromJSONString(R"json({"TestParam":{"lValue":"77","ulValue":"88"}})json", "TestParam", true); + + BOOST_CHECK_EQUAL(TestParam::Instance().lValue, 77); + BOOST_CHECK_EQUAL(TestParam::Instance().ulValue, 2); +} + +BOOST_AUTO_TEST_CASE(ConfigurableParam_JSONString_FromAsJSON) +{ + ConfigurableParam::setValue("TestParam.iValue", "321"); + ConfigurableParam::setValue("TestParam.sValue", "json-source"); + ConfigurableParam::setValues({{"TestParam.map", "{7:8,9:10}"}}); + const auto mapBefore = TestParam::Instance().map; + const auto json = ConfigurableParam::asJSON("TestParam"); + + ConfigurableParam::setValue("TestParam.iValue", "999"); + ConfigurableParam::setValue("TestParam.sValue", "json-modified"); + ConfigurableParam::setValues({{"TestParam.map", "{1:2}"}}); + ConfigurableParam::updateFromJSONString(json, "TestParam"); + + BOOST_CHECK_EQUAL(TestParam::Instance().iValue, 321); + BOOST_CHECK_EQUAL(TestParam::Instance().sValue, "json-source"); + BOOST_CHECK_EQUAL(TestParam::Instance().map.size(), mapBefore.size()); + BOOST_CHECK_EQUAL(TestParam::Instance().map.at(7), mapBefore.at(7)); + BOOST_CHECK_EQUAL(TestParam::Instance().map.at(9), mapBefore.at(9)); +} + +BOOST_AUTO_TEST_CASE(ConfigurableParam_JSONString_ParamsListMissing) +{ + BOOST_CHECK_THROW(ConfigurableParam::updateFromJSONString(ConfigurableParam::asJSON("TestParam"), "MissingParam"), std::runtime_error); +} + BOOST_AUTO_TEST_CASE(ConfigurableParam_FileIO_ROOT) { // test for root file serialization From 8e64c6f77c2b3d00c10176471ddec33d048db01a Mon Sep 17 00:00:00 2001 From: shahoian Date: Tue, 7 Jul 2026 18:15:28 +0200 Subject: [PATCH 5/6] Send reco config.params to AODMetadata --- Detectors/ITSMFT/ITS/workflow/src/TrackerSpec.cxx | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/Detectors/ITSMFT/ITS/workflow/src/TrackerSpec.cxx b/Detectors/ITSMFT/ITS/workflow/src/TrackerSpec.cxx index bbafc48e931ed..bfa1b05f08923 100644 --- a/Detectors/ITSMFT/ITS/workflow/src/TrackerSpec.cxx +++ b/Detectors/ITSMFT/ITS/workflow/src/TrackerSpec.cxx @@ -10,7 +10,8 @@ // or submit itself to any jurisdiction. #include - +#include +#include #include "Framework/ControlService.h" #include "Framework/ConfigParamRegistry.h" #include "Framework/CCDBParamSpec.h" @@ -67,8 +68,15 @@ void TrackerDPL::run(ProcessingContext& pc) if (first) { first = false; if (pc.services().get().inputTimesliceId == 0) { - o2::conf::ConfigurableParam::write(o2::base::NameConf::getConfigOutputFileName(pc.services().get().name, o2::its::VertexerParamConfig::Instance().getName()), o2::its::VertexerParamConfig::Instance().getName()); - o2::conf::ConfigurableParam::write(o2::base::NameConf::getConfigOutputFileName(pc.services().get().name, o2::its::TrackerParamConfig::Instance().getName()), o2::its::TrackerParamConfig::Instance().getName()); + const auto& vtconf = o2::its::VertexerParamConfig::Instance(); + const auto& trconf = o2::its::TrackerParamConfig::Instance(); + o2::conf::ConfigurableParam::write(o2::base::NameConf::getConfigOutputFileName(pc.services().get().name, vtconf.getName()), vtconf.getName()); + o2::conf::ConfigurableParam::write(o2::base::NameConf::getConfigOutputFileName(pc.services().get().name, trconf.getName()), trconf.getName()); + TMap md; + md.SetOwnerKeyValue(); + md.Add(new TObjString(vtconf.getName().c_str()), new TObjString(o2::conf::ConfigurableParam::asJSON(vtconf.getName()).c_str())); + md.Add(new TObjString(trconf.getName().c_str()), new TObjString(o2::conf::ConfigurableParam::asJSON(trconf.getName()).c_str())); + pc.outputs().snapshot(Output{"META", "ITSTRACKER", 0}, md); } } } @@ -138,6 +146,7 @@ DataProcessorSpec getTrackerSpec(bool useMC, bool doStag, bool useGeom, int trgT outputs.emplace_back("ITS", "VERTICESMCPUR", 0, Lifetime::Timeframe); outputs.emplace_back("ITS", "TRACKSMCTR", 0, Lifetime::Timeframe); } + outputs.emplace_back("META", "ITSTRACKER", 0, Lifetime::Sporadic); return DataProcessorSpec{ .name = "its-tracker", From 8ab03f582338fad1fbf07afd1cbd29d22511020b Mon Sep 17 00:00:00 2001 From: shahoian Date: Wed, 8 Jul 2026 19:59:09 +0200 Subject: [PATCH 6/6] Fixes from codex: leads to aod-writer waiting for inputs --- .../Core/src/CompletionPolicyHelpers.cxx | 24 +++++- Framework/Core/src/DataProcessingDevice.cxx | 30 +++++++- Framework/Core/src/WorkflowHelpers.cxx | 76 +++++++++++++++---- Framework/Core/test/test_WorkflowHelpers.cxx | 44 +++++++++++ 4 files changed, 153 insertions(+), 21 deletions(-) diff --git a/Framework/Core/src/CompletionPolicyHelpers.cxx b/Framework/Core/src/CompletionPolicyHelpers.cxx index d2c63cbf4c90b..7b7b7a9727ada 100644 --- a/Framework/Core/src/CompletionPolicyHelpers.cxx +++ b/Framework/Core/src/CompletionPolicyHelpers.cxx @@ -175,13 +175,25 @@ CompletionPolicy CompletionPolicyHelpers::consumeWhenAll(const char* name, Compl CompletionPolicy CompletionPolicyHelpers::consumeWhenAllOrdered(const char* name, CompletionPolicy::Matcher matcher) { - auto callbackFull = [](InputSpan const& inputs, std::vector const&, ServiceRegistryRef& ref) -> CompletionPolicy::CompletionOp { + auto callbackFull = [](InputSpan const& inputs, std::vector const& specs, ServiceRegistryRef& ref) -> CompletionPolicy::CompletionOp { + assert(inputs.size() == specs.size()); auto& decongestionService = ref.get(); decongestionService.orderedCompletionPolicyActive = true; + bool hasPresentSporadic = false; + bool hasOrderedInput = false; + bool missingOrderedInput = false; + size_t si = 0; for (auto& input : inputs) { + auto const& spec = specs[si++]; + if (spec.lifetime == Lifetime::Sporadic) { + hasPresentSporadic |= input.header != nullptr; + continue; + } if (input.header == nullptr) { - return CompletionPolicy::CompletionOp::Wait; + missingOrderedInput = true; + continue; } + hasOrderedInput = true; long int startTime = framework::DataRefUtils::getHeader(input)->startTime; if (startTime == 0) { LOGP(debug, "startTime is 0, which means we have the first message, so we can process it."); @@ -191,6 +203,14 @@ CompletionPolicy CompletionPolicyHelpers::consumeWhenAllOrdered(const char* name return CompletionPolicy::CompletionOp::Retry; } } + // Sporadic side inputs, like metadata collected for the AOD writer, are not + // produced for every TF. Process them when they are present, but do not let + // them block or advance the ordered timeframe progression. The device loop + // removes only the sporadic parts for ConsumeExisting, leaving any partial + // ordered TF record in place. + if (missingOrderedInput || !hasOrderedInput) { + return hasPresentSporadic ? CompletionPolicy::CompletionOp::ConsumeExisting : CompletionPolicy::CompletionOp::Wait; + } decongestionService.nextTimeslice++; return CompletionPolicy::CompletionOp::ConsumeAndRescan; }; diff --git a/Framework/Core/src/DataProcessingDevice.cxx b/Framework/Core/src/DataProcessingDevice.cxx index 4121d333f6b56..24366860a19e4 100644 --- a/Framework/Core/src/DataProcessingDevice.cxx +++ b/Framework/Core/src/DataProcessingDevice.cxx @@ -2263,6 +2263,17 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v } }; + auto cleanSporadic = [¤tSetOfInputs](TimesliceSlot, InputRecord& record) { + assert(record.size() == currentSetOfInputs.size()); + for (size_t ii = 0, ie = record.size(); ii < ie; ++ii) { + DataRef input = record.getByPos(ii); + if (input.spec == nullptr || input.spec->lifetime != Lifetime::Sporadic || input.header == nullptr) { + continue; + } + currentSetOfInputs[ii].clear(); + } + }; + // Function to cleanup record. For the moment we // simply use it to keep track of input messages // which are not needed, to display them in the GUI. @@ -2448,7 +2459,7 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v static bool noCatch = getenv("O2_NO_CATCHALL_EXCEPTIONS") && strcmp(getenv("O2_NO_CATCHALL_EXCEPTIONS"), "0"); - auto runNoCatch = [&context, ref, &processContext](DataRelayer::RecordAction& action) mutable { + auto runNoCatch = [&context, ref, &processContext, &record](DataRelayer::RecordAction& action) mutable { auto& state = ref.get(); auto& spec = ref.get(); auto& streamContext = ref.get(); @@ -2498,8 +2509,18 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v } } - // Notify the sink we just consumed some timeframe data - if (context.isSink && action.op == CompletionPolicy::CompletionOp::Consume) { + // Notify the sink we just consumed some timeframe data. A sink can also + // consume sporadic side inputs, e.g. AOD metadata, without consuming a + // TF; those records must not emit DPL/SUMMARY feedback. + auto consumedTimeframeInput = [&record, &spec]() { + for (size_t ii = 0; ii < spec.inputs.size(); ++ii) { + if (spec.inputs[ii].matcher.lifetime != Lifetime::Sporadic && record.isValid(ii)) { + return true; + } + } + return false; + }; + if (context.isSink && action.op == CompletionPolicy::CompletionOp::Consume && consumedTimeframeInput()) { O2_SIGNPOST_EVENT_EMIT(device, pcid, "device", "Sending dpl-summary"); auto& allocator = ref.get(); allocator.make(OutputRef{"dpl-summary", runtime_hash(spec.name.c_str())}, 1); @@ -2558,6 +2579,9 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v context.postDispatchingCallbacks(processContext); ref.get().call(o2::framework::ServiceRegistryRef{ref}); } + if (action.op == CompletionPolicy::CompletionOp::ConsumeExisting) { + cleanSporadic(action.slot, record); + } if ((context.forwardPolicy == ForwardPolicy::AfterProcessing) && hasForwards && consumeSomething) { O2_SIGNPOST_EVENT_EMIT(device, aid, "device", "Late forwarding"); auto& timesliceIndex = ref.get(); diff --git a/Framework/Core/src/WorkflowHelpers.cxx b/Framework/Core/src/WorkflowHelpers.cxx index df689d961245a..35f9133f6006a 100644 --- a/Framework/Core/src/WorkflowHelpers.cxx +++ b/Framework/Core/src/WorkflowHelpers.cxx @@ -586,9 +586,27 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext extraSpecs.push_back(rootSink); } - // Inject a collector which merges all META messages and republishes them as - // the AOD metadata keys/vals the AOD writer writes into the AO2D file. - if (hasMetaOutput) { + auto hasWritableAODOutput = std::ranges::any_of(workflow, [](DataProcessorSpec const& spec) { + return std::ranges::any_of(spec.outputs, [](OutputSpec const& output) { + return DataSpecUtils::partialMatch(output, writableAODOrigins); + }); + }); + auto hasTFNumberOutput = std::ranges::any_of(workflow, [](DataProcessorSpec const& spec) { + return std::ranges::any_of(spec.outputs, [](OutputSpec const& output) { + return DataSpecUtils::match(output, "TFN", "TFNumber", 0); + }); + }); + auto hasTFFileNameOutput = std::ranges::any_of(workflow, [](DataProcessorSpec const& spec) { + return std::ranges::any_of(spec.outputs, [](OutputSpec const& output) { + return DataSpecUtils::match(output, "TFF", "TFFilename", 0); + }); + }); + + // Inject a collector which merges META messages only when the same topology + // can also write the resulting AMD metadata to an AO2D. Otherwise split + // detector subworkflows get an internal collector whose AMD output has no + // meaningful AOD writer route and the service device can hang the workflow. + if (hasMetaOutput && (hasWritableAODOutput || (hasTFNumberOutput && hasTFFileNameOutput))) { extraSpecs.push_back(AnalysisSupportHelpers::getMetadataCollectorSink(ctx)); } @@ -753,13 +771,38 @@ void WorkflowHelpers::injectAODWriter(WorkflowSpec& workflow, ConfigContext cons // create DataOutputDescriptor std::shared_ptr dod = AnalysisSupportHelpers::getDataOutputDirector(ctx); + auto hasWritableAOD = std::ranges::any_of(dec.outputsInputs, [](InputSpec const& spec) { + return DataSpecUtils::partialMatch(spec, writableAODOrigins); + }); + auto hasTFNumber = std::ranges::any_of(dec.outputsInputs, [](InputSpec const& spec) { + return DataSpecUtils::partialMatch(spec, header::DataOrigin{"TFN"}) && + DataSpecUtils::partialMatch(spec, header::DataDescription{"TFNumber"}); + }); + auto hasTFFileName = std::ranges::any_of(dec.outputsInputs, [](InputSpec const& spec) { + return DataSpecUtils::partialMatch(spec, header::DataOrigin{"TFF"}) && + DataSpecUtils::partialMatch(spec, header::DataDescription{"TFFilename"}); + }); + auto canWriteMetadataOnly = hasTFNumber && hasTFFileName; + // select outputs of type AOD which need to be saved dec.outputsInputsAOD.clear(); for (auto ii = 0u; ii < dec.outputsInputs.size(); ii++) { - if (DataSpecUtils::partialMatch(dec.outputsInputs[ii], AODOrigins)) { - auto ds = dod->getDataOutputDescriptors(dec.outputsInputs[ii]); - if (ds.size() > 0 || dec.isDangling[ii]) { - dec.outputsInputsAOD.emplace_back(dec.outputsInputs[ii]); + auto const& input = dec.outputsInputs[ii]; + if (!DataSpecUtils::partialMatch(input, AODOrigins)) { + continue; + } + if (DataSpecUtils::partialMatch(input, header::DataOrigin{"AMD"}) && !hasWritableAOD && !canWriteMetadataOnly) { + continue; + } + auto ds = dod->getDataOutputDescriptors(input); + if (ds.size() > 0 || dec.isDangling[ii]) { + dec.outputsInputsAOD.emplace_back(input); + if (DataSpecUtils::partialMatch(input, header::DataOrigin{"AMD"})) { + // Metadata collected from META outputs is republished as AMD sporadic + // data. Keep the AOD writer input sporadic so the ordered completion + // policy can consume it as side information without requiring it for + // every TF. + dec.outputsInputsAOD.back().lifetime = Lifetime::Sporadic; } } } @@ -772,15 +815,16 @@ void WorkflowHelpers::injectAODWriter(WorkflowSpec& workflow, ConfigContext cons auto fileSink = AnalysisSupportHelpers::getGlobalAODSink(ctx); workflow.push_back(fileSink); - auto it = std::find_if(dec.outputsInputs.begin(), dec.outputsInputs.end(), [](InputSpec const& spec) -> bool { - return DataSpecUtils::partialMatch(spec, o2::header::DataOrigin("TFN")); - }); - dec.isDangling[std::distance(dec.outputsInputs.begin(), it)] = false; - - it = std::find_if(dec.outputsInputs.begin(), dec.outputsInputs.end(), [](InputSpec const& spec) -> bool { - return DataSpecUtils::partialMatch(spec, o2::header::DataOrigin("TFF")); - }); - dec.isDangling[std::distance(dec.outputsInputs.begin(), it)] = false; + auto markAsMatched = [&](header::DataOrigin origin) { + auto it = std::find_if(dec.outputsInputs.begin(), dec.outputsInputs.end(), [&](InputSpec const& spec) -> bool { + return DataSpecUtils::partialMatch(spec, origin); + }); + if (it != dec.outputsInputs.end()) { + dec.isDangling[std::distance(dec.outputsInputs.begin(), it)] = false; + } + }; + markAsMatched(header::DataOrigin{"TFN"}); + markAsMatched(header::DataOrigin{"TFF"}); } } diff --git a/Framework/Core/test/test_WorkflowHelpers.cxx b/Framework/Core/test/test_WorkflowHelpers.cxx index d43e74558c0bb..02e7742d0592e 100644 --- a/Framework/Core/test/test_WorkflowHelpers.cxx +++ b/Framework/Core/test/test_WorkflowHelpers.cxx @@ -20,6 +20,7 @@ #include #include #include +#include using namespace o2::framework; @@ -494,6 +495,49 @@ TEST_CASE("TestExternalInput") availableForwardsInfo); } +TEST_CASE("TestMetadataOnlyAODWriterInjection") +{ + auto hasProcessor = [](WorkflowSpec const& workflow, std::string_view name) { + return std::ranges::any_of(workflow, [name](DataProcessorSpec const& spec) { + return spec.name == name; + }); + }; + auto findProcessor = [](WorkflowSpec const& workflow, std::string_view name) { + return std::ranges::find_if(workflow, [name](DataProcessorSpec const& spec) { + return spec.name == name; + }); + }; + + WorkflowSpec metadataOnlyWorkflow{ + {.name = "metadata-producer", + .outputs = {OutputSpec{"META", "TRACKER", 0, Lifetime::Sporadic}}}}; + + auto context = makeEmptyConfigContext(); + WorkflowHelpers::injectServiceDevices(metadataOnlyWorkflow, *context); + + REQUIRE_FALSE(hasProcessor(metadataOnlyWorkflow, "internal-dpl-metadata-collector")); + REQUIRE_FALSE(hasProcessor(metadataOnlyWorkflow, "internal-dpl-aod-writer")); + + WorkflowSpec metadataWithTFSourceWorkflow{ + {.name = "metadata-producer", + .outputs = {OutputSpec{"META", "TRACKER", 0, Lifetime::Sporadic}}}, + {.name = "tf-source", + .outputs = {OutputSpec{"TFN", "TFNumber"}, OutputSpec{"TFF", "TFFilename"}}}}; + + context = makeEmptyConfigContext(); + WorkflowHelpers::injectServiceDevices(metadataWithTFSourceWorkflow, *context); + + REQUIRE(hasProcessor(metadataWithTFSourceWorkflow, "internal-dpl-metadata-collector")); + REQUIRE(hasProcessor(metadataWithTFSourceWorkflow, "internal-dpl-aod-writer")); + auto writer = findProcessor(metadataWithTFSourceWorkflow, "internal-dpl-aod-writer"); + REQUIRE(writer != metadataWithTFSourceWorkflow.end()); + for (auto const& input : writer->inputs) { + if (DataSpecUtils::partialMatch(input, o2::header::DataOrigin{"AMD"})) { + REQUIRE(input.lifetime == Lifetime::Sporadic); + } + } +} + TEST_CASE("DetermineDanglingOutputs") { WorkflowSpec workflow0{