delayed algo loading

Author: aalkin

Framework/Core/src/ArrowSupport.cxx

@@ -13,6 +13,7 @@
 #include "Framework/ArrowContext.h"
 #include "Framework/ArrowTableSlicingCache.h"
 #include "Framework/DataProcessor.h"
+#include "Framework/CommonDataProcessors.h"
 #include "Framework/DataProcessingStats.h"
 #include "Framework/ServiceRegistry.h"
 #include "Framework/ConfigContext.h"
@@ -609,9 +610,9 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
         // recreate inputs and outputs
         builder->inputs.clear();
         builder->outputs.clear();
-        // replace AlgorithmSpec
-        //  FIXME: it should be made more generic, so it does not need replacement...
-        builder->algorithm = PluginManager::loadAlgorithmFromPlugin("O2FrameworkOnDemandTablesSupport", "IndexTableBuilder", ctx); // readers::AODReaderHelpers::indexBuilderCallback(ctx);
+
+        // load real AlgorithmSpec before deployment
+        builder->algorithm = PluginManager::loadAlgorithmFromPlugin("O2FrameworkOnDemandTablesSupport", "IndexTableBuilder", ctx);
         AnalysisSupportHelpers::addMissingOutputsToBuilder(dec.requestedIDXs, dec.requestedAODs, dec.requestedDYNs, *builder);
       }
 
@@ -634,10 +635,10 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
         // recreate inputs and outputs
         spawner->outputs.clear();
         spawner->inputs.clear();
-        AnalysisSupportHelpers::addMissingOutputsToSpawner({}, dec.spawnerInputs, dec.requestedAODs, *spawner);
-        // replace AlgorithmSpec
-        // FIXME: it should be made more generic, so it does not need replacement...
+
+        // load real AlgorithmSpec before deployment
         spawner->algorithm = PluginManager::loadAlgorithmFromPlugin("O2FrameworkOnDemandTablesSupport", "ExtendedTableSpawner", ctx);
+        AnalysisSupportHelpers::addMissingOutputsToSpawner({}, dec.spawnerInputs, dec.requestedAODs, *spawner);
       }
 
       if (analysisCCDB != workflow.end()) {
@@ -654,8 +655,7 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
         // recreate inputs and outputs
         analysisCCDB->outputs.clear();
         analysisCCDB->inputs.clear();
-        // replace AlgorithmSpec
-        // FIXME: it should be made more generic, so it does not need replacement...
+        // load real AlgorithmSpec before deployment
         // FIXME how can I make the lookup depend on DYN tables as well??
         analysisCCDB->algorithm = PluginManager::loadAlgorithmFromPlugin("O2FrameworkCCDBSupport", "AnalysisCCDBFetcherPlugin", ctx);
         AnalysisSupportHelpers::addMissingOutputsToBuilder(dec.analysisCCDBInputs, dec.requestedAODs, dec.requestedDYNs, *analysisCCDB);
@@ -682,6 +682,10 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
         if (reader->outputs.empty()) {
           // nothing to read
           workflow.erase(reader);
+        } else {
+          // load reader algorithm before deployment
+          auto&& algo = PluginManager::loadAlgorithmFromPlugin("O2FrameworkAnalysisSupport", "ROOTFileReader", ctx);
+          reader->algorithm = CommonDataProcessors::wrapWithTimesliceConsumption(algo);
         }
       }
 

Framework/Core/src/WorkflowHelpers.cxx

@@ -384,7 +384,7 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
     "internal-dpl-aod-index-builder",
     {},
     {},
-    PluginManager::loadAlgorithmFromPlugin("O2FrameworkOnDemandTablesSupport", "IndexTableBuilder", ctx), // readers::AODReaderHelpers::indexBuilderCallback(ctx),
+    AlgorithmSpec::dummyAlgorithm(), // real algorithm will be set in adjustTopology
     {}};
   AnalysisSupportHelpers::addMissingOutputsToBuilder(dec.requestedIDXs, dec.requestedAODs, dec.requestedDYNs, indexBuilder);
 
@@ -400,7 +400,7 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
     "internal-dpl-aod-spawner",
     {},
     {},
-    PluginManager::loadAlgorithmFromPlugin("O2FrameworkOnDemandTablesSupport", "ExtendedTableSpawner", ctx), // readers::AODReaderHelpers::aodSpawnerCallback(ctx),
+    AlgorithmSpec::dummyAlgorithm(), // real algorithm will be set in adjustTopology
     {}};
   AnalysisSupportHelpers::addMissingOutputsToSpawner({}, dec.spawnerInputs, dec.requestedAODs, aodSpawner);
   AnalysisSupportHelpers::addMissingOutputsToReader(dec.providedAODs, dec.requestedAODs, aodReader);
@@ -431,13 +431,11 @@ 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", ctx);
-      aodReader.algorithm = CommonDataProcessors::wrapWithTimesliceConsumption(algo);
       aodReader.outputs.emplace_back(OutputSpec{"TFN", "TFNumber"});
       aodReader.outputs.emplace_back(OutputSpec{"TFF", "TFFilename"});
     } else {
-      // AODs are being injected on-the-fly, add dummy reader
-      auto algo = AlgorithmSpec{
+      // AODs are being injected on-the-fly, add error-handler reader
+      aodReader.algorithm = AlgorithmSpec{
         adaptStateful(
           [outputs = aodReader.outputs](DeviceSpec const&) {
             LOGP(warn, "Workflow with injected AODs has unsatisfied inputs:");
@@ -448,7 +446,6 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
             // to ensure the output type for adaptStateful
             return adaptStateless([](DataAllocator&) {});
           })};
-      aodReader.algorithm = CommonDataProcessors::wrapWithTimesliceConsumption(algo);
     }
     auto concrete = DataSpecUtils::asConcreteDataMatcher(aodReader.inputs[0]);
     timer.outputs.emplace_back(concrete.origin, concrete.description, concrete.subSpec, Lifetime::Enumeration);
@@ -533,9 +530,6 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
 
   // add the Analysys CCDB backend which reads CCDB objects using a provided table
   if (analysisCCDBBackend.outputs.empty() == false) {
-    // add normal reader
-    auto&& algo = PluginManager::loadAlgorithmFromPlugin("O2FrameworkCCDBSupport", "AnalysisCCDBFetcherPlugin", ctx);
-    analysisCCDBBackend.algorithm = algo;
     extraSpecs.push_back(analysisCCDBBackend);
   }