diff --git a/Framework/Core/src/CompletionPolicyHelpers.cxx b/Framework/Core/src/CompletionPolicyHelpers.cxx index 8c4b38ba3355c..a6240b1165b81 100644 --- a/Framework/Core/src/CompletionPolicyHelpers.cxx +++ b/Framework/Core/src/CompletionPolicyHelpers.cxx @@ -18,10 +18,13 @@ #include "Framework/TimesliceIndex.h" #include "Framework/TimingInfo.h" #include "DecongestionService.h" +#include "Framework/Signpost.h" #include #include +O2_DECLARE_DYNAMIC_LOG(completion); + namespace o2::framework { @@ -108,6 +111,8 @@ CompletionPolicy CompletionPolicyHelpers::consumeWhenAll(const char* name, Compl { auto callback = [](InputSpan const& inputs, std::vector const& specs, ServiceRegistryRef& ref) -> CompletionPolicy::CompletionOp { assert(inputs.size() == specs.size()); + O2_SIGNPOST_ID_GENERATE(sid, completion); + O2_SIGNPOST_START(completion, sid, "consumeWhenAll", "Completion policy invoked"); size_t si = 0; bool missingSporadic = false; @@ -116,15 +121,18 @@ CompletionPolicy CompletionPolicyHelpers::consumeWhenAll(const char* name, Compl assert(si < specs.size()); auto& spec = specs[si++]; if (input.header == nullptr && spec.lifetime != Lifetime::Sporadic) { + O2_SIGNPOST_END(completion, sid, "consumeWhenAll", "Completion policy returned %{public}s due to missing input %lu", "Wait", si); return CompletionPolicy::CompletionOp::Wait; } if (input.header == nullptr && spec.lifetime == Lifetime::Sporadic) { + O2_SIGNPOST_EVENT_EMIT(completion, sid, "consumeWhenAll", "Missing sporadic found for route index %lu", si); missingSporadic = true; } if (input.header != nullptr && currentTimeslice == -1) { auto const* dph = framework::DataRefUtils::getHeader(input); if (dph && !TimingInfo::timesliceIsTimer(dph->startTime)) { currentTimeslice = dph->startTime; + O2_SIGNPOST_EVENT_EMIT(completion, sid, "consumeWhenAll", "currentTimeslice %lu from route index %lu", currentTimeslice, si); } } } @@ -134,8 +142,10 @@ CompletionPolicy CompletionPolicyHelpers::consumeWhenAll(const char* name, Compl auto oldestPossibleTimeslice = timesliceIndex.getOldestPossibleInput().timeslice.value; if (missingSporadic && currentTimeslice >= oldestPossibleTimeslice) { + O2_SIGNPOST_END(completion, sid, "consumeWhenAll", "Completion policy returned %{public}s for timeslice %lu > oldestPossibleTimeslice %lu", "Retry", currentTimeslice, oldestPossibleTimeslice); return CompletionPolicy::CompletionOp::Retry; } + O2_SIGNPOST_END(completion, sid, "consumeWhenAll", "Completion policy returned %{public}s for timeslice %lu <= oldestPossibleTimeslice %lu", "Consume", currentTimeslice, oldestPossibleTimeslice); return CompletionPolicy::CompletionOp::Consume; }; return CompletionPolicy{name, matcher, callback};