160 LOG(info) <<
"This is not a real device, merely a placeholder for external inputs";
161 LOG(info) <<
"To be hidden / removed at some point.";
172 .
name =
"internal-dpl-ccdb-backend",
175 {
"condition-not-before",
VariantType::Int64, 0ll, {
"do not fetch from CCDB objects created before provide timestamp"}},
176 {
"condition-not-after",
VariantType::Int64, 3385078236000ll, {
"do not fetch from CCDB objects created after the timestamp"}},
177 {
"condition-remap",
VariantType::String,
"", {
"remap condition path in CCDB based on the provided string."}},
180 {
"condition-time-tolerance",
VariantType::Int64, 5000ll, {
"prefer creation time if its difference to orbit-derived time exceeds threshold (ms), impose if <0"}},
181 {
"orbit-offset-enumeration",
VariantType::Int64, 0ll, {
"initial value for the orbit"}},
182 {
"orbit-multiplier-enumeration",
VariantType::Int64, 0ll, {
"multiplier to get the orbit from the counter"}},
183 {
"start-value-enumeration",
VariantType::Int64, 0ll, {
"initial value for the enumeration"}},
184 {
"end-value-enumeration",
VariantType::Int64, -1ll, {
"final value for the enumeration"}},
185 {
"step-value-enumeration",
VariantType::Int64, 1ll, {
"step between one value and the other"}}},
207 auto aodLifetime = Lifetime::Enumeration;
210 .
name =
"internal-dpl-aod-reader",
229 int rateLimitingIPCID = std::stoi(ctx.
options().
get<std::string>(
"timeframes-rate-limit-ipcid"));
230 std::string rateLimitingChannelConfigInput;
231 std::string rateLimitingChannelConfigOutput;
232 bool internalRateLimiting =
false;
236 if (rateLimitingIPCID >= 0) {
237 rateLimitingChannelConfigInput = fmt::format(
"name=metric-feedback,type=pull,method=connect,address=ipc://{}metric-feedback-{},transport=shmem,rateLogging=0",
239 rateLimitingChannelConfigOutput = fmt::format(
"name=metric-feedback,type=push,method=bind,address=ipc://{}metric-feedback-{},transport=shmem,rateLogging=0",
241 internalRateLimiting =
true;
242 aodReader.options.emplace_back(
ConfigParamSpec{
"channel-config",
VariantType::String, rateLimitingChannelConfigInput, {
"how many timeframes can be in flight at the same time"}});
248 std::vector<InputSpec> requestedCCDBs;
249 std::vector<OutputSpec> providedCCDBs;
251 for (
size_t wi = 0; wi < workflow.size(); ++wi) {
252 auto& processor = workflow[wi];
253 auto name = processor.name;
255 ac.outTskMap.push_back({hash,
name});
257 std::string prefix =
"internal-dpl-";
258 if (processor.inputs.empty() && processor.name.compare(0, prefix.size(), prefix) != 0) {
266 bool hasTimeframeInputs =
false;
267 for (
auto& input : processor.inputs) {
268 if (input.lifetime == Lifetime::Timeframe) {
269 hasTimeframeInputs =
true;
273 bool hasTimeframeOutputs =
false;
274 for (
auto&
output : processor.outputs) {
275 if (
output.lifetime == Lifetime::Timeframe) {
276 hasTimeframeOutputs =
true;
282 bool timeframeSink = hasTimeframeInputs && !hasTimeframeOutputs;
283 if (std::stoi(ctx.options().get<std::string>(
"timeframes-rate-limit-ipcid")) != -1) {
284 if (timeframeSink && processor.name.find(
"internal-dpl-injected-dummy-sink") == std::string::npos) {
287 bool hasMatch =
false;
289 for (
auto&
output : processor.outputs) {
291 O2_SIGNPOST_EVENT_EMIT(workflow_helpers, sid,
"output enumeration",
"%{public}s already there in %{public}s",
298 O2_SIGNPOST_EVENT_EMIT(workflow_helpers, sid,
"output enumeration",
"Adding DPL/SUMMARY/%d to %{public}s", hash, processor.name.c_str());
303 bool hasConditionOption =
false;
304 for (
size_t ii = 0; ii < processor.inputs.size(); ++ii) {
305 auto& input = processor.inputs[ii];
306 switch (input.lifetime) {
307 case Lifetime::Timer: {
309 auto hasOption = std::any_of(processor.options.begin(), processor.options.end(), [&input](
auto const& option) { return (option.name ==
"period-" + input.binding); });
310 if (hasOption ==
false) {
311 processor.options.push_back(ConfigParamSpec{
"period-" + input.binding,
VariantType::Int, 1000, {
"period of the timer in milliseconds"}});
313 timer.outputs.emplace_back(
OutputSpec{concrete.origin, concrete.description, concrete.subSpec, Lifetime::Timer});
315 case Lifetime::Signal: {
317 timer.outputs.emplace_back(
OutputSpec{concrete.origin, concrete.description, concrete.subSpec, Lifetime::Signal});
319 case Lifetime::Enumeration: {
321 timer.outputs.emplace_back(
OutputSpec{concrete.origin, concrete.description, concrete.subSpec, Lifetime::Enumeration});
323 case Lifetime::Condition: {
324 for (
auto& option : processor.options) {
325 if (option.name ==
"condition-backend") {
326 hasConditionOption =
true;
330 if (hasConditionOption ==
false) {
332 processor.options.emplace_back(ConfigParamSpec{
"condition-timestamp",
VariantType::Int64, 0ll, {
"Force timestamp for CCDB lookup"}});
333 hasConditionOption =
true;
335 requestedCCDBs.emplace_back(input);
337 case Lifetime::OutOfBand: {
339 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); });
340 if (hasOption ==
false) {
341 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"}});
343 timer.outputs.emplace_back(
OutputSpec{concrete.origin, concrete.description, concrete.subSpec, Lifetime::Enumeration});
346 case Lifetime::Transient:
347 case Lifetime::Timeframe:
348 case Lifetime::Optional:
362 std::stable_sort(timer.outputs.begin(), timer.outputs.end(), [](
OutputSpec const&
a,
OutputSpec const&
b) { return *DataSpecUtils::getOptionalSubSpec(a) < *DataSpecUtils::getOptionalSubSpec(b); });
364 for (
auto&
output : processor.outputs) {
366 ac.providedAODs.emplace_back(
output);
368 ac.providedDYNs.emplace_back(
output);
370 ac.providedOutputObjHist.emplace_back(
output);
371 auto it = std::find_if(ac.outObjHistMap.begin(), ac.outObjHistMap.end(), [&](
auto&&
x) { return x.id == hash; });
372 if (it == ac.outObjHistMap.end()) {
373 ac.outObjHistMap.push_back({hash, {
output.binding.value}});
375 it->bindings.push_back(
output.binding.value);
378 if (
output.lifetime == Lifetime::Condition) {
379 providedCCDBs.push_back(
output);
386 std::sort(ac.requestedDYNs.begin(), ac.requestedDYNs.end(), inputSpecLessThan);
387 std::sort(ac.providedDYNs.begin(), ac.providedDYNs.end(), outputSpecLessThan);
388 for (
auto& input : ac.requestedDYNs) {
389 if (std::none_of(ac.providedDYNs.begin(), ac.providedDYNs.end(), [&input](
auto const&
x) { return DataSpecUtils::match(input, x); })) {
390 ac.spawnerInputs.emplace_back(input);
395 "internal-dpl-aod-spawner",
402 "internal-dpl-aod-index-builder",
414 std::vector<DataProcessorSpec> extraSpecs;
416 if (transientStore.outputs.empty() ==
false) {
417 extraSpecs.push_back(transientStore);
419 if (qaStore.outputs.empty() ==
false) {
420 extraSpecs.push_back(qaStore);
423 if (aodSpawner.outputs.empty() ==
false) {
424 extraSpecs.push_back(
timePipeline(aodSpawner, ctx.options().get<int64_t>(
"spawners")));
427 if (indexBuilder.outputs.empty() ==
false) {
428 extraSpecs.push_back(indexBuilder);
432 if (aodReader.outputs.empty() ==
false) {
433 auto mctracks2aod = std::find_if(workflow.begin(), workflow.end(), [](
auto const&
x) { return x.name ==
"mctracks-to-aod"; });
434 if (mctracks2aod == workflow.end()) {
437 if (internalRateLimiting) {
440 aodReader.algorithm = algo;
442 aodReader.outputs.emplace_back(
OutputSpec{
"TFN",
"TFNumber"});
443 aodReader.outputs.emplace_back(
OutputSpec{
"TFF",
"TFFilename"});
446 aodReader.algorithm = AlgorithmSpec{
448 [outputs = aodReader.outputs](DeviceSpec
const&) {
449 LOGP(warn,
"Workflow with injected AODs has unsatisfied inputs:");
450 for (auto const& output : outputs) {
451 LOGP(warn,
" {}", DataSpecUtils::describe(output));
453 LOGP(fatal,
"Stopping.");
459 timer.outputs.emplace_back(concrete.origin, concrete.description, concrete.subSpec, Lifetime::Enumeration);
460 extraSpecs.push_back(
timePipeline(aodReader, ctx.options().get<int64_t>(
"readers")));
464 if (ccdbBackend.outputs.empty() ==
false) {
465 ccdbBackend.outputs.push_back(
OutputSpec{
"CTP",
"OrbitReset", 0});
466 InputSpec matcher{
"dstf",
"FLP",
"DISTSUBTIMEFRAME", 0xccdb};
467 bool providesDISTSTF =
false;
470 for (
auto& dp : workflow) {
471 for (
auto&
output : dp.outputs) {
473 providesDISTSTF =
true;
478 if (providesDISTSTF) {
488 if (aodReader.outputs.empty() ==
false) {
489 ccdbBackend.inputs.push_back(
InputSpec{
"tfn",
"TFN",
"TFNumber"});
490 }
else if (providesDISTSTF) {
491 ccdbBackend.inputs.push_back(
InputSpec{
"tfn", dstf, Lifetime::Timeframe});
497 int enumCandidate = -1;
498 int timerCandidate = -1;
499 for (
size_t wi = 0; wi < workflow.size(); wi++) {
500 auto& dp = workflow[wi];
501 if (dp.inputs.size() != 1) {
504 auto lifetime = dp.inputs[0].lifetime;
505 if (lifetime == Lifetime::Enumeration && (enumCandidate == -1 || workflow[enumCandidate].
name > dp.name)) {
508 if (lifetime == Lifetime::Timer && (timerCandidate == -1 || workflow[timerCandidate].
name > dp.name)) {
512 if (enumCandidate != -1) {
513 auto& dp = workflow[enumCandidate];
515 ccdbBackend.inputs.push_back(
InputSpec{
"tfn", dstf, Lifetime::Timeframe});
516 }
else if (timerCandidate != -1) {
517 auto& dp = workflow[timerCandidate];
519 ccdbBackend.inputs.push_back(
InputSpec{{
"tfn"}, dstf, Lifetime::Timeframe});
525 extraSpecs.push_back(ccdbBackend);
530 bool requiresDISTSUBTIMEFRAME =
false;
531 for (
auto& dp : workflow) {
532 for (
auto& input : dp.inputs) {
534 requiresDISTSUBTIMEFRAME =
true;
539 if (requiresDISTSUBTIMEFRAME) {
544 int enumCandidate = -1;
545 int timerCandidate = -1;
546 for (
size_t wi = 0; wi < workflow.size(); wi++) {
547 auto& dp = workflow[wi];
548 if (dp.inputs.size() != 1) {
551 auto lifetime = dp.inputs[0].lifetime;
552 if (lifetime == Lifetime::Enumeration && (enumCandidate == -1 || workflow[enumCandidate].
name > dp.name)) {
555 if (lifetime == Lifetime::Timer && (timerCandidate == -1 || workflow[timerCandidate].
name > dp.name)) {
559 if (enumCandidate != -1) {
560 auto& dp = workflow[enumCandidate];
562 ccdbBackend.inputs.push_back(
InputSpec{
"tfn", dstf, Lifetime::Timeframe});
563 }
else if (timerCandidate != -1) {
564 auto& dp = workflow[timerCandidate];
566 ccdbBackend.inputs.push_back(
InputSpec{{
"tfn"}, dstf, Lifetime::Timeframe});
572 if (timer.outputs.empty() ==
false) {
573 extraSpecs.push_back(timer);
578 if (ac.providedOutputObjHist.empty() ==
false) {
580 extraSpecs.push_back(rootSink);
583 workflow.insert(workflow.end(), extraSpecs.begin(), extraSpecs.end());
587 auto [outputsInputsTmp, isDanglingTmp] = analyzeOutputs(workflow);
588 ac.isDangling = isDanglingTmp;
589 ac.outputsInputs = outputsInputsTmp;
597 for (
auto ii = 0u; ii < ac.outputsInputs.size(); ii++) {
599 auto ds = dod->getDataOutputDescriptors(ac.outputsInputs[ii]);
600 if (
ds.size() > 0 || ac.isDangling[ii]) {
601 ac.outputsInputsAOD.emplace_back(ac.outputsInputs[ii]);
607 if (ac.outputsInputsAOD.size() > 0) {
609 ac.outputsInputsAOD.emplace_back(
InputSpec{
"tfn",
"TFN",
"TFNumber"});
610 ac.outputsInputsAOD.emplace_back(
InputSpec{
"tff",
"TFF",
"TFFilename"});
612 extraSpecs.push_back(fileSink);
614 auto it = std::find_if(ac.outputsInputs.begin(), ac.outputsInputs.end(), [](
InputSpec& spec) ->
bool {
615 return DataSpecUtils::partialMatch(spec, o2::header::DataOrigin(
"TFN"));
617 size_t ii = std::distance(ac.outputsInputs.begin(), it);
618 ac.isDangling[ii] =
false;
621 workflow.insert(workflow.end(), extraSpecs.begin(), extraSpecs.end());
625 std::vector<InputSpec> redirectedOutputsInputs;
626 for (
auto ii = 0u; ii < ac.outputsInputs.size(); ii++) {
627 if (ctx.options().get<std::string>(
"forwarding-policy") ==
"none") {
632 if (!ac.isDangling[ii] && (ctx.options().get<std::string>(
"forwarding-policy") !=
"all")) {
639 redirectedOutputsInputs.emplace_back(ac.outputsInputs[ii]);
642 std::vector<InputSpec> unmatched;
643 auto forwardingDestination = ctx.options().get<std::string>(
"forwarding-destination");
644 if (redirectedOutputsInputs.size() > 0 && forwardingDestination ==
"file") {
646 if (unmatched.size() != redirectedOutputsInputs.size()) {
647 extraSpecs.push_back(fileSink);
649 }
else if (redirectedOutputsInputs.size() > 0 && forwardingDestination ==
"fairmq") {
651 extraSpecs.push_back(fairMQSink);
652 }
else if (forwardingDestination !=
"drop") {
653 throw runtime_error_f(
"Unknown forwarding destination %s", forwardingDestination.c_str());
655 if (unmatched.size() > 0 || redirectedOutputsInputs.size() > 0) {
656 std::vector<InputSpec> ignored = unmatched;
657 ignored.insert(ignored.end(), redirectedOutputsInputs.begin(), redirectedOutputsInputs.end());
658 for (
auto& ignoredInput : ignored) {
659 ignoredInput.lifetime = Lifetime::Sporadic;
665 workflow.insert(workflow.end(), extraSpecs.begin(), extraSpecs.end());
755 std::vector<DeviceConnectionEdge>& logicalEdges,
756 std::vector<OutputSpec>& outputs,
757 std::vector<LogicalForwardInfo>& forwardedInputsInfo)
760 if (workflow.empty()) {
765 std::vector<LogicalOutputInfo> availableOutputsInfo;
766 auto const& constOutputs = outputs;
768 std::vector<LogicalOutputInfo> forwards;
772 auto enumerateAvailableOutputs = [&workflow, &outputs, &availableOutputsInfo]() {
774 for (
size_t wi = 0; wi < workflow.size(); ++wi) {
775 auto& producer = workflow[wi];
776 if (producer.outputs.empty()) {
777 O2_SIGNPOST_EVENT_EMIT(workflow_helpers, sid,
"output enumeration",
"No outputs for [%zu] %{public}s", wi, producer.name.c_str());
779 O2_SIGNPOST_START(workflow_helpers, sid,
"output enumeration",
"Enumerating outputs for producer [%zu] %{}s public", wi, producer.name.c_str());
781 for (
size_t oi = 0; oi < producer.outputs.size(); ++oi) {
782 auto& out = producer.outputs[oi];
783 auto uniqueOutputId = outputs.size();
787 outputs.push_back(out);
793 auto errorDueToMissingOutputFor = [&workflow, &constOutputs](
size_t ci,
size_t ii) {
794 auto input = workflow[ci].inputs[ii];
795 std::ostringstream
str;
796 str <<
"No matching output found for "
797 <<
DataSpecUtils::describe(input) <<
" as requested by data processor \"" << workflow[ci].name <<
"\". Candidates:\n";
799 for (
auto&
output : constOutputs) {
803 throw std::runtime_error(
str.str());
817 enumerateAvailableOutputs();
819 std::vector<bool> matches(constOutputs.size());
820 for (
size_t consumer = 0; consumer < workflow.size(); ++consumer) {
822 O2_SIGNPOST_START(workflow_helpers, sid,
"input matching",
"Matching inputs of consumer [%zu] %{}s public", consumer, workflow[consumer].
name.c_str());
823 for (
size_t input = 0; input < workflow[consumer].inputs.size(); ++input) {
825 for (
size_t i = 0;
i < constOutputs.size();
i++) {
834 for (
size_t i = 0;
i < availableOutputsInfo.size();
i++) {
836 if (!matches[availableOutputsInfo[
i].outputGlobalIndex]) {
839 auto* oif = &availableOutputsInfo[
i];
843 auto producer = oif->specIndex;
844 auto uniqueOutputId = oif->outputGlobalIndex;
845 for (
size_t tpi = 0; tpi < workflow[consumer].maxInputTimeslices; ++tpi) {
846 for (
size_t ptpi = 0; ptpi < workflow[producer].maxInputTimeslices; ++ptpi) {
847 O2_SIGNPOST_EVENT_EMIT(workflow_helpers, sid,
"output",
"Adding edge between %{public}s and %{public}s", workflow[consumer].
name.c_str(),
848 workflow[producer].name.c_str());
849 logicalEdges.emplace_back(
DeviceConnectionEdge{producer, consumer, tpi, ptpi, uniqueOutputId, input, oif->forward});
854 oif->enabled =
false;
856 if (forwards.empty()) {
857 errorDueToMissingOutputFor(consumer, input);
859 availableOutputsInfo.erase(std::remove_if(availableOutputsInfo.begin(), availableOutputsInfo.end(), [](
auto& info) { return info.enabled == false; }), availableOutputsInfo.end());
860 for (
auto& forward : forwards) {
861 availableOutputsInfo.push_back(forward);