Propagate dangling edges context to init context and delay algo loading

Author: aalkin

Framework/AnalysisSupport/src/AODReaderHelpers.cxx

@@ -79,12 +79,12 @@ struct Buildable {
 
 } // namespace
 
-AlgorithmSpec AODReaderHelpers::indexBuilderCallback(ConfigContext const& ctx)
+AlgorithmSpec AODReaderHelpers::indexBuilderCallback(ConfigContext const& /*ctx*/)
 {
-  auto& ac = ctx.services().get<DanglingEdgesContext>();
-  return AlgorithmSpec::InitCallback{[requested = ac.requestedIDXs](InitContext& /*ic*/) {
+  return AlgorithmSpec::InitCallback{[](InitContext& ic) {
+    auto const& requested = ic.services().get<DanglingEdgesContext>().requestedIDXs;
     std::vector<Buildable> buildables;
-    for (auto& i : requested) {
+    for (auto const& i : requested) {
       buildables.emplace_back(i);
     }
     std::vector<Builder> builders;
@@ -181,12 +181,12 @@ struct Spawnable {
 
 } // namespace
 
-AlgorithmSpec AODReaderHelpers::aodSpawnerCallback(ConfigContext const& ctx)
+AlgorithmSpec AODReaderHelpers::aodSpawnerCallback(ConfigContext const& /*ctx*/)
 {
-  auto& ac = ctx.services().get<DanglingEdgesContext>();
-  return AlgorithmSpec::InitCallback{[requested = ac.spawnerInputs](InitContext& /*ic*/) {
+  return AlgorithmSpec::InitCallback{[](InitContext& ic) {
+    auto const& requested = ic.services().get<DanglingEdgesContext>().spawnerInputs;
     std::vector<Spawnable> spawnables;
-    for (auto& i : requested) {
+    for (auto const& i : requested) {
       spawnables.emplace_back(i);
     }
     std::vector<Spawner> spawners;

Framework/AnalysisSupport/src/AODReaderHelpers.h

@@ -20,8 +20,8 @@ namespace o2::framework::readers
 
 struct AODReaderHelpers {
   static AlgorithmSpec rootFileReaderCallback();
-  static AlgorithmSpec aodSpawnerCallback(ConfigContext const& ctx);
-  static AlgorithmSpec indexBuilderCallback(ConfigContext const& ctx);
+  static AlgorithmSpec aodSpawnerCallback(ConfigContext const& /*ctx*/);
+  static AlgorithmSpec indexBuilderCallback(ConfigContext const& /*ctx*/);
 };
 
 } // namespace o2::framework::readers

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

@@ -156,6 +156,7 @@ int defaultConditionQueryRateMultiplier()
 
 void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext& ctx)
 {
+  int rateLimitingIPCID = std::stoi(ctx.options().get<std::string>("timeframes-rate-limit-ipcid"));
   DataProcessorSpec ccdbBackend{
     .name = "internal-dpl-ccdb-backend",
     .outputs = {},
@@ -230,23 +231,6 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
                 ConfigParamSpec{"step-value-enumeration", VariantType::Int64, 1ll, {"step between one value and the other"}}},
     .requiredServices = CommonServices::defaultServices("O2FrameworkAnalysisSupport:RunSummary")};
 
-  // AOD reader can be rate limited
-  int rateLimitingIPCID = std::stoi(ctx.options().get<std::string>("timeframes-rate-limit-ipcid"));
-  std::string rateLimitingChannelConfigInput;
-  std::string rateLimitingChannelConfigOutput;
-  bool internalRateLimiting = false;
-
-  // In case we have rate-limiting requested, any device without an input will get one on the special
-  // "DPL/RATE" message.
-  if (rateLimitingIPCID >= 0) {
-    rateLimitingChannelConfigInput = fmt::format("name=metric-feedback,type=pull,method=connect,address=ipc://{}metric-feedback-{},transport=shmem,rateLogging=0",
-                                                 ChannelSpecHelpers::defaultIPCFolder(), rateLimitingIPCID);
-    rateLimitingChannelConfigOutput = fmt::format("name=metric-feedback,type=push,method=bind,address=ipc://{}metric-feedback-{},transport=shmem,rateLogging=0",
-                                                  ChannelSpecHelpers::defaultIPCFolder(), rateLimitingIPCID);
-    internalRateLimiting = true;
-    aodReader.options.emplace_back(ConfigParamSpec{"channel-config", VariantType::String, rateLimitingChannelConfigInput, {"how many timeframes can be in flight at the same time"}});
-  }
-
   ctx.services().registerService(ServiceRegistryHelpers::handleForService<DanglingEdgesContext>(new DanglingEdgesContext));
   auto& dec = ctx.services().get<DanglingEdgesContext>();
 
@@ -274,7 +258,7 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
     // A timeframeSink consumes timeframes without creating new
     // timeframe data.
     bool timeframeSink = hasTimeframeInputs && !hasTimeframeOutputs;
-    if (std::stoi(ctx.options().get<std::string>("timeframes-rate-limit-ipcid")) != -1) {
+    if (rateLimitingIPCID != -1) {
       if (timeframeSink && processor.name.find("internal-dpl-injected-dummy-sink") == std::string::npos) {
         O2_SIGNPOST_ID_GENERATE(sid, workflow_helpers);
         uint32_t hash = runtime_hash(processor.name.c_str());
@@ -384,7 +368,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 +384,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 +415,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 +430,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 +514,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);
   }
 
@@ -637,6 +615,10 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
       extraSpecs.push_back(CommonDataProcessors::getScheduledDummySink(ignored));
     } else {
       O2_SIGNPOST_EVENT_EMIT(workflow_helpers, sid, "injectServiceDevices", "Injecting rate limited dummy sink");
+      std::string rateLimitingChannelConfigOutput;
+      if (rateLimitingIPCID != -1) {
+        rateLimitingChannelConfigOutput = fmt::format("name=metric-feedback,type=push,method=bind,address=ipc://{}metric-feedback-{},transport=shmem,rateLogging=0", ChannelSpecHelpers::defaultIPCFolder(), rateLimitingIPCID);
+      }
       extraSpecs.push_back(CommonDataProcessors::getDummySink(ignored, rateLimitingChannelConfigOutput));
     }
   }

Framework/Core/src/runDataProcessing.cxx

@@ -9,6 +9,7 @@
 // granted to it by virtue of its status as an Intergovernmental Organization
 // or submit itself to any jurisdiction.
 #include <memory>
+#include "Framework/DanglingEdgesContext.h"
 #include "Framework/TopologyPolicyHelpers.h"
 #define BOOST_BIND_GLOBAL_PLACEHOLDERS
 #include <stdexcept>
@@ -1016,6 +1017,7 @@ void doDefaultWorkflowTerminationHook()
 }
 
 int doChild(int argc, char** argv, ServiceRegistry& serviceRegistry,
+            DanglingEdgesContext& danglingEdgesContext,
             RunningWorkflowInfo const& runningWorkflow,
             RunningDeviceRef ref,
             DriverConfig const& driverConfig,
@@ -1078,6 +1080,7 @@ int doChild(int argc, char** argv, ServiceRegistry& serviceRegistry,
                                      &spec,
                                      &quotaEvaluator,
                                      &serviceRegistry,
+                                     &danglingEdgesContext,
                                      &deviceState,
                                      &deviceProxy,
                                      &processingPolicies,
@@ -1101,6 +1104,7 @@ int doChild(int argc, char** argv, ServiceRegistry& serviceRegistry,
     serviceRef.registerService(ServiceRegistryHelpers::handleForService<RunningWorkflowInfo const>(&runningWorkflow));
     serviceRef.registerService(ServiceRegistryHelpers::handleForService<DeviceContext>(deviceContext.get()));
     serviceRef.registerService(ServiceRegistryHelpers::handleForService<DriverConfig const>(&driverConfig));
+    serviceRef.registerService(ServiceRegistryHelpers::handleForService<DanglingEdgesContext>(&danglingEdgesContext));
 
     auto device = std::make_unique<DataProcessingDevice>(ref, serviceRegistry);
 
@@ -1953,6 +1957,7 @@ int runStateMachine(DataProcessorSpecs const& workflow,
           if (runningWorkflow.devices[di].id == frameworkId) {
             return doChild(driverInfo.argc, driverInfo.argv,
                            serviceRegistry,
+                           driverInfo.configContext->services().get<DanglingEdgesContext>(),
                            runningWorkflow, ref,
                            driverConfig,
                            driverInfo.processingPolicies,