Merge ed328b5 into sapling-pr-archive-ktf

Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx

@@ -22,6 +22,7 @@
 #include "Framework/DeviceSpec.h"
 #include "Framework/RawDeviceService.h"
 #include "Framework/DataSpecUtils.h"
+#include "Framework/ConfigContext.h"
 #include "DataInputDirector.h"
 #include "Framework/SourceInfoHeader.h"
 #include "Framework/ChannelInfo.h"
@@ -117,12 +118,19 @@ static inline auto extractOriginalsTuple(framework::pack<Os...>, ProcessingConte
   return std::make_tuple(extractTypedOriginal<Os>(pc)...);
 }
 
-AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback()
+AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const& config)
 {
-  auto callback = AlgorithmSpec{adaptStateful([](ConfigParamRegistry const& options,
-                                                 DeviceSpec const& spec,
-                                                 Monitoring& monitoring,
-                                                 DataProcessingStats& stats) {
+  // aod-parent-base-path-replacement is now a workflow option, so it needs to be
+  // retrieved from the ConfigContext. This is because we do not allow workflow options
+  // to change over start-stop-start because they can affect the topology generation.
+  std::string parentFileReplacement;
+  if (config.options().isSet("aod-parent-base-path-replacement")) {
+    parentFileReplacement = config.options().get<std::string>("aod-parent-base-path-replacement");
+  }
+  auto callback = AlgorithmSpec{adaptStateful([parentFileReplacement](ConfigParamRegistry const& options,
+                                                                      DeviceSpec const& spec,
+                                                                      Monitoring& monitoring,
+                                                                      DataProcessingStats& stats) {
     // FIXME: not actually needed, since data processing stats can specify that we should
     // send the initial value.
     stats.updateStats({static_cast<short>(ProcessingStatsId::ARROW_BYTES_CREATED), DataProcessingStats::Op::Set, 0});
@@ -140,11 +148,6 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback()
 
     auto maxRate = options.get<float>("aod-max-io-rate");
 
-    std::string parentFileReplacement;
-    if (options.isSet("aod-parent-base-path-replacement")) {
-      parentFileReplacement = options.get<std::string>("aod-parent-base-path-replacement");
-    }
-
     int parentAccessLevel = 0;
     if (options.isSet("aod-parent-access-level")) {
       parentAccessLevel = options.get<int>("aod-parent-access-level");

Framework/AnalysisSupport/src/AODJAlienReaderHelpers.h

@@ -24,7 +24,7 @@ namespace o2::framework::readers
 {
 
 struct AODJAlienReaderHelpers {
-  static AlgorithmSpec rootFileReaderCallback();
+  static AlgorithmSpec rootFileReaderCallback(ConfigContext const&context);
   static void dumpFileMetrics(o2::monitoring::Monitoring& monitoring, TFile* currentFile, uint64_t startedAt, uint64_t ioTime, int tfPerFile, int tfRead);
 };
 

Framework/AnalysisSupport/src/Plugin.cxx

@@ -27,9 +27,9 @@
 O2_DECLARE_DYNAMIC_LOG(analysis_support);
 
 struct ROOTFileReader : o2::framework::AlgorithmPlugin {
-  o2::framework::AlgorithmSpec create() override
+  o2::framework::AlgorithmSpec create(o2::framework::ConfigContext const& config) override
   {
-    return o2::framework::readers::AODJAlienReaderHelpers::rootFileReaderCallback();
+    return o2::framework::readers::AODJAlienReaderHelpers::rootFileReaderCallback(config);
   }
 };
 

Framework/CCDBSupport/src/Plugin.cxx

@@ -13,7 +13,7 @@
 #include "CCDBHelpers.h"
 
 struct CCDBFetcherPlugin : o2::framework::AlgorithmPlugin {
-  o2::framework::AlgorithmSpec create() final
+  o2::framework::AlgorithmSpec create(o2::framework::ConfigContext const&) final
   {
     return o2::framework::CCDBHelpers::fetchFromCCDB();
   }

Framework/Core/include/Framework/AlgorithmSpec.h

@@ -81,7 +81,7 @@ struct AlgorithmSpec {
 
 /// Helper class for an algorithm which is loaded as a plugin.
 struct AlgorithmPlugin {
-  virtual AlgorithmSpec create() = 0;
+  virtual AlgorithmSpec create(ConfigContext const&) = 0;
 };
 // Allow fetching inputs from the context using a string literal.
 template <StringLiteral lit, typename T>

Framework/Core/include/Framework/PluginManager.h

@@ -51,8 +51,9 @@ struct PluginManager {
   /// the DPLPluginHandle provided by the library.
   static void load(std::vector<PluginInfo>& infos, const char* dso, std::function<void(DPLPluginHandle*)>& onSuccess);
   /// Load an called @plugin from a library called @a library and
-  /// return the associtated AlgorithmSpec.
-  static auto loadAlgorithmFromPlugin(std::string library, std::string plugin) -> AlgorithmSpec;
+  /// @return the associated AlgorithmSpec.
+  /// The config @a context can be used to determine the workflow options which affect such plugin.
+  static auto loadAlgorithmFromPlugin(std::string library, std::string plugin, ConfigContext const& context) -> AlgorithmSpec;
   /// Wrap an algorithm with some lambda @wrapper which will be called
   /// with the original callback and the ProcessingContext.
   static auto wrapAlgorithm(AlgorithmSpec const& spec, WrapperProcessCallback&& wrapper) -> AlgorithmSpec;

Framework/Core/include/Framework/runDataProcessing.h

@@ -197,7 +197,6 @@ int mainNoCatch(int argc, char** argv)
   for (auto& extra : extraOptions) {
     workflowOptions.push_back(extra);
   }
-  workflowOptionsRegistry.loadExtra(extraOptions);
 
   ConfigContext configContext(workflowOptionsRegistry, argc, argv);
   o2::framework::WorkflowSpec specs = defineDataProcessing(configContext);

Framework/Core/src/ConfigParamDiscovery.cxx

@@ -25,9 +25,9 @@ namespace o2::framework
 std::vector<ConfigParamSpec> ConfigParamDiscovery::discover(ConfigParamRegistry& registry, int argc, char** argv)
 {
   std::vector<char const*> capabilitiesSpecs = {
+    "O2Framework:DiscoverAODOptionsInCommandLineCapability",
     "O2Framework:DiscoverMetadataInAODCapability",
     "O2Framework:DiscoverMetadataInCommandLineCapability",
-    "O2Framework:DiscoverAODOptionsInCommandLineCapability",
   };
 
   // Load all the requested plugins and discover what we can do.
@@ -75,6 +75,7 @@ std::vector<ConfigParamSpec> ConfigParamDiscovery::discover(ConfigParamRegistry&
     for (auto& extra : extras) {
       result.push_back(extra);
     }
+    registry.loadExtra(extras);
   }
   return result;
 }

Framework/Core/src/Plugin.cxx

@@ -57,6 +57,10 @@ auto lookForCommandLineAODOptions = [](ConfigParamRegistry& registry, int argc,
       O2_SIGNPOST_EVENT_EMIT(capabilities, sid, "DiscoverAODOptionsInCommandLineCapability", "AOD options found in arguments. Populating from them.");
       return true;
     }
+    if (arg.starts_with("--aod-parent-base-path-replacement")) {
+      O2_SIGNPOST_EVENT_EMIT(capabilities, sid, "DiscoverAODOptionsInCommandLineCapability", "AOD options found in arguments. Populating from them.");
+      return true;
+    }
   }
   return false;
 };
@@ -137,7 +141,7 @@ struct DiscoverAODOptionsInCommandLine : o2::framework::ConfigDiscoveryPlugin {
         bool injectOption = true;
         for (size_t i = 0; i < argc; i++) {
           std::string_view arg = argv[i];
-          if (!arg.starts_with("--aod-writer-")) {
+          if (!arg.starts_with("--aod-writer-") && arg != "--aod-parent-base-path-replacement") {
             continue;
           }
           std::string key = arg.data() + 2;
@@ -149,6 +153,9 @@ struct DiscoverAODOptionsInCommandLine : o2::framework::ConfigDiscoveryPlugin {
             results.push_back(ConfigParamSpec{"aod-writer-compression", VariantType::Int, numericValue, {"AOD Compression options"}});
             injectOption = false;
           }
+          if (key == "aod-parent-base-path-replacement") {
+            results.push_back(ConfigParamSpec{"aod-parent-base-path-replacement", VariantType::String, value, {R"(Replace base path of parent files. Syntax: FROM;TO. E.g. "alien:///path/in/alien;/local/path". Enclose in "" on the command line.)"}});
+          }
         }
         if (injectOption) {
           results.push_back(ConfigParamSpec{"aod-writer-compression", VariantType::Int, 505, {"AOD Compression options"}});

Framework/Core/src/PluginManager.cxx

@@ -101,10 +101,10 @@ void PluginManager::load(std::vector<PluginInfo>& libs, const char* dso, std::fu
   onSuccess(pluginInstance);
 }
 
-auto PluginManager::loadAlgorithmFromPlugin(std::string library, std::string plugin) -> AlgorithmSpec
+auto PluginManager::loadAlgorithmFromPlugin(std::string library, std::string plugin, ConfigContext const& context) -> AlgorithmSpec
 {
   std::shared_ptr<AlgorithmSpec> algorithm{nullptr};
-  return AlgorithmSpec{[algorithm, library, plugin](InitContext& ic) mutable -> AlgorithmSpec::ProcessCallback {
+  return AlgorithmSpec{[algorithm, library, plugin, &context](InitContext& ic) mutable -> AlgorithmSpec::ProcessCallback {
     if (algorithm.get()) {
       return algorithm->onInit(ic);
     }
@@ -134,7 +134,7 @@ auto PluginManager::loadAlgorithmFromPlugin(std::string library, std::string plu
     if (!creator) {
       LOGP(fatal, "Could not find the {} plugin in {}.", plugin, libName);
     }
-    algorithm = std::make_shared<AlgorithmSpec>(creator->create());
+    algorithm = std::make_shared<AlgorithmSpec>(creator->create(context));
     return algorithm->onInit(ic);
   }};
 };

Framework/Core/src/WorkflowHelpers.cxx

@@ -217,7 +217,6 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
                 ConfigParamSpec{"aod-max-io-rate", VariantType::Float, 0.f, {"Maximum I/O rate in MB/s"}},
                 ConfigParamSpec{"aod-reader-json", VariantType::String, {"json configuration file"}},
                 ConfigParamSpec{"aod-parent-access-level", VariantType::String, {"Allow parent file access up to specified level. Default: no (0)"}},
-                ConfigParamSpec{"aod-parent-base-path-replacement", VariantType::String, {R"(Replace base path of parent files. Syntax: FROM;TO. E.g. "alien:///path/in/alien;/local/path". Enclose in "" on the command line.)"}},
                 ConfigParamSpec{"time-limit", VariantType::Int64, 0ll, {"Maximum run time limit in seconds"}},
                 ConfigParamSpec{"orbit-offset-enumeration", VariantType::Int64, 0ll, {"initial value for the orbit"}},
                 ConfigParamSpec{"orbit-multiplier-enumeration", VariantType::Int64, 0ll, {"multiplier to get the orbit from the counter"}},
@@ -432,7 +431,7 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
     auto mctracks2aod = std::find_if(workflow.begin(), workflow.end(), [](auto const& x) { return x.name == "mctracks-to-aod"; });
     if (mctracks2aod == workflow.end()) {
       // add normal reader
-      auto&& algo = PluginManager::loadAlgorithmFromPlugin("O2FrameworkAnalysisSupport", "ROOTFileReader");
+      auto&& algo = PluginManager::loadAlgorithmFromPlugin("O2FrameworkAnalysisSupport", "ROOTFileReader", ctx);
       if (internalRateLimiting) {
         aodReader.algorithm = CommonDataProcessors::wrapWithRateLimiting(algo);
       } else {
@@ -520,7 +519,7 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
     }
 
     // Load the CCDB backend from the plugin
-    ccdbBackend.algorithm = PluginManager::loadAlgorithmFromPlugin("O2FrameworkCCDBSupport", "CCDBFetcherPlugin");
+    ccdbBackend.algorithm = PluginManager::loadAlgorithmFromPlugin("O2FrameworkCCDBSupport", "CCDBFetcherPlugin", ctx);
     extraSpecs.push_back(ccdbBackend);
   } else {
     // If there is no CCDB requested, but we still ask for a FLP/DISTSUBTIMEFRAME/0xccdb