Fixes from codex: leads to aod-writer waiting for inputs

Author: shahor02

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<InputSpec> const&, ServiceRegistryRef& ref) -> CompletionPolicy::CompletionOp {
+  auto callbackFull = [](InputSpan const& inputs, std::vector<InputSpec> const& specs, ServiceRegistryRef& ref) -> CompletionPolicy::CompletionOp {
+    assert(inputs.size() == specs.size());
     auto& decongestionService = ref.get<DecongestionService>();
     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<o2::framework::DataProcessingHeader*>(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;
   };

Framework/Core/src/DataProcessingDevice.cxx

@@ -2263,6 +2263,17 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v
     }
   };
 
+  auto cleanSporadic = [&currentSetOfInputs](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<DeviceState>();
       auto& spec = ref.get<DeviceSpec const>();
       auto& streamContext = ref.get<StreamContext>();
@@ -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<DataAllocator>();
           allocator.make<int>(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<CallbackService>().call<CallbackService::Id::DataConsumed>(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<TimesliceIndex>();

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<DataOutputDirector> 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"});
   }
 }
 

Framework/Core/test/test_WorkflowHelpers.cxx

@@ -20,6 +20,7 @@
 #include <algorithm>
 #include <memory>
 #include <list>
+#include <string_view>
 
 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{