244 static std::vector<bool> present;
245 static std::vector<bool> ignored;
246 static std::vector<size_t> dataSizes;
247 static std::vector<bool> showSize;
249 present.resize(
routes.size(),
false);
251 ignored.resize(
routes.size(),
false);
253 dataSizes.resize(
routes.size(), 0);
255 showSize.resize(
routes.size(),
false);
257 static std::vector<size_t> unmatchedDescriptions;
258 unmatchedDescriptions.clear();
264 size_t expectedDataSpecs = 0;
265 for (
size_t pi = 0; pi < present.size(); ++pi) {
266 auto& spec =
routes[pi].matcher;
271 if (
routes[pi].timeslice == 0) {
276 size_t foundDataSpecs = 0;
277 bool skipAsAllFound =
false;
278 for (
int msgidx = 0; msgidx < parts.Size(); msgidx += 2) {
279 bool allFound =
true;
281 const auto dh = o2::header::get<DataHeader*>(parts.At(msgidx)->GetData());
282 auto const sih = o2::header::get<SourceInfoHeader*>(parts.At(msgidx)->GetData());
283 if (sih !=
nullptr) {
287 if (parts.At(msgidx).get() ==
nullptr) {
288 LOG(error) <<
"unexpected nullptr found. Skipping message pair.";
292 LOG(error) <<
"data on input " << msgidx <<
" does not follow the O2 data model, DataHeader missing";
298 if (firstDH ==
nullptr) {
300 if (doPrintSizes && firstDH->
tfCounter % doPrintSizes != 0) {
305 if (dph ==
nullptr) {
306 dph = o2::header::get<DataProcessingHeader*>(parts.At(msgidx)->GetData());
307 for (
size_t pi = 0; pi < present.size(); ++pi) {
313 for (
size_t pi = 0; pi < present.size(); ++pi) {
314 if ((present[pi] || ignored[pi]) && !doPrintSizes) {
323 auto& spec =
routes[pi].matcher;
324 OutputSpec query{dh->dataOrigin, dh->dataDescription, dh->subSpecification};
326 if (!present[pi] && !ignored[pi]) {
336 if (addToSize >= 0) {
337 int increment = (dh->splitPayloadParts > 0 && dh->splitPayloadParts == dh->splitPayloadIndex) ? 1 : 2;
338 for (
int msgidx2 = msgidx + 1; msgidx2 < msgidxLast; msgidx2 += increment) {
339 dataSizes[addToSize] += parts.At(msgidx2)->GetSize();
343 msgidx = msgidxLast - 2;
344 if (allFound && !doPrintSizes) {
345 skipAsAllFound =
true;
351 for (
size_t pi = 0; pi < present.size(); ++pi) {
352 if (present[pi] && !ignored[pi]) {
355 if (!present[pi] && !ignored[pi]) {
357 unmatchedDescriptions.push_back(pi);
360 int timeframeCompleteness = emptyTf ? 0 : (unmatchedDescriptions.size() ? -1 : 1);
361 (
void)timeframeCompleteness;
363 if (skipAsAllFound && !doPrintSizes) {
367 if (firstDH && doPrintSizes) {
368 std::string
sizes =
"";
369 size_t totalSize = 0;
370 for (
size_t pi = 0; pi < present.size(); ++pi) {
372 totalSize += dataSizes[pi];
373 auto& spec =
routes[pi].matcher;
377 LOGP(important,
"RAW {} size report:{}- Total:{}", firstDH->
tfCounter,
sizes, fmt::group_digits(totalSize));
380 if (!doInjectMissingData) {
384 if (unmatchedDescriptions.size() > 0) {
387 LOG(error) <<
"Received an EndOfStream message together with data. This should not happen.";
389 LOG(detail) <<
"This is an End Of Stream message. Not injecting anything.";
392 if (firstDH ==
nullptr) {
393 LOG(error) <<
"Input proxy received incomplete data without any data header. This should not happen! Cannot inject missing data as requsted.";
396 if (dph ==
nullptr) {
397 LOG(error) <<
"Input proxy received incomplete data without any data processing header. This should happen! Cannot inject missing data as requsted.";
400 std::string missing =
"";
401 bool showAlarm =
false;
402 uint32_t runNumber = 0;
404 runNumber = strtoul(device.fConfig->GetProperty<std::string>(
"runNumber",
"").c_str(),
nullptr, 10);
407 for (
auto mi : unmatchedDescriptions) {
408 auto& spec =
routes[mi].matcher;
414 if (subSpec == std::nullopt) {
415 *subSpec = 0xDEADBEEF;
420 dh.subSpecification = *subSpec;
422 dh.runNumber = runNumber;
423 dh.splitPayloadParts = 0;
424 dh.splitPayloadIndex = 0;
427 auto& channelName =
routes[mi].channel;
428 auto& channelInfo = device.GetChannel(channelName);
429 auto channelAlloc = o2::pmr::getTransportAllocator(channelInfo.Transport());
431 parts.AddPart(std::move(headerMessage));
433 parts.AddPart(device.NewMessageFor(channelName, 0, 0));
438 static int maxWarn = 10;
439 static int contDeadBeef = 0;
440 if (showAlarm && ++contDeadBeef <= maxWarn) {
441 LOGP(alarm,
"Found {}/{} data specs, missing data specs: {}, injecting 0xDEADBEEF{}", foundDataSpecs, expectedDataSpecs, missing, contDeadBeef == maxWarn ?
" - disabling alarm now to stop flooding the log" :
"");
450 class DroppedDataSpecs
453 DroppedDataSpecs() =
default;
459 [[nodiscard]]
bool find(std::string
const& desc)
const
461 return descriptions.find(desc) != std::string::npos;
464 void add(std::string
const& desc)
466 descriptions +=
"\n " + desc;
471 if (not descriptions.empty()) {
472 LOG(warning) <<
"Some input data could not be matched by filter rules to output specs\n"
473 <<
"Active rules: " << descriptions <<
"\n"
474 <<
"DROPPING OF THESE MESSAGES HAS BEEN ENABLED BY CONFIGURATION";
479 std::string descriptions;
482 return [filterSpecs = std::move(filterSpecs), throwOnUnmatchedInputs, droppedDataSpecs = std::make_shared<DroppedDataSpecs>()](
TimingInfo& timingInfo,
ServiceRegistryRef const& services, fair::mq::Parts& parts,
ChannelRetriever channelRetriever,
size_t newTimesliceId,
bool& stop) {
484 std::unordered_map<std::string, fair::mq::Parts> outputs;
485 std::vector<std::string> unmatchedDescriptions;
488 static bool override_creation_env = getenv(
"DPL_RAWPROXY_OVERRIDE_ORBITRESET");
489 bool override_creation =
false;
490 uint64_t creationVal = 0;
491 if (override_creation_env) {
492 static uint64_t creationValBase = std::stoul(getenv(
"DPL_RAWPROXY_OVERRIDE_ORBITRESET"));
493 creationVal = creationValBase;
494 override_creation =
true;
496 auto orbitResetTimeUrl = device->fConfig->GetProperty<std::string>(
"orbit-reset-time",
"ccdb://CTP/Calib/OrbitResetTime");
498 creationVal = std::strtoll(orbitResetTimeUrl.c_str(), &err, 10);
499 if (err && *err == 0 && creationVal) {
500 override_creation =
true;
504 int fmqRunNumber = -1;
506 fmqRunNumber = atoi(device->fConfig->GetProperty<std::string>(
"runNumber",
"").c_str());
510 for (
int msgidx = 0; msgidx < parts.Size(); msgidx += 2) {
511 if (parts.At(msgidx).get() ==
nullptr) {
512 LOG(error) <<
"unexpected nullptr found. Skipping message pair.";
515 auto* header = parts.At(msgidx)->GetData();
516 const auto dh = o2::header::get<DataHeader*>(header);
518 LOG(error) <<
"data on input " << msgidx <<
" does not follow the O2 data model, DataHeader missing";
524 auto dph = o2::header::get<DataProcessingHeader*>(header);
526 LOG(error) <<
"data on input " << msgidx <<
" does not follow the O2 data model, DataProcessingHeader missing";
530 if (override_creation) {
534 timingInfo.creation = dph->creation;
535 timingInfo.firstTForbit = dh->firstTForbit;
536 timingInfo.runNumber = dh->runNumber;
537 timingInfo.tfCounter = dh->tfCounter;
538 LOG(
debug) << msgidx <<
": " <<
DataSpecUtils::describe(
OutputSpec{dh->dataOrigin, dh->dataDescription, dh->subSpecification}) <<
" part " << dh->splitPayloadIndex <<
" of " << dh->splitPayloadParts <<
" payload " << parts.At(msgidx + 1)->GetSize();
540 LOG(error) <<
"INVALID runNumber / tfCounter: runNumber " << dh->runNumber
541 <<
", tfCounter " << dh->tfCounter <<
", FMQ runNumber " << fmqRunNumber
542 <<
" for msgidx " << msgidx <<
": " <<
DataSpecUtils::describe(
OutputSpec{dh->dataOrigin, dh->dataDescription, dh->subSpecification}) <<
" part " << dh->splitPayloadIndex <<
" of " << dh->splitPayloadParts <<
" payload " << parts.At(msgidx + 1)->GetSize();
545 OutputSpec query{dh->dataOrigin, dh->dataDescription, dh->subSpecification};
546 LOG(
debug) <<
"processing " <<
DataSpecUtils::describe(
OutputSpec{dh->dataOrigin, dh->dataDescription, dh->subSpecification}) <<
" time slice " << dph->startTime <<
" part " << dh->splitPayloadIndex <<
" of " << dh->splitPayloadParts;
547 int finalBlockIndex = 0;
548 std::string channelName =
"";
550 for (
auto const& spec : filterSpecs) {
554 channelName = channelRetriever(query, dph->startTime);
563 if (finalBlockIndex > parts.Size()) {
569 if (!channelName.empty()) {
583 LOGP(
debug,
"associating {} part(s) at index {} to channel {} ({})", finalBlockIndex - msgidx, msgidx, channelName, outputs[channelName].
Size());
584 for (; msgidx < finalBlockIndex; ++msgidx) {
585 outputs[channelName].AddPart(std::move(parts.At(msgidx)));
589 msgidx = finalBlockIndex - 2;
596 bool didSendParts =
false;
597 for (
auto& [channelName, channelParts] : outputs) {
598 if (channelParts.Size() == 0) {
602 sendOnChannel(*device, channelParts, channelName, newTimesliceId);
604 if (not unmatchedDescriptions.empty()) {
605 if (throwOnUnmatchedInputs) {
606 std::string descriptions;
607 for (
auto const& desc : unmatchedDescriptions) {
608 descriptions +=
"\n " + desc;
610 throw std::runtime_error(
"No matching filter rule for input data " + descriptions +
611 "\n Add appropriate matcher(s) to dataspec definition or allow to drop unmatched data");
613 bool changed =
false;
614 for (
auto const& desc : unmatchedDescriptions) {
615 if (not droppedDataSpecs->find(desc)) {
617 droppedDataSpecs->add(desc);
622 droppedDataSpecs->warning();
669 std::vector<OutputSpec>
const& outputs,
670 char const* defaultChannelConfig,
674 bool doInjectMissingData,
675 unsigned int doPrintSizes)
681 static std::vector<std::string>
channels;
682 static std::vector<int> numberOfEoS(
channels.size(), 0);
683 static std::vector<int> eosPeersCount(
channels.size(), 0);
691 auto outputChannels = ctx.services().get<
RawDeviceService>().spec().outputChannels;
700 auto channelConfigurationChecker = [device, deviceName, services = ctx.services()]() {
704 eosPeersCount.clear();
705 for (
auto& [channelName, _] : services.get<
RawDeviceService>().device()->GetChannels()) {
707 if (strncmp(channelName.c_str(), deviceName.c_str(), deviceName.size()) == 0) {
712 LOGP(detail,
"Injecting channel '{}' into DPL configuration", channel);
714 auto& channelPtr = services.get<
RawDeviceService>().device()->GetChannel(channel, 0);
717 .hasPendingEvents =
false,
719 .channel = &channelPtr,
724 numberOfEoS.resize(
channels.size(), 0);
725 eosPeersCount.resize(
channels.size(), 0);
733 static bool wasRunning =
false;
734 if (fair::mq::State{
state} == fair::mq::State::Running) {
737 if (fair::mq::State{
state} != fair::mq::State::Ready || !wasRunning) {
740 uv_update_time(deviceState.loop);
745 int64_t cleanupCount = deviceState.cleanupCount.load();
750 doDrain = device->NewStatePending() ==
false && deviceState.cleanupCount == cleanupCount;
751 fair::mq::Parts parts;
752 for (
size_t ci = 0; ci < deviceState.inputChannelInfos.size(); ++ci) {
753 auto& info = deviceState.inputChannelInfos[ci];
758 info.channel->Receive(parts, 10);
761 uv_run(deviceState.loop, UV_RUN_NOWAIT);
765 ctx.services().get<
CallbackService>().set<CallbackService::Id::Start>(channelConfigurationChecker);
766 if (ctx.options().get<std::string>(
"ready-state-policy") ==
"drain") {
767 LOG(info) <<
"Drain mode requested while in Ready state";
768 ctx.services().get<
CallbackService>().set<CallbackService::Id::DeviceStateChanged>(drainMessages);
771 static auto countEoS = [](fair::mq::Parts& inputs) ->
int {
773 for (
int msgidx = 0; msgidx < inputs.Size() / 2; ++msgidx) {
776 if (inputs.At(msgidx * 2).get() ==
nullptr) {
779 auto const sih = o2::header::get<SourceInfoHeader*>(inputs.At(msgidx * 2)->GetData());
788 auto dataHandler = [converter, doInjectMissingData, doPrintSizes,
789 outputRoutes = std::move(outputRoutes),
797 static std::string emptyChannel =
"";
798 for (
auto& route : outputRoutes) {
800 if (
DataSpecUtils::match(route.matcher, query) && ((timeslice % route.maxTimeslices) == route.timeslice)) {
801 return route.channel;
807 std::string
const& channel =
channels[ci];
809 int nEos = countEoS(inputs);
811 std::fill(numberOfEoS.begin(), numberOfEoS.end(), 0);
812 std::fill(eosPeersCount.begin(), eosPeersCount.end(), 0);
814 numberOfEoS[ci] += nEos;
815 if (numberOfEoS[ci]) {
816 eosPeersCount[ci] = std::max<int>(eosPeersCount[ci], device->GetNumberOfConnectedPeers(channel));
819 bool shouldstop =
false;
820 if (doInjectMissingData) {
821 injectMissingData(*device, inputs, outputRoutes, doInjectMissingData, doPrintSizes);
823 bool didSendParts = converter(timingInfo,
ref, inputs, channelRetriever, timesliceIndex->getOldestPossibleOutput().timeslice.value, shouldstop);
831 bool everyEoS = shouldstop;
832 if (!shouldstop && nEos) {
834 for (
unsigned int i = 0;
i < numberOfEoS.size();
i++) {
835 if (numberOfEoS[
i] < eosPeersCount[
i]) {
843 LOG(info) <<
"Received (on channel " << ci <<
") " << numberOfEoS[ci] <<
" end-of-stream from " << eosPeersCount[ci] <<
" peers, forwarding end-of-stream (shouldstop " << (
int)shouldstop <<
", nEos " << nEos <<
", newRun " << (
int)newRun <<
")";
845 for (
auto& info : deviceState->inputChannelInfos) {
848 std::fill(numberOfEoS.begin(), numberOfEoS.end(), 0);
849 std::fill(eosPeersCount.begin(), eosPeersCount.end(), 0);
850 control->endOfStream();
857 static size_t currentRunNumber = -1;
858 static bool inStopTransition =
false;
861 if (limiter.
check(ctx, std::stoi(device->fConfig->GetValue<std::string>(
"timeframes-rate-limit")), minSHM)) {
862 inStopTransition =
true;
865 bool didSendParts =
false;
866 for (
size_t ci = 0; ci <
channels.size(); ++ci) {
867 std::string
const& channel =
channels[ci];
868 int waitTime =
channels.size() == 1 ? -1 : 1;
870 while (maxRead-- > 0) {
871 fair::mq::Parts parts;
872 auto res = device->Receive(parts, channel, 0, waitTime);
873 if (
res == (
size_t)fair::mq::TransferCode::error) {
874 LOGP(error,
"Error while receiving on channel {}", channel);
877 unsigned int nReceived = parts.Size();
878 if (nReceived != 0) {
879 auto const dh = o2::header::get<DataHeader*>(parts.At(0)->GetData());
880 auto& timingInfo = ctx.services().get<
TimingInfo>();
882 if (currentRunNumber != -1 && dh->runNumber != 0 && dh->runNumber != currentRunNumber) {
884 inStopTransition =
false;
886 if (currentRunNumber == -1 || dh->runNumber != 0) {
889 timingInfo.runNumber = dh->runNumber;
890 timingInfo.firstTForbit = dh->firstTForbit;
891 timingInfo.tfCounter = dh->tfCounter;
893 auto const dph = o2::header::get<DataProcessingHeader*>(parts.At(0)->GetData());
894 if (dph !=
nullptr) {
895 timingInfo.timeslice = dph->startTime;
896 timingInfo.creation = dph->creation;
898 if (!inStopTransition) {
899 didSendParts |= dataHandler(ctx.services(), timingInfo, parts, 0, ci, newRun);
902 ctx.services().get<o2::monitoring::Monitoring>().send(o2::monitoring::Metric{(uint64_t)timingInfo.tfCounter,
"df-sent"}.addTag(o2::monitoring::tags::Key::Subsystem, o2::monitoring::tags::Value::DPL));
905 if (nReceived == 0 ||
channels.size() == 1) {
918 decongestion.nextEnumerationTimeslice -= 1;
924 const char* d = strdup(((std::string(defaultChannelConfig).find(
"name=") == std::string::npos ? (std::string(
"name=") +
name +
",") :
"") + std::string(defaultChannelConfig)).c_str());
939 const char* defaultChannelConfig)
950 auto channelConfig = options.
get<std::string>(
"channel-config");
951 std::regex
r{R
"(name=([^,]*))"};
952 std::vector<std::string> values{std::sregex_token_iterator{std::begin(channelConfig), std::end(channelConfig), r, 1},
953 std::sregex_token_iterator{}};
955 throw std::runtime_error(
"failed to extract channel name from channel configuration parameter '" + channelConfig +
"'");
957 std::string outputChannelName =
values[0];
959 auto* device = rds.
device();
963 auto channelConfigurationChecker = [inputSpecs = std::move(inputSpecs), device, outputChannelName]() {
964 LOG(info) <<
"checking channel configuration";
965 if (device->GetChannels().count(outputChannelName) == 0) {
966 throw std::runtime_error(
"no corresponding output channel found for input '" + outputChannelName +
"'");
970 auto lastDataProcessingHeader = std::make_shared<DataProcessingHeader>(0, 0);
972 auto& spec =
const_cast<DeviceSpec&
>(deviceSpec);
974 for (
auto const& inputSpec : inputSpecs) {
980 .matcher = inputSpec,
981 .channel = outputChannelName,
983 spec.forwards.emplace_back(route);
986 auto forwardEos = [device, lastDataProcessingHeader, outputChannelName](
EndOfStreamContext&) {
990 for (
auto& channelInfo : device->GetChannels()) {
991 auto& channelName = channelInfo.first;
992 if (channelName != outputChannelName) {
996 uint32_t runNumber = 0;
998 runNumber = strtoul(device->fConfig->GetProperty<std::string>(
"runNumber",
"").c_str(),
nullptr, 10);
1013 auto channelAlloc = o2::pmr::getTransportAllocator(channelInfo.second[0].Transport());
1015 fair::mq::Parts out;
1016 out.AddPart(std::move(headerMessage));
1018 out.AddPart(device->NewMessageFor(channelName, 0, 0));
1025 for (
size_t ii = 0; ii != inputs.
size(); ++ii) {
1026 for (
size_t pi = 0; pi < inputs.
getNofParts(ii); ++pi) {
1027 auto part = inputs.
getByPos(ii, pi);
1028 const auto* dph = o2::header::get<DataProcessingHeader*>(part.header);
1031 lastDataProcessingHeader->startTime = dph->startTime;
1032 lastDataProcessingHeader->duration = dph->duration;
1033 lastDataProcessingHeader->creation = dph->creation;
1039 const char* d = strdup(((std::string(defaultChannelConfig).find(
"name=") == std::string::npos ? (std::string(
"name=") +
name +
",") :
"") + std::string(defaultChannelConfig)).c_str());
1048 Inputs const& inputSpecs,
1049 const char* defaultChannelConfig,
1056 spec.
inputs = inputSpecs;
1059 auto device = rds.
device();
1064 auto channelNames = std::make_shared<std::vector<std::string>>();
1065 auto channelConfigurationInitializer = [&proxy, inputSpecs = std::move(inputSpecs), device, channelSelector, &deviceSpec, channelNames]() {
1066 channelNames->clear();
1067 auto& mutableDeviceSpec =
const_cast<DeviceSpec&
>(deviceSpec);
1068 for (
auto const& spec : inputSpecs) {
1069 auto channel = channelSelector(spec, device->GetChannels());
1070 if (device->GetChannels().count(channel) == 0) {
1071 throw std::runtime_error(
"no corresponding output channel found for input '" + channel +
"'");
1084 mutableDeviceSpec.forwards.emplace_back(route);
1086 channelNames->emplace_back(std::move(channel));
1088 proxy.
bind(mutableDeviceSpec.outputs, mutableDeviceSpec.inputs, mutableDeviceSpec.forwards, *device);
1091 auto channelConfigurationDisposer = [&deviceSpec]() {
1092 auto& mutableDeviceSpec =
const_cast<DeviceSpec&
>(deviceSpec);
1093 mutableDeviceSpec.
forwards.clear();
1098 auto lastDataProcessingHeader = std::make_shared<DataProcessingHeader>(0, 0);
1099 auto forwardEos = [device, lastDataProcessingHeader, channelNames](
EndOfStreamContext&) {
1103 for (
auto& channelInfo : device->GetChannels()) {
1104 auto& channelName = channelInfo.first;
1105 auto checkChannel = [channelNames = std::move(*channelNames)](std::string
const&
name) ->
bool {
1106 for (
auto const&
n : channelNames) {
1113 if (!checkChannel(channelName)) {
1116 uint32_t runNumber = 0;
1118 runNumber = strtoul(device->fConfig->GetProperty<std::string>(
"runNumber",
"").c_str(),
nullptr, 10);
1133 auto channelAlloc = o2::pmr::getTransportAllocator(channelInfo.second[0].Transport());
1135 fair::mq::Parts out;
1136 out.AddPart(std::move(headerMessage));
1138 out.AddPart(device->NewMessageFor(channelName, 0, 0));
1139 LOGP(detail,
"Forwarding EoS to {}", channelName);
1149 for (
size_t ii = 0; ii != inputs.
size(); ++ii) {
1150 for (
size_t pi = 0; pi < inputs.
getNofParts(ii); ++pi) {
1151 auto part = inputs.
getByPos(ii, pi);
1152 const auto* dph = o2::header::get<DataProcessingHeader*>(part.header);
1155 lastDataProcessingHeader->startTime = dph->startTime;
1156 lastDataProcessingHeader->duration = dph->duration;
1157 lastDataProcessingHeader->creation = dph->creation;
1163 const char* d = strdup(((std::string(defaultChannelConfig).find(
"name=") == std::string::npos ? (std::string(
"name=") +
name +
",") :
"") + std::string(defaultChannelConfig)).c_str());