161 LOG(info) <<
"This is not a real device, merely a placeholder for external inputs";
162 LOG(info) <<
"To be hidden / removed at some point.";
173 .
name =
"internal-dpl-ccdb-backend",
176 {
"condition-not-before",
VariantType::Int64, 0ll, {
"do not fetch from CCDB objects created before provide timestamp"}},
177 {
"condition-not-after",
VariantType::Int64, 3385078236000ll, {
"do not fetch from CCDB objects created after the timestamp"}},
178 {
"condition-remap",
VariantType::String,
"", {
"remap condition path in CCDB based on the provided string."}},
181 {
"condition-use-slice-for-prescaling",
VariantType::Int, 0, {
"use TFslice instead of TFcounter to control validation frequency. If > query rate, do not allow TFCounter excursion exceeding it"}},
182 {
"condition-time-tolerance",
VariantType::Int64, 5000ll, {
"prefer creation time if its difference to orbit-derived time exceeds threshold (ms), impose if <0"}},
183 {
"orbit-offset-enumeration",
VariantType::Int64, 0ll, {
"initial value for the orbit"}},
184 {
"orbit-multiplier-enumeration",
VariantType::Int64, 0ll, {
"multiplier to get the orbit from the counter"}},
185 {
"start-value-enumeration",
VariantType::Int64, 0ll, {
"initial value for the enumeration"}},
186 {
"end-value-enumeration",
VariantType::Int64, -1ll, {
"final value for the enumeration"}},
187 {
"step-value-enumeration",
VariantType::Int64, 1ll, {
"step between one value and the other"}}},
190 .
name =
"internal-dpl-aod-ccdb",
195 {
"condition-not-before",
VariantType::Int64, 0ll, {
"do not fetch from CCDB objects created before provide timestamp"}},
196 {
"condition-not-after",
VariantType::Int64, 3385078236000ll, {
"do not fetch from CCDB objects created after the timestamp"}},
197 {
"condition-remap",
VariantType::String,
"", {
"remap condition path in CCDB based on the provided string."}},
200 {
"condition-use-slice-for-prescaling",
VariantType::Int, 0, {
"use TFslice instead of TFcounter to control validation frequency. If > query rate, do not allow TFCounter excursion exceeding it"}},
201 {
"condition-time-tolerance",
VariantType::Int64, 5000ll, {
"prefer creation time if its difference to orbit-derived time exceeds threshold (ms), impose if <0"}},
202 {
"start-value-enumeration",
VariantType::Int64, 0ll, {
"initial value for the enumeration"}},
203 {
"end-value-enumeration",
VariantType::Int64, -1ll, {
"final value for the enumeration"}},
204 {
"step-value-enumeration",
VariantType::Int64, 1ll, {
"step between one value and the other"}}}};
225 auto aodLifetime = Lifetime::Enumeration;
228 .
name =
"internal-dpl-aod-reader",
247 int rateLimitingIPCID = std::stoi(ctx.
options().
get<std::string>(
"timeframes-rate-limit-ipcid"));
248 std::string rateLimitingChannelConfigInput;
249 std::string rateLimitingChannelConfigOutput;
250 bool internalRateLimiting =
false;
254 if (rateLimitingIPCID >= 0) {
255 rateLimitingChannelConfigInput = fmt::format(
"name=metric-feedback,type=pull,method=connect,address=ipc://{}metric-feedback-{},transport=shmem,rateLogging=0",
257 rateLimitingChannelConfigOutput = fmt::format(
"name=metric-feedback,type=push,method=bind,address=ipc://{}metric-feedback-{},transport=shmem,rateLogging=0",
259 internalRateLimiting =
true;
260 aodReader.options.emplace_back(
ConfigParamSpec{
"channel-config",
VariantType::String, rateLimitingChannelConfigInput, {
"how many timeframes can be in flight at the same time"}});
266 std::vector<InputSpec> requestedCCDBs;
267 std::vector<OutputSpec> providedCCDBs;
269 for (
size_t wi = 0; wi < workflow.size(); ++wi) {
270 auto& processor = workflow[wi];
271 auto name = processor.name;
273 ac.outTskMap.push_back({hash,
name});
275 std::string prefix =
"internal-dpl-";
276 if (processor.inputs.empty() && processor.name.compare(0, prefix.size(), prefix) != 0) {
284 bool hasTimeframeInputs =
false;
285 for (
auto& input : processor.inputs) {
286 if (input.lifetime == Lifetime::Timeframe) {
287 hasTimeframeInputs =
true;
291 bool hasTimeframeOutputs =
false;
292 for (
auto&
output : processor.outputs) {
293 if (
output.lifetime == Lifetime::Timeframe) {
294 hasTimeframeOutputs =
true;
300 bool timeframeSink = hasTimeframeInputs && !hasTimeframeOutputs;
301 if (std::stoi(ctx.options().get<std::string>(
"timeframes-rate-limit-ipcid")) != -1) {
302 if (timeframeSink && processor.name.find(
"internal-dpl-injected-dummy-sink") == std::string::npos) {
305 bool hasMatch =
false;
307 for (
auto&
output : processor.outputs) {
309 O2_SIGNPOST_EVENT_EMIT(workflow_helpers, sid,
"output enumeration",
"%{public}s already there in %{public}s",
316 O2_SIGNPOST_EVENT_EMIT(workflow_helpers, sid,
"output enumeration",
"Adding DPL/SUMMARY/%d to %{public}s", hash, processor.name.c_str());
321 bool hasConditionOption =
false;
322 for (
size_t ii = 0; ii < processor.inputs.size(); ++ii) {
323 auto& input = processor.inputs[ii];
324 switch (input.lifetime) {
325 case Lifetime::Timer: {
327 auto hasOption = std::any_of(processor.options.begin(), processor.options.end(), [&input](
auto const& option) { return (option.name ==
"period-" + input.binding); });
328 if (hasOption ==
false) {
329 processor.options.push_back(ConfigParamSpec{
"period-" + input.binding,
VariantType::Int, 1000, {
"period of the timer in milliseconds"}});
331 timer.outputs.emplace_back(
OutputSpec{concrete.origin, concrete.description, concrete.subSpec, Lifetime::Timer});
333 case Lifetime::Signal: {
335 timer.outputs.emplace_back(
OutputSpec{concrete.origin, concrete.description, concrete.subSpec, Lifetime::Signal});
337 case Lifetime::Enumeration: {
339 timer.outputs.emplace_back(
OutputSpec{concrete.origin, concrete.description, concrete.subSpec, Lifetime::Enumeration});
341 case Lifetime::Condition: {
342 for (
auto& option : processor.options) {
343 if (option.name ==
"condition-backend") {
344 hasConditionOption =
true;
348 if (hasConditionOption ==
false) {
350 processor.options.emplace_back(ConfigParamSpec{
"condition-timestamp",
VariantType::Int64, 0ll, {
"Force timestamp for CCDB lookup"}});
351 hasConditionOption =
true;
353 requestedCCDBs.emplace_back(input);
355 case Lifetime::OutOfBand: {
357 auto hasOption = std::any_of(processor.options.begin(), processor.options.end(), [&input](
auto const& option) { return (option.name ==
"out-of-band-channel-name-" + input.binding); });
358 if (hasOption ==
false) {
359 processor.options.push_back(ConfigParamSpec{
"out-of-band-channel-name-" + input.binding,
VariantType::String,
"out-of-band", {
"channel to listen for out of band data"}});
361 timer.outputs.emplace_back(
OutputSpec{concrete.origin, concrete.description, concrete.subSpec, Lifetime::Enumeration});
364 case Lifetime::Transient:
365 case Lifetime::Timeframe:
366 case Lifetime::Optional:
383 std::stable_sort(timer.outputs.begin(), timer.outputs.end(), [](
OutputSpec const&
a,
OutputSpec const&
b) { return *DataSpecUtils::getOptionalSubSpec(a) < *DataSpecUtils::getOptionalSubSpec(b); });
385 for (
auto&
output : processor.outputs) {
387 ac.providedAODs.emplace_back(
output);
389 ac.providedDYNs.emplace_back(
output);
391 ac.providedTIMs.emplace_back(
output);
393 ac.providedOutputObjHist.emplace_back(
output);
394 auto it = std::find_if(ac.outObjHistMap.begin(), ac.outObjHistMap.end(), [&](
auto&&
x) { return x.id == hash; });
395 if (it == ac.outObjHistMap.end()) {
396 ac.outObjHistMap.push_back({hash, {
output.binding.value}});
398 it->bindings.push_back(
output.binding.value);
401 if (
output.lifetime == Lifetime::Condition) {
402 providedCCDBs.push_back(
output);
409 std::sort(ac.requestedDYNs.begin(), ac.requestedDYNs.end(), inputSpecLessThan);
410 std::sort(ac.requestedTIMs.begin(), ac.requestedTIMs.end(), inputSpecLessThan);
411 std::sort(ac.providedDYNs.begin(), ac.providedDYNs.end(), outputSpecLessThan);
412 std::sort(ac.providedTIMs.begin(), ac.providedTIMs.end(), outputSpecLessThan);
415 "internal-dpl-aod-index-builder",
422 ac.requestedTIMs | views::filter_not_matching(ac.providedTIMs) | sinks::append_to{ac.analysisCCDBInputs};
424 if (deploymentMode != DeploymentMode::OnlineDDS && deploymentMode != DeploymentMode::OnlineECS) {
428 for (
auto& input : ac.requestedDYNs) {
429 if (std::none_of(ac.providedDYNs.begin(), ac.providedDYNs.end(), [&input](
auto const&
x) { return DataSpecUtils::match(input, x); })) {
430 ac.spawnerInputs.emplace_back(input);
435 "internal-dpl-aod-spawner",
445 std::vector<DataProcessorSpec> extraSpecs;
447 if (transientStore.outputs.empty() ==
false) {
448 extraSpecs.push_back(transientStore);
450 if (qaStore.outputs.empty() ==
false) {
451 extraSpecs.push_back(qaStore);
454 if (aodSpawner.outputs.empty() ==
false) {
455 extraSpecs.push_back(
timePipeline(aodSpawner, ctx.options().get<int64_t>(
"spawners")));
458 if (indexBuilder.outputs.empty() ==
false) {
459 extraSpecs.push_back(indexBuilder);
463 if (aodReader.outputs.empty() ==
false) {
464 auto mctracks2aod = std::find_if(workflow.begin(), workflow.end(), [](
auto const&
x) { return x.name ==
"mctracks-to-aod"; });
465 if (mctracks2aod == workflow.end()) {
468 if (internalRateLimiting) {
471 aodReader.algorithm = algo;
473 aodReader.outputs.emplace_back(
OutputSpec{
"TFN",
"TFNumber"});
474 aodReader.outputs.emplace_back(
OutputSpec{
"TFF",
"TFFilename"});
477 aodReader.algorithm = AlgorithmSpec{
479 [outputs = aodReader.outputs](DeviceSpec
const&) {
480 LOGP(warn,
"Workflow with injected AODs has unsatisfied inputs:");
481 for (auto const& output : outputs) {
482 LOGP(warn,
" {}", DataSpecUtils::describe(output));
484 LOGP(fatal,
"Stopping.");
490 timer.outputs.emplace_back(concrete.origin, concrete.description, concrete.subSpec, Lifetime::Enumeration);
491 extraSpecs.push_back(
timePipeline(aodReader, ctx.options().get<int64_t>(
"readers")));
495 if (ccdbBackend.outputs.empty() ==
false) {
496 ccdbBackend.outputs.push_back(
OutputSpec{
"CTP",
"OrbitReset", 0});
497 InputSpec matcher{
"dstf",
"FLP",
"DISTSUBTIMEFRAME", 0xccdb};
498 bool providesDISTSTF =
false;
501 for (
auto& dp : workflow) {
502 for (
auto&
output : dp.outputs) {
504 providesDISTSTF =
true;
509 if (providesDISTSTF) {
519 if (aodReader.outputs.empty() ==
false) {
520 ccdbBackend.inputs.push_back(
InputSpec{
"tfn",
"TFN",
"TFNumber"});
521 }
else if (providesDISTSTF) {
522 ccdbBackend.inputs.push_back(
InputSpec{
"tfn", dstf, Lifetime::Timeframe});
528 int enumCandidate = -1;
529 int timerCandidate = -1;
530 for (
size_t wi = 0; wi < workflow.size(); wi++) {
531 auto& dp = workflow[wi];
532 if (dp.inputs.size() != 1) {
535 auto lifetime = dp.inputs[0].lifetime;
536 if (lifetime == Lifetime::Enumeration && (enumCandidate == -1 || workflow[enumCandidate].
name > dp.name)) {
539 if (lifetime == Lifetime::Timer && (timerCandidate == -1 || workflow[timerCandidate].
name > dp.name)) {
543 if (enumCandidate != -1) {
544 auto& dp = workflow[enumCandidate];
546 ccdbBackend.inputs.push_back(
InputSpec{
"tfn", dstf, Lifetime::Timeframe});
547 }
else if (timerCandidate != -1) {
548 auto& dp = workflow[timerCandidate];
550 ccdbBackend.inputs.push_back(
InputSpec{{
"tfn"}, dstf, Lifetime::Timeframe});
556 extraSpecs.push_back(ccdbBackend);
561 bool requiresDISTSUBTIMEFRAME =
false;
562 for (
auto& dp : workflow) {
563 for (
auto& input : dp.inputs) {
565 requiresDISTSUBTIMEFRAME =
true;
570 if (requiresDISTSUBTIMEFRAME) {
575 int enumCandidate = -1;
576 int timerCandidate = -1;
577 for (
size_t wi = 0; wi < workflow.size(); wi++) {
578 auto& dp = workflow[wi];
579 if (dp.inputs.size() != 1) {
582 auto lifetime = dp.inputs[0].lifetime;
583 if (lifetime == Lifetime::Enumeration && (enumCandidate == -1 || workflow[enumCandidate].
name > dp.name)) {
586 if (lifetime == Lifetime::Timer && (timerCandidate == -1 || workflow[timerCandidate].
name > dp.name)) {
590 if (enumCandidate != -1) {
591 auto& dp = workflow[enumCandidate];
593 ccdbBackend.inputs.push_back(
InputSpec{
"tfn", dstf, Lifetime::Timeframe});
594 }
else if (timerCandidate != -1) {
595 auto& dp = workflow[timerCandidate];
597 ccdbBackend.inputs.push_back(
InputSpec{{
"tfn"}, dstf, Lifetime::Timeframe});
604 if (analysisCCDBBackend.outputs.empty() ==
false) {
607 analysisCCDBBackend.algorithm = algo;
608 extraSpecs.push_back(analysisCCDBBackend);
612 if (timer.outputs.empty() ==
false) {
613 extraSpecs.push_back(timer);
618 if (ac.providedOutputObjHist.empty() ==
false) {
620 extraSpecs.push_back(rootSink);
623 workflow.insert(workflow.end(), extraSpecs.begin(), extraSpecs.end());
627 auto [outputsInputsTmp, isDanglingTmp] = analyzeOutputs(workflow);
628 ac.isDangling = isDanglingTmp;
629 ac.outputsInputs = outputsInputsTmp;
637 for (
auto ii = 0u; ii < ac.outputsInputs.size(); ii++) {
639 auto ds = dod->getDataOutputDescriptors(ac.outputsInputs[ii]);
640 if (
ds.size() > 0 || ac.isDangling[ii]) {
641 ac.outputsInputsAOD.emplace_back(ac.outputsInputs[ii]);
647 if (ac.outputsInputsAOD.size() > 0) {
649 ac.outputsInputsAOD.emplace_back(
InputSpec{
"tfn",
"TFN",
"TFNumber"});
650 ac.outputsInputsAOD.emplace_back(
InputSpec{
"tff",
"TFF",
"TFFilename"});
652 extraSpecs.push_back(fileSink);
654 auto it = std::find_if(ac.outputsInputs.begin(), ac.outputsInputs.end(), [](
InputSpec& spec) ->
bool {
655 return DataSpecUtils::partialMatch(spec, o2::header::DataOrigin(
"TFN"));
657 size_t ii = std::distance(ac.outputsInputs.begin(), it);
658 ac.isDangling[ii] =
false;
661 workflow.insert(workflow.end(), extraSpecs.begin(), extraSpecs.end());
665 std::vector<InputSpec> redirectedOutputsInputs;
666 for (
auto ii = 0u; ii < ac.outputsInputs.size(); ii++) {
667 if (ctx.options().get<std::string>(
"forwarding-policy") ==
"none") {
672 if (!ac.isDangling[ii] && (ctx.options().get<std::string>(
"forwarding-policy") !=
"all")) {
679 redirectedOutputsInputs.emplace_back(ac.outputsInputs[ii]);
682 std::vector<InputSpec> unmatched;
683 auto forwardingDestination = ctx.options().get<std::string>(
"forwarding-destination");
684 if (redirectedOutputsInputs.size() > 0 && forwardingDestination ==
"file") {
686 if (unmatched.size() != redirectedOutputsInputs.size()) {
687 extraSpecs.push_back(fileSink);
689 }
else if (redirectedOutputsInputs.size() > 0 && forwardingDestination ==
"fairmq") {
691 extraSpecs.push_back(fairMQSink);
692 }
else if (forwardingDestination !=
"drop") {
693 throw runtime_error_f(
"Unknown forwarding destination %s", forwardingDestination.c_str());
695 if (unmatched.size() > 0 || redirectedOutputsInputs.size() > 0) {
696 std::vector<InputSpec> ignored = unmatched;
697 ignored.insert(ignored.end(), redirectedOutputsInputs.begin(), redirectedOutputsInputs.end());
698 for (
auto& ignoredInput : ignored) {
699 ignoredInput.lifetime = Lifetime::Sporadic;
705 workflow.insert(workflow.end(), extraSpecs.begin(), extraSpecs.end());
795 std::vector<DeviceConnectionEdge>& logicalEdges,
796 std::vector<OutputSpec>& outputs,
797 std::vector<LogicalForwardInfo>& forwardedInputsInfo)
800 if (workflow.empty()) {
805 std::vector<LogicalOutputInfo> availableOutputsInfo;
806 auto const& constOutputs = outputs;
808 std::vector<LogicalOutputInfo> forwards;
812 auto enumerateAvailableOutputs = [&workflow, &outputs, &availableOutputsInfo]() {
814 for (
size_t wi = 0; wi < workflow.size(); ++wi) {
815 auto& producer = workflow[wi];
816 if (producer.outputs.empty()) {
817 O2_SIGNPOST_EVENT_EMIT(workflow_helpers, sid,
"output enumeration",
"No outputs for [%zu] %{public}s", wi, producer.name.c_str());
819 O2_SIGNPOST_START(workflow_helpers, sid,
"output enumeration",
"Enumerating outputs for producer [%zu] %{}s public", wi, producer.name.c_str());
821 for (
size_t oi = 0; oi < producer.outputs.size(); ++oi) {
822 auto& out = producer.outputs[oi];
823 auto uniqueOutputId = outputs.size();
827 outputs.push_back(out);
833 auto errorDueToMissingOutputFor = [&workflow, &constOutputs](
size_t ci,
size_t ii) {
834 auto input = workflow[ci].inputs[ii];
835 std::ostringstream
str;
836 str <<
"No matching output found for "
837 <<
DataSpecUtils::describe(input) <<
" as requested by data processor \"" << workflow[ci].name <<
"\". Candidates:\n";
839 for (
auto&
output : constOutputs) {
843 throw std::runtime_error(
str.str());
857 enumerateAvailableOutputs();
859 std::vector<bool> matches(constOutputs.size());
860 for (
size_t consumer = 0; consumer < workflow.size(); ++consumer) {
862 O2_SIGNPOST_START(workflow_helpers, sid,
"input matching",
"Matching inputs of consumer [%zu] %{}s public", consumer, workflow[consumer].
name.c_str());
863 for (
size_t input = 0; input < workflow[consumer].inputs.size(); ++input) {
865 for (
size_t i = 0;
i < constOutputs.size();
i++) {
874 for (
size_t i = 0;
i < availableOutputsInfo.size();
i++) {
876 if (!matches[availableOutputsInfo[
i].outputGlobalIndex]) {
879 auto* oif = &availableOutputsInfo[
i];
883 auto producer = oif->specIndex;
884 auto uniqueOutputId = oif->outputGlobalIndex;
885 for (
size_t tpi = 0; tpi < workflow[consumer].maxInputTimeslices; ++tpi) {
886 for (
size_t ptpi = 0; ptpi < workflow[producer].maxInputTimeslices; ++ptpi) {
887 O2_SIGNPOST_EVENT_EMIT(workflow_helpers, sid,
"output",
"Adding edge between %{public}s and %{public}s", workflow[consumer].
name.c_str(),
888 workflow[producer].name.c_str());
889 logicalEdges.emplace_back(
DeviceConnectionEdge{producer, consumer, tpi, ptpi, uniqueOutputId, input, oif->forward});
894 oif->enabled =
false;
896 if (forwards.empty()) {
897 errorDueToMissingOutputFor(consumer, input);
899 availableOutputsInfo.erase(std::remove_if(availableOutputsInfo.begin(), availableOutputsInfo.end(), [](
auto& info) { return info.enabled == false; }), availableOutputsInfo.end());
900 for (
auto& forward : forwards) {
901 availableOutputsInfo.push_back(forward);