consolidate writer configuration between injection and adjusting the topology

Author: aalkin

Framework/Core/src/ArrowSupport.cxx

@@ -691,33 +691,8 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
         }
       }
 
-      // replace writer as some outputs may have become dangling and some are now consumed
-      auto [outputsInputs, isDangling] = WorkflowHelpers::analyzeOutputs(workflow);
-
-      // create DataOutputDescriptor
-      std::shared_ptr<DataOutputDirector> dod = AnalysisSupportHelpers::getDataOutputDirector(ctx);
-
-      // select outputs of type AOD which need to be saved
-      // ATTENTION: if there are dangling outputs the getGlobalAODSink
-      // has to be created in any case!
-      dec.outputsInputsAOD.clear();
-
-      for (auto ii = 0u; ii < outputsInputs.size(); ii++) {
-        if (DataSpecUtils::partialMatch(outputsInputs[ii], extendedAODOrigins)) {
-          auto ds = dod->getDataOutputDescriptors(outputsInputs[ii]);
-          if (!ds.empty() || isDangling[ii]) {
-            dec.outputsInputsAOD.emplace_back(outputsInputs[ii]);
-          }
-        }
-      }
+      WorkflowHelpers::injectAODWriter(workflow, ctx);
 
-      // file sink for any AOD output
-      if (!dec.outputsInputsAOD.empty()) {
-        // add TFNumber and TFFilename as input to the writer
-        dec.outputsInputsAOD.emplace_back("tfn", "TFN", "TFNumber");
-        dec.outputsInputsAOD.emplace_back("tff", "TFF", "TFFilename");
-        workflow.push_back(AnalysisSupportHelpers::getGlobalAODSink(ctx));
-      }
       // Move the dummy sink at the end, if needed
       for (size_t i = 0; i < workflow.size(); ++i) {
         if (workflow[i].name == "internal-dpl-injected-dummy-sink") {

Framework/Core/src/WorkflowHelpers.cxx

@@ -532,43 +532,7 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
   workflow.insert(workflow.end(), extraSpecs.begin(), extraSpecs.end());
   extraSpecs.clear();
 
-  /// Analyze all ouputs
-  auto [outputsInputsTmp, isDanglingTmp] = analyzeOutputs(workflow);
-  dec.isDangling = isDanglingTmp;
-  dec.outputsInputs = outputsInputsTmp;
-
-  // create DataOutputDescriptor
-  std::shared_ptr<DataOutputDirector> dod = AnalysisSupportHelpers::getDataOutputDirector(ctx);
-
-  // select outputs of type AOD which need to be saved
-  // ATTENTION: if there are dangling outputs the getGlobalAODSink
-  // has to be created in any case!
-  for (auto ii = 0u; ii < dec.outputsInputs.size(); ii++) {
-    if (DataSpecUtils::partialMatch(dec.outputsInputs[ii], extendedAODOrigins)) {
-      auto ds = dod->getDataOutputDescriptors(dec.outputsInputs[ii]);
-      if (ds.size() > 0 || dec.isDangling[ii]) {
-        dec.outputsInputsAOD.emplace_back(dec.outputsInputs[ii]);
-      }
-    }
-  }
-
-  // file sink for any AOD output
-  if (dec.outputsInputsAOD.size() > 0) {
-    // add TFNumber and TFFilename as input to the writer
-    dec.outputsInputsAOD.emplace_back(InputSpec{"tfn", "TFN", "TFNumber"});
-    dec.outputsInputsAOD.emplace_back(InputSpec{"tff", "TFF", "TFFilename"});
-    auto fileSink = AnalysisSupportHelpers::getGlobalAODSink(ctx);
-    extraSpecs.push_back(fileSink);
-
-    auto it = std::find_if(dec.outputsInputs.begin(), dec.outputsInputs.end(), [](InputSpec& spec) -> bool {
-      return DataSpecUtils::partialMatch(spec, o2::header::DataOrigin("TFN"));
-    });
-    size_t ii = std::distance(dec.outputsInputs.begin(), it);
-    dec.isDangling[ii] = false;
-  }
-
-  workflow.insert(workflow.end(), extraSpecs.begin(), extraSpecs.end());
-  extraSpecs.clear();
+  injectAODWriter(workflow, ctx);
 
   // Select dangling outputs which are not of type AOD
   std::vector<InputSpec> redirectedOutputsInputs;
@@ -712,6 +676,41 @@ void WorkflowHelpers::adjustTopology(WorkflowSpec& workflow, ConfigContext const
   }
 }
 
+void WorkflowHelpers::injectAODWriter(WorkflowSpec& workflow, ConfigContext const& ctx)
+{
+  auto& dec = ctx.services().get<DanglingEdgesContext>();
+  /// Analyze all ouputs
+  std::tie(dec.outputsInputs, dec.isDangling) = analyzeOutputs(workflow);
+
+  // create DataOutputDescriptor
+  std::shared_ptr<DataOutputDirector> dod = AnalysisSupportHelpers::getDataOutputDirector(ctx);
+
+  // 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], extendedAODOrigins)) {
+      auto ds = dod->getDataOutputDescriptors(dec.outputsInputs[ii]);
+      if (ds.size() > 0 || dec.isDangling[ii]) {
+        dec.outputsInputsAOD.emplace_back(dec.outputsInputs[ii]);
+      }
+    }
+  }
+
+  // file sink for any AOD output
+  if (dec.outputsInputsAOD.size() > 0) {
+    // add TFNumber and TFFilename as input to the writer
+    DataSpecUtils::updateInputList(dec.outputsInputsAOD, InputSpec{"tfn", "TFN", "TFNumber"});
+    DataSpecUtils::updateInputList(dec.outputsInputsAOD, InputSpec{"tff", "TFF", "TFFilename"});
+    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;
+  }
+}
+
 void WorkflowHelpers::constructGraph(const WorkflowSpec& workflow,
                                      std::vector<DeviceConnectionEdge>& logicalEdges,
                                      std::vector<OutputSpec>& outputs,

Framework/Core/src/WorkflowHelpers.h

@@ -182,6 +182,9 @@ struct WorkflowHelpers {
   // @a ctx the context for the configuration phase
   static void injectServiceDevices(WorkflowSpec& workflow, ConfigContext& ctx);
 
+  // Function to correctly add AOD writer
+  static void injectAODWriter(WorkflowSpec& workflow, ConfigContext const& ctx);
+
   // Final adjustments to @a workflow after service devices have been injected.
   static void adjustTopology(WorkflowSpec& workflow, ConfigContext const& ctx);