Project
Loading...
Searching...
No Matches
ExternalFairMQDeviceProxy.cxx
Go to the documentation of this file.
1// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
2// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
3// All rights not expressly granted are reserved.
4//
5// This software is distributed under the terms of the GNU General Public
6// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
7//
8// In applying this license CERN does not waive the privileges and immunities
9// granted to it by virtue of its status as an Intergovernmental Organization
10// or submit itself to any jurisdiction.
35#include "Headers/DataHeader.h"
36#include "Headers/Stack.h"
37#include "DecongestionService.h"
39
40#include "./DeviceSpecHelpers.h"
41#include "Monitoring/Monitoring.h"
42
43#include <fairmq/Parts.h>
44#include <fairmq/Device.h>
45#include <uv.h>
46#include <cstring>
47#include <cassert>
48#include <memory>
49#include <optional>
50#include <unordered_map>
51#include <numeric> // std::accumulate
52#include <sstream>
53#include <stdexcept>
54#include <regex>
55
56namespace o2::framework
57{
59
64
69
71
72void sendOnChannel(fair::mq::Device& device, fair::mq::Parts& messages, std::string const& channel, size_t timeSlice)
73{
74 // Note: DPL is only setting up one instance of a channel while FairMQ allows to have an
75 // array of channels, the index is 0 in the call
76 constexpr auto index = 0;
77 LOG(debug) << "sending " << messages.Size() << " messages on " << channel;
78 // TODO: we can make this configurable
79 const int maxTimeout = 10000;
80 int timeout = 0;
81 // try dispatch with increasing timeout in order to also drop a warning if the dispatching
82 // has been tried multiple times within max timeout
83 // since we do not want any messages to be dropped at this stage, we stay in the loop until
84 // the downstream congestion is resolved
85 // TODO: we might want to treat this error condition some levels higher up, but for
86 // the moment its an appropriate solution. The important thing is not to drop
87 // messages and to be informed about the congestion.
88 while (device.Send(messages, channel, index, timeout) < 0) {
89 if (timeout == 0) {
90 timeout = 1;
91 } else if (timeout < maxTimeout) {
92 timeout *= 10;
93 } else {
94 LOG(alarm) << "Cannot dispatch to channel " << channel << " due to DOWNSTREAM BACKPRESSURE. NO DATA IS DROPPED,"
95 << " will keep retrying. This is only a problem if downstream congestion does not resolve by itself.";
96 if (timeout == maxTimeout) {
97 // we add 1ms to disable the warning below
98 timeout += 1;
99 }
100 }
101 if (device.NewStatePending()) {
102 LOG(alarm) << "Device state change is requested, dropping " << messages.Size() << " pending message(s) "
103 << "on channel " << channel << ". "
104 << "ATTENTION: DATA IS LOST! Could not dispatch data to downstream consumer(s), check if "
105 << "consumers have been terminated too early";
106 // make sure we disable the warning below
107 timeout = maxTimeout + 1;
108 break;
109 }
110 }
111
112 // FIXME: we need a better logic for avoiding message spam
113 if (timeout > 100 && timeout <= maxTimeout) {
114 LOG(warning) << "dispatching on channel " << channel << " was delayed by " << timeout / 1000.f << " s";
115 }
116 // TODO: feeling this is a bit awkward, but the interface of fair::mq::Parts does not provide a
117 // method to clear the content.
118 // Maybe the FairMQ API can be improved at some point. Actually the ownership of all messages should be passed
119 // on to the transport and the messages should be empty after sending and the parts content can be cleared.
120 // assert(std::accumulate(messages.begin(), messages.end(), true, [](bool a, auto const& msg) {return a && (msg.get() == nullptr);}));
121 messages.fParts.clear();
122}
123
124void sendOnChannel(fair::mq::Device& device, fair::mq::Parts& messages, OutputSpec const& spec, DataProcessingHeader::StartTime tslice, ChannelRetriever& channelRetriever)
125{
126 // Note: DPL is only setting up one instance of a channel while FairMQ allows to have an
127 // array of channels, the index is 0 in the call
128 auto channel = channelRetriever(spec, tslice);
129 if (channel.empty()) {
130 LOG(warning) << "can not find matching channel for " << DataSpecUtils::describe(spec) << " timeslice " << tslice;
131 return;
132 }
133 sendOnChannel(device, messages, channel, tslice);
134}
135
136void sendOnChannel(fair::mq::Device& device, o2::header::Stack&& headerStack, fair::mq::MessagePtr&& payloadMessage, OutputSpec const& spec, ChannelRetriever& channelRetriever)
137{
138 const auto* dph = o2::header::get<DataProcessingHeader*>(headerStack.data());
139 if (!dph) {
140 LOG(error) << "Header Stack does not follow the O2 data model, DataProcessingHeader missing";
141 return;
142 }
143 auto channelName = channelRetriever(spec, dph->startTime);
144 constexpr auto index = 0;
145 if (channelName.empty()) {
146 LOG(warning) << "can not find matching channel for " << DataSpecUtils::describe(spec);
147 return;
148 }
149 for (auto& channelInfo : device.GetChannels()) {
150 if (channelInfo.first != channelName) {
151 continue;
152 }
153 assert(channelInfo.second.size() == 1);
154 // allocate the header message using the underlying transport of the channel
155 auto channelAlloc = o2::pmr::getTransportAllocator(channelInfo.second[index].Transport());
156 fair::mq::MessagePtr headerMessage = o2::pmr::getMessage(std::move(headerStack), channelAlloc);
157
158 fair::mq::Parts out;
159 out.AddPart(std::move(headerMessage));
160 out.AddPart(std::move(payloadMessage));
161 sendOnChannel(device, out, channelName, dph->startTime);
162 return;
163 }
164 LOG(error) << "internal mismatch, can not find channel " << channelName << " in the list of channel infos of the device";
165}
166
167void sendOnChannel(fair::mq::Device& device, fair::mq::MessagePtr&& headerMessage, fair::mq::MessagePtr&& payloadMessage, OutputSpec const& spec, ChannelRetriever& channelRetriever)
168{
169 // const auto* dph = o2::header::get<DataProcessingHeader*>( *reinterpret_cast<o2::header::Stack*>(headerMessage->GetData()) );
170 const auto* dph = o2::header::get<DataProcessingHeader*>(headerMessage->GetData());
171 if (!dph) {
172 LOG(error) << "Header does not follow the O2 data model, DataProcessingHeader missing";
173 return;
174 }
175 auto tslice = dph->startTime;
176 fair::mq::Parts out;
177 out.AddPart(std::move(headerMessage));
178 out.AddPart(std::move(payloadMessage));
179 sendOnChannel(device, out, spec, tslice, channelRetriever);
180}
181
182void appendForSending(fair::mq::Device& device, o2::header::Stack&& headerStack, size_t timeSliceID, fair::mq::MessagePtr&& payloadMessage, OutputSpec const& spec, fair::mq::Parts& messageCache, ChannelRetriever& channelRetriever)
183{
184 auto channelName = channelRetriever(spec, timeSliceID);
185 constexpr auto index = 0;
186 if (channelName.empty()) {
187 LOG(warning) << "can not find matching channel for " << DataSpecUtils::describe(spec);
188 return;
189 }
190 for (auto& channelInfo : device.GetChannels()) {
191 if (channelInfo.first != channelName) {
192 continue;
193 }
194 assert(channelInfo.second.size() == 1);
195 // allocate the header message using the underlying transport of the channel
196 auto channelAlloc = o2::pmr::getTransportAllocator(channelInfo.second[index].Transport());
197 fair::mq::MessagePtr headerMessage = o2::pmr::getMessage(std::move(headerStack), channelAlloc);
198
199 fair::mq::Parts out;
200 messageCache.AddPart(std::move(headerMessage));
201 messageCache.AddPart(std::move(payloadMessage));
202 return;
203 }
204 LOG(error) << "internal mismatch, can not find channel " << channelName << " in the list of channel infos of the device";
205}
206
207InjectorFunction o2DataModelAdaptor(OutputSpec const& spec, uint64_t startTime, uint64_t /*step*/)
208{
209 return [spec](TimingInfo&, ServiceRegistryRef const& ref, fair::mq::Parts& parts, ChannelRetriever channelRetriever, size_t newTimesliceId, bool& stop) -> bool {
210 auto* device = ref.get<RawDeviceService>().device();
211 for (int i = 0; i < parts.Size() / 2; ++i) {
212 auto dh = o2::header::get<DataHeader*>(parts.At(i * 2)->GetData());
213
214 DataProcessingHeader dph{newTimesliceId, 0};
215 o2::header::Stack headerStack{*dh, dph};
216 sendOnChannel(*device, std::move(headerStack), std::move(parts.At(i * 2 + 1)), spec, channelRetriever);
217 }
218 return parts.Size() > 0;
219 };
220}
221
222auto getFinalIndex(DataHeader const& dh, size_t msgidx) -> size_t
223{
224 size_t finalBlockIndex = 0;
225 if (dh.splitPayloadParts > 0 && dh.splitPayloadParts == dh.splitPayloadIndex) {
226 // this is indicating a sequence of payloads following the header
227 // FIXME: we will probably also set the DataHeader version
228 // Current position + number of parts + 1 (for the header)
229 finalBlockIndex = msgidx + dh.splitPayloadParts + 1;
230 } else {
231 // We can consider the next splitPayloadParts as one block of messages pairs
232 // because we are guaranteed they are all the same.
233 // If splitPayloadParts = 0, we assume that means there is only one (header, payload)
234 // pair.
235 finalBlockIndex = msgidx + (dh.splitPayloadParts > 0 ? dh.splitPayloadParts : 1) * 2;
236 }
237 assert(finalBlockIndex >= msgidx + 2);
238 return finalBlockIndex;
239};
240
241void injectMissingData(fair::mq::Device& device, fair::mq::Parts& parts, std::vector<OutputRoute> const& routes, bool doInjectMissingData, unsigned int doPrintSizes)
242{
243 // Check for missing data.
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;
248 present.clear();
249 present.resize(routes.size(), false);
250 ignored.clear();
251 ignored.resize(routes.size(), false);
252 dataSizes.clear();
253 dataSizes.resize(routes.size(), 0);
254 showSize.clear();
255 showSize.resize(routes.size(), false);
256
257 static std::vector<size_t> unmatchedDescriptions;
258 unmatchedDescriptions.clear();
259 DataProcessingHeader const* dph = nullptr;
260 DataHeader const* firstDH = nullptr;
261 bool hassih = false;
262
263 // Do not check anything which has DISTSUBTIMEFRAME in it.
264 size_t expectedDataSpecs = 0;
265 for (size_t pi = 0; pi < present.size(); ++pi) {
266 auto& spec = routes[pi].matcher;
267 if (DataSpecUtils::asConcreteDataTypeMatcher(spec).description == header::DataDescription("DISTSUBTIMEFRAME")) {
268 ignored[pi] = true;
269 continue;
270 }
271 if (routes[pi].timeslice == 0) {
272 ++expectedDataSpecs;
273 }
274 }
275
276 size_t foundDataSpecs = 0;
277 bool skipAsAllFound = false;
278 for (int msgidx = 0; msgidx < parts.Size(); msgidx += 2) {
279 bool allFound = true;
280 int addToSize = -1;
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) {
284 hassih = true;
285 continue;
286 }
287 if (parts.At(msgidx).get() == nullptr) {
288 LOG(error) << "unexpected nullptr found. Skipping message pair.";
289 continue;
290 }
291 if (!dh) {
292 LOG(error) << "data on input " << msgidx << " does not follow the O2 data model, DataHeader missing";
293 if (msgidx > 0) {
294 --msgidx;
295 }
296 continue;
297 }
298 if (firstDH == nullptr) {
299 firstDH = dh;
300 if (doPrintSizes && firstDH->tfCounter % doPrintSizes != 0) {
301 doPrintSizes = 0;
302 }
303 }
304 // Copy the DataProcessingHeader from the first message.
305 if (dph == nullptr) {
306 dph = o2::header::get<DataProcessingHeader*>(parts.At(msgidx)->GetData());
307 for (size_t pi = 0; pi < present.size(); ++pi) {
308 if (routes[pi].timeslice != (dph->startTime % routes[pi].maxTimeslices)) {
309 ignored[pi] = true;
310 }
311 }
312 }
313 for (size_t pi = 0; pi < present.size(); ++pi) {
314 if ((present[pi] || ignored[pi]) && !doPrintSizes) {
315 continue;
316 }
317 // Consider uninvolved pipelines as present.
318 if (routes[pi].timeslice != (dph->startTime % routes[pi].maxTimeslices)) {
319 ignored[pi] = true;
320 continue;
321 }
322 allFound = false;
323 auto& spec = routes[pi].matcher;
324 OutputSpec query{dh->dataOrigin, dh->dataDescription, dh->subSpecification};
325 if (DataSpecUtils::match(spec, query)) {
326 if (!present[pi] && !ignored[pi]) {
327 ++foundDataSpecs;
328 present[pi] = true;
329 showSize[pi] = true;
330 }
331 addToSize = pi;
332 break;
333 }
334 }
335 int msgidxLast = getFinalIndex(*dh, msgidx);
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();
340 }
341 }
342 // Skip the rest of the block of messages. We subtract 2 because above we increment by 2.
343 msgidx = msgidxLast - 2;
344 if (allFound && !doPrintSizes) {
345 skipAsAllFound = true;
346 break;
347 }
348 }
349
350 bool emptyTf = true;
351 for (size_t pi = 0; pi < present.size(); ++pi) {
352 if (present[pi] && !ignored[pi]) {
353 emptyTf = false;
354 }
355 if (!present[pi] && !ignored[pi]) {
356 showSize[pi] = true;
357 unmatchedDescriptions.push_back(pi);
358 }
359 }
360 int timeframeCompleteness = emptyTf ? 0 : (unmatchedDescriptions.size() ? -1 : 1);
361 (void)timeframeCompleteness; // To be sent as message
362
363 if (skipAsAllFound && !doPrintSizes) {
364 return;
365 }
366
367 if (firstDH && doPrintSizes) {
368 std::string sizes = "";
369 size_t totalSize = 0;
370 for (size_t pi = 0; pi < present.size(); ++pi) {
371 if (showSize[pi]) {
372 totalSize += dataSizes[pi];
373 auto& spec = routes[pi].matcher;
374 sizes += DataSpecUtils::describe(spec) + fmt::format(":{} ", fmt::group_digits(dataSizes[pi]));
375 }
376 }
377 LOGP(important, "RAW {} size report:{}- Total:{}", firstDH->tfCounter, sizes, fmt::group_digits(totalSize));
378 }
379
380 if (!doInjectMissingData) {
381 return;
382 }
383
384 if (unmatchedDescriptions.size() > 0) {
385 if (hassih) {
386 if (firstDH) {
387 LOG(error) << "Received an EndOfStream message together with data. This should not happen.";
388 }
389 LOG(detail) << "This is an End Of Stream message. Not injecting anything.";
390 return;
391 }
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.";
394 return;
395 }
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.";
398 return;
399 }
400 std::string missing = "";
401 bool showAlarm = false;
402 uint32_t runNumber = 0;
403 try {
404 runNumber = strtoul(device.fConfig->GetProperty<std::string>("runNumber", "").c_str(), nullptr, 10);
405 } catch (...) {
406 }
407 for (auto mi : unmatchedDescriptions) {
408 auto& spec = routes[mi].matcher;
409 missing += " " + DataSpecUtils::describe(spec);
410 // If we have a ConcreteDataMatcher, we can create a message with the correct header.
411 // If we have a ConcreteDataTypeMatcher, we use 0xdeadbeef as subSpecification.
413 auto subSpec = DataSpecUtils::getOptionalSubSpec(spec);
414 if (subSpec == std::nullopt) {
415 *subSpec = 0xDEADBEEF;
416 }
417 o2::header::DataHeader dh{*firstDH};
418 dh.dataOrigin = concrete.origin;
419 dh.dataDescription = concrete.description;
420 dh.subSpecification = *subSpec;
421 dh.payloadSize = 0;
422 dh.runNumber = runNumber;
423 dh.splitPayloadParts = 0;
424 dh.splitPayloadIndex = 0;
425 dh.payloadSerializationMethod = header::gSerializationMethodNone;
426
427 auto& channelName = routes[mi].channel;
428 auto& channelInfo = device.GetChannel(channelName);
429 auto channelAlloc = o2::pmr::getTransportAllocator(channelInfo.Transport());
430 auto headerMessage = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh, *dph});
431 parts.AddPart(std::move(headerMessage));
432 // add empty payload message
433 parts.AddPart(device.NewMessageFor(channelName, 0, 0));
435 showAlarm = true;
436 }
437 }
438 static int maxWarn = 10; // Correct would be o2::conf::VerbosityConfig::Instance().maxWarnDeadBeef, but Framework does not depend on CommonUtils..., but not so critical since receives will send correct number of DEADBEEF messages
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" : "");
442 }
443 }
444}
445
446InjectorFunction dplModelAdaptor(std::vector<OutputSpec> const& filterSpecs, DPLModelAdapterConfig config)
447{
448 bool throwOnUnmatchedInputs = config.throwOnUnmatchedInputs;
449 // structure to hold information on the unmatched data and print a warning at cleanup
450 class DroppedDataSpecs
451 {
452 public:
453 DroppedDataSpecs() = default;
454 ~DroppedDataSpecs()
455 {
456 warning();
457 }
458
459 [[nodiscard]] bool find(std::string const& desc) const
460 {
461 return descriptions.find(desc) != std::string::npos;
462 }
463
464 void add(std::string const& desc)
465 {
466 descriptions += "\n " + desc;
467 }
468
469 void warning() const
470 {
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";
475 }
476 }
477
478 private:
479 std::string descriptions;
480 };
481
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) {
483 // FIXME: this in not thread safe, but better than an alloc of a map per message...
484 std::unordered_map<std::string, fair::mq::Parts> outputs;
485 std::vector<std::string> unmatchedDescriptions;
486 auto* device = services.get<RawDeviceService>().device();
487
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;
495 } else {
496 auto orbitResetTimeUrl = device->fConfig->GetProperty<std::string>("orbit-reset-time", "ccdb://CTP/Calib/OrbitResetTime");
497 char* err = nullptr;
498 creationVal = std::strtoll(orbitResetTimeUrl.c_str(), &err, 10);
499 if (err && *err == 0 && creationVal) {
500 override_creation = true;
501 }
502 }
503
504 int fmqRunNumber = -1;
505 try {
506 fmqRunNumber = atoi(device->fConfig->GetProperty<std::string>("runNumber", "").c_str());
507 } catch (...) {
508 }
509
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.";
513 continue;
514 }
515 auto* header = parts.At(msgidx)->GetData();
516 const auto dh = o2::header::get<DataHeader*>(header);
517 if (!dh) {
518 LOG(error) << "data on input " << msgidx << " does not follow the O2 data model, DataHeader missing";
519 if (msgidx > 0) {
520 --msgidx;
521 }
522 continue;
523 }
524 auto dph = o2::header::get<DataProcessingHeader*>(header);
525 if (!dph) {
526 LOG(error) << "data on input " << msgidx << " does not follow the O2 data model, DataProcessingHeader missing";
527 continue;
528 }
529 const_cast<DataProcessingHeader*>(dph)->startTime = newTimesliceId;
530 if (override_creation) {
531 const_cast<DataProcessingHeader*>(dph)->creation = creationVal + (dh->firstTForbit * o2::constants::lhc::LHCOrbitNS * 0.000001f);
532 }
533 timingInfo.timeslice = dph->startTime;
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();
539 if (DefaultsHelpers::deploymentMode() != DeploymentMode::FST && (dh->runNumber == 0 || (dh->tfCounter == 0 && dh->dataDescription.as<std::string>() != "EOS") || (fmqRunNumber > 0 && fmqRunNumber != dh->runNumber))) {
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();
543 }
544
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 = "";
549
550 for (auto const& spec : filterSpecs) {
551 // filter on the specified OutputSpecs, the default value is a ConcreteDataTypeMatcher with origin and description 'any'
553 DataSpecUtils::match(spec, query)) {
554 channelName = channelRetriever(query, dph->startTime);
555 // We do not complain about DPL/EOS/0, since it's normal not to forward it.
556 if (channelName.empty() && DataSpecUtils::describe(query) != "DPL/EOS/0") {
557 LOG(warning) << "can not find matching channel, not able to adopt " << DataSpecUtils::describe(query);
558 }
559 break;
560 }
561 }
562 finalBlockIndex = getFinalIndex(*dh, msgidx);
563 if (finalBlockIndex > parts.Size()) {
564 // TODO error handling
565 // LOGP(error, "DataHeader::splitPayloadParts invalid");
566 continue;
567 }
568
569 if (!channelName.empty()) {
570 // the checks for consistency of split payload parts are of informative nature
571 // forwarding happens independently
572 // if (dh->splitPayloadParts > 1 && dh->splitPayloadParts != std::numeric_limits<decltype(dh->splitPayloadParts)>::max()) {
573 // if (lastSplitPartIndex == -1 && dh->splitPayloadIndex != 0) {
574 // LOG(warning) << "wrong split part index, expecting the first of " << dh->splitPayloadParts << " part(s)";
575 // } else if (dh->splitPayloadIndex != lastSplitPartIndex + 1) {
576 // LOG(warning) << "unordered split parts, expecting part " << lastSplitPartIndex + 1 << ", got " << dh->splitPayloadIndex
577 // << " of " << dh->splitPayloadParts;
578 // } else if (channelNameForSplitParts.empty() == false && channelName != channelNameForSplitParts) {
579 // LOG(error) << "inconsistent channel for split part " << dh->splitPayloadIndex
580 // << ", matching " << channelName << ", expecting " << channelNameForSplitParts;
581 // }
582 //}
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)));
586 }
587 msgidx -= 2;
588 } else {
589 msgidx = finalBlockIndex - 2;
590 }
591 if (finalBlockIndex == 0 && !DataSpecUtils::match(query, "DPL", "EOS", 0)) {
592 unmatchedDescriptions.emplace_back(DataSpecUtils::describe(query));
593 }
594 } // end of loop over parts
595
596 bool didSendParts = false;
597 for (auto& [channelName, channelParts] : outputs) {
598 if (channelParts.Size() == 0) {
599 continue;
600 }
601 didSendParts = true;
602 sendOnChannel(*device, channelParts, channelName, newTimesliceId);
603 }
604 if (not unmatchedDescriptions.empty()) {
605 if (throwOnUnmatchedInputs) {
606 std::string descriptions;
607 for (auto const& desc : unmatchedDescriptions) {
608 descriptions += "\n " + desc;
609 }
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");
612 } else {
613 bool changed = false;
614 for (auto const& desc : unmatchedDescriptions) {
615 if (not droppedDataSpecs->find(desc)) {
616 // a new description
617 droppedDataSpecs->add(desc);
618 changed = true;
619 }
620 }
621 if (changed) {
622 droppedDataSpecs->warning();
623 }
624 }
625 }
626 return didSendParts;
627 };
628}
629
630InjectorFunction incrementalConverter(OutputSpec const& spec, o2::header::SerializationMethod method, uint64_t startTime, uint64_t step)
631{
632 auto timesliceId = std::make_shared<size_t>(startTime);
633 return [timesliceId, spec, step, method](TimingInfo&, ServiceRegistryRef const& services, fair::mq::Parts& parts, ChannelRetriever channelRetriever, size_t newTimesliceId, bool&) {
634 auto* device = services.get<RawDeviceService>().device();
635 uint32_t runNumber = 0;
636 try {
637 runNumber = strtoul(device->fConfig->GetProperty<std::string>("runNumber", "").c_str(), nullptr, 10);
638 } catch (...) {
639 }
640 // We iterate on all the parts and we send them two by two,
641 // adding the appropriate O2 header.
642 for (int i = 0; i < parts.Size(); ++i) {
643 DataHeader dh;
644 dh.payloadSerializationMethod = method;
645
646 // FIXME: this only supports fully specified output specs...
648 dh.dataOrigin = matcher.origin;
649 dh.dataDescription = matcher.description;
650 dh.subSpecification = matcher.subSpec;
651 dh.payloadSize = parts.At(i)->GetSize();
652 dh.runNumber = runNumber;
653
654 DataProcessingHeader dph{newTimesliceId, 0};
655 if (*timesliceId != newTimesliceId) {
656 LOG(fatal) << "Time slice ID provided from oldestPossible mechanism " << newTimesliceId << " is out of sync with expected value " << *timesliceId;
657 }
658 *timesliceId += step;
659 // we have to move the incoming data
660 o2::header::Stack headerStack{dh, dph};
661
662 sendOnChannel(*device, std::move(headerStack), std::move(parts.At(i)), spec, channelRetriever);
663 }
664 return parts.Size();
665 };
666}
667
669 std::vector<OutputSpec> const& outputs,
670 char const* defaultChannelConfig,
671 InjectorFunction converter,
672 uint64_t minSHM,
673 bool sendTFcounter,
674 bool doInjectMissingData,
675 unsigned int doPrintSizes)
676{
678 spec.name = strdup(name);
679 spec.inputs = {};
680 spec.outputs = outputs;
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);
684 // The Init method will register a new "Out of band" channel and
685 // attach an OnData to it which is responsible for converting incoming
686 // messages into DPL messages.
687 spec.algorithm = AlgorithmSpec{[converter, minSHM, deviceName = spec.name, sendTFcounter, doInjectMissingData, doPrintSizes](InitContext& ctx) {
688 auto* device = ctx.services().get<RawDeviceService>().device();
689 // make a copy of the output routes and pass to the lambda by move
690 auto outputRoutes = ctx.services().get<RawDeviceService>().spec().outputs;
691 auto outputChannels = ctx.services().get<RawDeviceService>().spec().outputChannels;
692 assert(device);
693
694 // check that the name used for registering the OnData callback corresponds
695 // to the configured output channel, unfortunately we can not automatically
696 // deduce this from list of channels without knowing the name, because there
697 // will be multiple channels. At least we throw a more informative exception.
698 // fair::mq::Device calls the custom init before the channels have been configured
699 // so we do the check before starting in a dedicated callback
700 auto channelConfigurationChecker = [device, deviceName, services = ctx.services()]() {
701 auto& deviceState = services.get<DeviceState>();
702 channels.clear();
703 numberOfEoS.clear();
704 eosPeersCount.clear();
705 for (auto& [channelName, _] : services.get<RawDeviceService>().device()->GetChannels()) {
706 // Out of band channels must start with the proxy name, at least for now
707 if (strncmp(channelName.c_str(), deviceName.c_str(), deviceName.size()) == 0) {
708 channels.push_back(channelName);
709 }
710 }
711 for (auto& channel : channels) {
712 LOGP(detail, "Injecting channel '{}' into DPL configuration", channel);
713 // Converter should pump messages
714 auto& channelPtr = services.get<RawDeviceService>().device()->GetChannel(channel, 0);
715 deviceState.inputChannelInfos.push_back(InputChannelInfo{
717 .hasPendingEvents = false,
718 .readPolled = false,
719 .channel = &channelPtr,
720 .id = {ChannelIndex::INVALID},
721 .channelType = ChannelAccountingType::RAWFMQ,
722 });
723 }
724 numberOfEoS.resize(channels.size(), 0);
725 eosPeersCount.resize(channels.size(), 0);
726 };
727
728 auto drainMessages = [](ServiceRegistryRef registry, int state) {
729 auto* device = registry.get<RawDeviceService>().device();
730 auto& deviceState = registry.get<DeviceState>();
731 // We drop messages in input only when in ready.
732 // FIXME: should we drop messages in input the first time we are in ready?
733 static bool wasRunning = false;
734 if (fair::mq::State{state} == fair::mq::State::Running) {
735 wasRunning = true;
736 }
737 if (fair::mq::State{state} != fair::mq::State::Ready || !wasRunning) {
738 return;
739 }
740 uv_update_time(deviceState.loop);
741 bool doDrain = true;
742 // Cleanup count is set by the cleanup property of the device.
743 // It is incremented every time the device is cleaned up.
744 // We use it to detect when the device is cleaned up.
745 int64_t cleanupCount = deviceState.cleanupCount.load();
746
747 // Continue iterating we saw the cleanup property being reset or
748 // the device state changing.
749 while (doDrain) {
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];
754 // We only care about rawfmq channels.
755 if (info.channelType != ChannelAccountingType::RAWFMQ) {
756 continue;
757 }
758 info.channel->Receive(parts, 10);
759 }
760 // Keep state transitions going also when running with the standalone GUI.
761 uv_run(deviceState.loop, UV_RUN_NOWAIT);
762 }
763 };
764
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);
769 }
770
771 static auto countEoS = [](fair::mq::Parts& inputs) -> int {
772 int count = 0;
773 for (int msgidx = 0; msgidx < inputs.Size() / 2; ++msgidx) {
774 // Skip when we have nullptr for the header.
775 // Not sure it can actually happen, but does not hurt.
776 if (inputs.At(msgidx * 2).get() == nullptr) {
777 continue;
778 }
779 auto const sih = o2::header::get<SourceInfoHeader*>(inputs.At(msgidx * 2)->GetData());
780 if (sih != nullptr && sih->state == InputChannelState::Completed) {
781 count++;
782 }
783 }
784 return count;
785 };
786
787 // Data handler for incoming data. Must return true if it sent any data.
788 auto dataHandler = [converter, doInjectMissingData, doPrintSizes,
789 outputRoutes = std::move(outputRoutes),
790 control = &ctx.services().get<ControlService>(),
791 deviceState = &ctx.services().get<DeviceState>(),
792 timesliceIndex = &ctx.services().get<TimesliceIndex>(),
793 outputChannels = std::move(outputChannels)](ServiceRegistryRef ref, TimingInfo& timingInfo, fair::mq::Parts& inputs, int, size_t ci, bool newRun) -> bool {
794 auto* device = ref.get<RawDeviceService>().device();
795 // pass a copy of the outputRoutes
796 auto channelRetriever = [&outputRoutes](OutputSpec const& query, DataProcessingHeader::StartTime timeslice) -> std::string const& {
797 static std::string emptyChannel = "";
798 for (auto& route : outputRoutes) {
799 LOG(debug) << "matching: " << DataSpecUtils::describe(query) << " to route " << DataSpecUtils::describe(route.matcher);
800 if (DataSpecUtils::match(route.matcher, query) && ((timeslice % route.maxTimeslices) == route.timeslice)) {
801 return route.channel;
802 }
803 }
804 return emptyChannel;
805 };
806
807 std::string const& channel = channels[ci];
808 // we buffer the condition since the converter will forward messages by move
809 int nEos = countEoS(inputs);
810 if (newRun) {
811 std::fill(numberOfEoS.begin(), numberOfEoS.end(), 0);
812 std::fill(eosPeersCount.begin(), eosPeersCount.end(), 0);
813 }
814 numberOfEoS[ci] += nEos;
815 if (numberOfEoS[ci]) {
816 eosPeersCount[ci] = std::max<int>(eosPeersCount[ci], device->GetNumberOfConnectedPeers(channel));
817 }
818 // For reference, the oldest possible timeframe passed as newTimesliceId here comes from LifetimeHelpers::enumDrivenCreation()
819 bool shouldstop = false;
820 if (doInjectMissingData) {
821 injectMissingData(*device, inputs, outputRoutes, doInjectMissingData, doPrintSizes);
822 }
823 bool didSendParts = converter(timingInfo, ref, inputs, channelRetriever, timesliceIndex->getOldestPossibleOutput().timeslice.value, shouldstop);
824
825 // If we have enough EoS messages, we can stop the device
826 // Notice that this has a number of failure modes:
827 // * If a connection sends the EoS and then closes before the GetNumberOfConnectedPeers command above.
828 // * If a connection sends two EoS.
829 // * If a connection sends an end of stream closes and another one opens.
830 // Finally, if we didn't receive an EoS this time, out counting of the connected peers is off, so the best thing we can do is delay the EoS reporting
831 bool everyEoS = shouldstop;
832 if (!shouldstop && nEos) {
833 everyEoS = true;
834 for (unsigned int i = 0; i < numberOfEoS.size(); i++) {
835 if (numberOfEoS[i] < eosPeersCount[i]) {
836 everyEoS = false;
837 break;
838 }
839 }
840 }
841
842 if (everyEoS) {
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 << ")";
844 // Mark all input channels as closed
845 for (auto& info : deviceState->inputChannelInfos) {
846 info.state = InputChannelState::Completed;
847 }
848 std::fill(numberOfEoS.begin(), numberOfEoS.end(), 0);
849 std::fill(eosPeersCount.begin(), eosPeersCount.end(), 0);
850 control->endOfStream();
851 }
852 return didSendParts;
853 };
854
855 auto runHandler = [dataHandler, minSHM, sendTFcounter](ProcessingContext& ctx) {
856 static RateLimiter limiter;
857 static size_t currentRunNumber = -1;
858 static bool inStopTransition = false;
859 bool newRun = false;
860 auto device = ctx.services().get<RawDeviceService>().device();
861 if (limiter.check(ctx, std::stoi(device->fConfig->GetValue<std::string>("timeframes-rate-limit")), minSHM)) {
862 inStopTransition = true;
863 }
864
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;
869 int maxRead = 1000;
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);
875 }
876 // Populate TimingInfo from the first message
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>();
881 if (dh != nullptr) {
882 if (currentRunNumber != -1 && dh->runNumber != 0 && dh->runNumber != currentRunNumber) {
883 newRun = true;
884 inStopTransition = false;
885 }
886 if (currentRunNumber == -1 || dh->runNumber != 0) {
887 currentRunNumber = dh->runNumber;
888 }
889 timingInfo.runNumber = dh->runNumber;
890 timingInfo.firstTForbit = dh->firstTForbit;
891 timingInfo.tfCounter = dh->tfCounter;
892 }
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;
897 }
898 if (!inStopTransition) {
899 didSendParts |= dataHandler(ctx.services(), timingInfo, parts, 0, ci, newRun);
900 }
901 if (sendTFcounter) {
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));
903 }
904 }
905 if (nReceived == 0 || channels.size() == 1) {
906 break;
907 }
908 waitTime = 0;
909 }
910 }
911 // In case we did not send any part at all, we need to rewind by one
912 // to avoid creating extra timeslices.
913 auto& decongestion = ctx.services().get<DecongestionService>();
914 decongestion.nextEnumerationTimesliceRewinded = !didSendParts;
915 if (didSendParts) {
916 ctx.services().get<MessageContext>().fakeDispatch();
917 } else {
918 decongestion.nextEnumerationTimeslice -= 1;
919 }
920 };
921
922 return runHandler;
923 }};
924 const char* d = strdup(((std::string(defaultChannelConfig).find("name=") == std::string::npos ? (std::string("name=") + name + ",") : "") + std::string(defaultChannelConfig)).c_str());
925 spec.options = {
926 ConfigParamSpec{"ready-state-policy", VariantType::String, "keep", {"What to do when the device is in ready state: *keep*, drain"}},
927 ConfigParamSpec{"channel-config", VariantType::String, d, {"Out-of-band channel config"}}};
928 return spec;
929}
930
931// Decide where to sent the output. Everything to "downstream" if there is such a channel.
932std::string defaultOutputProxyChannelSelector(InputSpec const& input, const std::unordered_map<std::string, std::vector<fair::mq::Channel>>& channels)
933{
934 return channels.count("downstream") ? "downstream" : input.binding;
935}
936
938 Inputs const& inputSpecs,
939 const char* defaultChannelConfig)
940{
942 spec.name = name;
943 spec.inputs = inputSpecs;
944 spec.outputs = {};
945 spec.algorithm = adaptStateful([inputSpecs](FairMQDeviceProxy& proxy, CallbackService& callbacks, RawDeviceService& rds, DeviceSpec const& deviceSpec, ConfigParamRegistry const& options) {
946 // we can retrieve the channel name from the channel configuration string
947 // FIXME: even if a --channel-config option is specified on the command line, always the default string
948 // is retrieved from the config registry. The channel name thus needs to be configured in the default
949 // string AND must match the name in an optional channel config.
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{}};
954 if (values.size() != 1 || values[0].empty()) {
955 throw std::runtime_error("failed to extract channel name from channel configuration parameter '" + channelConfig + "'");
956 }
957 std::string outputChannelName = values[0];
958
959 auto* device = rds.device();
960 // check that the input spec bindings have corresponding output channels
961 // fair::mq::Device calls the custom init before the channels have been configured
962 // so we do the check before starting in a dedicated callback
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 + "'");
967 }
968 };
969 callbacks.set<CallbackService::Id::Start>(channelConfigurationChecker);
970 auto lastDataProcessingHeader = std::make_shared<DataProcessingHeader>(0, 0);
971
972 auto& spec = const_cast<DeviceSpec&>(deviceSpec);
974 for (auto const& inputSpec : inputSpecs) {
975 // this is a prototype, in principle we want to have all spec objects const
976 // and so only the const object can be retrieved from service registry
977 ForwardRoute route{
978 .timeslice = 0,
979 .maxTimeslices = 1,
980 .matcher = inputSpec,
981 .channel = outputChannelName,
982 .policy = &policy};
983 spec.forwards.emplace_back(route);
984 }
985
986 auto forwardEos = [device, lastDataProcessingHeader, outputChannelName](EndOfStreamContext&) {
987 // DPL implements an internal end of stream signal, which is propagated through
988 // all downstream channels if a source is dry, make it available to other external
989 // devices via a message of type {DPL/EOS/0}
990 for (auto& channelInfo : device->GetChannels()) {
991 auto& channelName = channelInfo.first;
992 if (channelName != outputChannelName) {
993 continue;
994 }
995
996 uint32_t runNumber = 0;
997 try {
998 runNumber = strtoul(device->fConfig->GetProperty<std::string>("runNumber", "").c_str(), nullptr, 10);
999 } catch (...) {
1000 }
1001 DataHeader dh;
1002 dh.dataOrigin = "DPL";
1003 dh.dataDescription = "EOS";
1004 dh.subSpecification = 0;
1005 dh.payloadSize = 0;
1006 dh.runNumber = runNumber;
1008 dh.tfCounter = 0;
1009 dh.firstTForbit = 0;
1010 SourceInfoHeader sih;
1012 // allocate the header message using the underlying transport of the channel
1013 auto channelAlloc = o2::pmr::getTransportAllocator(channelInfo.second[0].Transport());
1014 auto headerMessage = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh, *lastDataProcessingHeader, sih});
1015 fair::mq::Parts out;
1016 out.AddPart(std::move(headerMessage));
1017 // add empty payload message
1018 out.AddPart(device->NewMessageFor(channelName, 0, 0));
1019 sendOnChannel(*device, out, channelName, (size_t)-1);
1020 }
1021 };
1022 callbacks.set<CallbackService::Id::EndOfStream>(forwardEos);
1023
1024 return adaptStateless([lastDataProcessingHeader](InputRecord& inputs) {
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);
1029 if (dph) {
1030 // FIXME: should we implement an assignment operator for DataProcessingHeader?
1031 lastDataProcessingHeader->startTime = dph->startTime;
1032 lastDataProcessingHeader->duration = dph->duration;
1033 lastDataProcessingHeader->creation = dph->creation;
1034 }
1035 }
1036 }
1037 });
1038 });
1039 const char* d = strdup(((std::string(defaultChannelConfig).find("name=") == std::string::npos ? (std::string("name=") + name + ",") : "") + std::string(defaultChannelConfig)).c_str());
1040 spec.options = {
1041 ConfigParamSpec{"channel-config", VariantType::String, d, {"Out-of-band channel config"}},
1042 };
1043
1044 return spec;
1045}
1046
1048 Inputs const& inputSpecs,
1049 const char* defaultChannelConfig,
1050 ChannelSelector channelSelector)
1051{
1052 // FIXME: this looks like a code duplication with the function above, check if the
1053 // two can be combined
1054 DataProcessorSpec spec;
1055 spec.name = name;
1056 spec.inputs = inputSpecs;
1057 spec.outputs = {};
1058 spec.algorithm = adaptStateful([inputSpecs, channelSelector](FairMQDeviceProxy& proxy, CallbackService& callbacks, RawDeviceService& rds, const DeviceSpec& deviceSpec) {
1059 auto device = rds.device();
1060 // check that the input spec bindings have corresponding output channels
1061 // fair::mq::Device calls the custom init before the channels have been configured
1062 // so we do the check before starting in a dedicated callback
1063 // also we set forwards for all input specs and keep a list of all channels so we can send EOS on them
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 + "'");
1072 }
1074 ForwardRoute route{
1075 .timeslice = 0,
1076 .maxTimeslices = 1,
1077 .matcher = spec,
1078 .channel = channel,
1079 .policy = &policy};
1080 // this we will try to fix on the framework level, there will be an API to
1081 // set external routes. Basically, this has to be added while setting up the
1082 // workflow. After that, the actual spec provided by the service is supposed
1083 // to be const by design
1084 mutableDeviceSpec.forwards.emplace_back(route);
1085
1086 channelNames->emplace_back(std::move(channel));
1087 }
1088 proxy.bind(mutableDeviceSpec.outputs, mutableDeviceSpec.inputs, mutableDeviceSpec.forwards, *device);
1089 };
1090 // We need to clear the channels on stop, because we will check and add them
1091 auto channelConfigurationDisposer = [&deviceSpec]() {
1092 auto& mutableDeviceSpec = const_cast<DeviceSpec&>(deviceSpec);
1093 mutableDeviceSpec.forwards.clear();
1094 };
1095 callbacks.set<CallbackService::Id::Start>(channelConfigurationInitializer);
1096 callbacks.set<CallbackService::Id::Stop>(channelConfigurationDisposer);
1097
1098 auto lastDataProcessingHeader = std::make_shared<DataProcessingHeader>(0, 0);
1099 auto forwardEos = [device, lastDataProcessingHeader, channelNames](EndOfStreamContext&) {
1100 // DPL implements an internal end of stream signal, which is propagated through
1101 // all downstream channels if a source is dry, make it available to other external
1102 // devices via a message of type {DPL/EOS/0}
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) {
1107 if (n == name) {
1108 return true;
1109 }
1110 }
1111 return false;
1112 };
1113 if (!checkChannel(channelName)) {
1114 continue;
1115 }
1116 uint32_t runNumber = 0;
1117 try {
1118 runNumber = strtoul(device->fConfig->GetProperty<std::string>("runNumber", "").c_str(), nullptr, 10);
1119 } catch (...) {
1120 }
1121 DataHeader dh;
1122 dh.dataOrigin = "DPL";
1123 dh.dataDescription = "EOS";
1124 dh.subSpecification = 0;
1125 dh.payloadSize = 0;
1127 dh.runNumber = runNumber;
1128 dh.tfCounter = 0;
1129 dh.firstTForbit = 0;
1130 SourceInfoHeader sih;
1132 // allocate the header message using the underlying transport of the channel
1133 auto channelAlloc = o2::pmr::getTransportAllocator(channelInfo.second[0].Transport());
1134 auto headerMessage = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh, *lastDataProcessingHeader, sih});
1135 fair::mq::Parts out;
1136 out.AddPart(std::move(headerMessage));
1137 // add empty payload message
1138 out.AddPart(device->NewMessageFor(channelName, 0, 0));
1139 LOGP(detail, "Forwarding EoS to {}", channelName);
1140 sendOnChannel(*device, out, channelName, (size_t)-1);
1141 }
1142 };
1143 callbacks.set<CallbackService::Id::EndOfStream>(forwardEos);
1144
1145 return adaptStateless([channelSelector, lastDataProcessingHeader](InputRecord& inputs) {
1146 // there is nothing to do if the forwarding is handled on the framework level
1147 // as forward routes but we need to keep a copy of the last DataProcessingHeader
1148 // for sending the EOS
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);
1153 if (dph) {
1154 // FIXME: should we implement an assignment operator for DataProcessingHeader?
1155 lastDataProcessingHeader->startTime = dph->startTime;
1156 lastDataProcessingHeader->duration = dph->duration;
1157 lastDataProcessingHeader->creation = dph->creation;
1158 }
1159 }
1160 }
1161 });
1162 });
1163 const char* d = strdup(((std::string(defaultChannelConfig).find("name=") == std::string::npos ? (std::string("name=") + name + ",") : "") + std::string(defaultChannelConfig)).c_str());
1164 spec.options = {
1165 ConfigParamSpec{"channel-config", VariantType::String, d, {"Out-of-band channel config"}},
1166 };
1167
1168 return spec;
1169}
1170
1171} // namespace o2::framework
benchmark::State & state
std::vector< OutputRoute > routes
int32_t i
Header to collect LHC related constants.
uint32_t res
Definition RawData.h:0
std::ostringstream debug
void bind(std::vector< OutputRoute > const &outputs, std::vector< InputRoute > const &inputs, std::vector< ForwardRoute > const &forwards, fair::mq::Device &device)
The input API of the Data Processing Layer This class holds the inputs which are valid for processing...
static DataRef getByPos(std::vector< InputRoute > const &routes, InputSpan const &span, int pos, int part=0)
size_t getNofParts(int pos) const
int check(ProcessingContext &ctx, int maxInFlight, size_t minSHM)
virtual fair::mq::Device * device()=0
GLdouble n
Definition glcorearb.h:1982
GLint GLsizei count
Definition glcorearb.h:399
GLuint GLsizei const GLuint const GLintptr const GLsizeiptr * sizes
Definition glcorearb.h:2595
GLuint index
Definition glcorearb.h:781
GLuint const GLchar * name
Definition glcorearb.h:781
GLenum GLsizei GLsizei GLint * values
Definition glcorearb.h:1576
typedef void(APIENTRYP PFNGLCULLFACEPROC)(GLenum mode)
GLboolean r
Definition glcorearb.h:1233
GLbitfield GLuint64 timeout
Definition glcorearb.h:1573
GLint ref
Definition glcorearb.h:291
constexpr o2::header::DataOrigin gDataOriginPHS
Definition DataHeader.h:574
constexpr o2::header::DataOrigin gDataOriginHMP
Definition DataHeader.h:569
constexpr o2::header::DataOrigin gDataOriginEMC
Definition DataHeader.h:565
constexpr o2::header::DataOrigin gDataOriginAny
Definition DataHeader.h:560
constexpr o2::header::DataDescription gDataDescriptionAny
Definition DataHeader.h:595
constexpr double LHCOrbitNS
Defining PrimaryVertex explicitly as messageable.
Definition TFIDInfo.h:20
void injectMissingData(fair::mq::Device &device, fair::mq::Parts &parts, std::vector< OutputRoute > const &routes, bool doInjectMissingData, unsigned int doPrintSizes)
DataProcessorSpec specifyExternalFairMQDeviceProxy(char const *label, std::vector< OutputSpec > const &outputs, const char *defaultChannelConfig, InjectorFunction converter, uint64_t minSHM=0, bool sendTFcounter=false, bool doInjectMissingData=false, unsigned int doPrintSizes=0)
@ RAWFMQ
A raw FairMQ channel which is not accounted by the framework.
std::function< std::string(InputSpec const &input, const std::unordered_map< std::string, std::vector< fair::mq::Channel > > &channels)> ChannelSelector
std::function< bool(TimingInfo &, ServiceRegistryRef const &services, fair::mq::Parts &inputs, ChannelRetriever, size_t newTimesliceId, bool &stop)> InjectorFunction
DataProcessorSpec specifyFairMQDeviceMultiOutputProxy(char const *label, Inputs const &inputSpecs, const char *defaultChannelConfig, ChannelSelector channelSelector=defaultOutputProxyChannelSelector)
InjectorFunction dplModelAdaptor(std::vector< OutputSpec > const &specs={{header::gDataOriginAny, header::gDataDescriptionAny}}, DPLModelAdapterConfig config=DPLModelAdapterConfig{})
InjectorFunction o2DataModelAdaptor(OutputSpec const &spec, uint64_t startTime, uint64_t step)
DataProcessorSpec specifyFairMQDeviceOutputProxy(char const *label, Inputs const &inputSpecs, const char *defaultChannelConfig)
void sendOnChannel(fair::mq::Device &device, o2::header::Stack &&headerStack, fair::mq::MessagePtr &&payloadMessage, OutputSpec const &spec, ChannelRetriever &channelRetriever)
@ Completed
The channel was signaled it will not receive any data.
@ Running
The channel is actively receiving data.
std::string formatExternalChannelConfiguration(InputChannelSpec const &)
helper method to format a configuration string for an external channel
std::function< std::string const &(OutputSpec const &, DataProcessingHeader::StartTime)> ChannelRetriever
AlgorithmSpec::ProcessCallback adaptStateless(LAMBDA l)
std::string defaultOutputProxyChannelSelector(InputSpec const &input, const std::unordered_map< std::string, std::vector< fair::mq::Channel > > &channels)
Default way to select an output channel for multi-output proxy.
auto getFinalIndex(DataHeader const &dh, size_t msgidx) -> size_t
std::vector< InputSpec > Inputs
void appendForSending(fair::mq::Device &device, o2::header::Stack &&headerStack, size_t timeSliceID, fair::mq::MessagePtr &&payloadMessage, OutputSpec const &spec, fair::mq::Parts &messageCache, ChannelRetriever &channelRetriever)
AlgorithmSpec::InitCallback adaptStateful(LAMBDA l)
InjectorFunction incrementalConverter(OutputSpec const &spec, o2::header::SerializationMethod method, uint64_t startTime, uint64_t step)
constexpr o2::header::SerializationMethod gSerializationMethodNone
Definition DataHeader.h:327
fair::mq::MessagePtr getMessage(ContainerT &&container, FairMQMemoryResource *targetResource=nullptr)
static constexpr int INVALID
header::DataHeader::SubSpecificationType subSpec
bool throwOnUnmatchedInputs
throw runtime error if an input message is not matched by filter rules
static std::string describe(InputSpec const &spec)
static ConcreteDataTypeMatcher asConcreteDataTypeMatcher(OutputSpec const &spec)
static ConcreteDataMatcher asConcreteDataMatcher(InputSpec const &input)
static std::optional< header::DataHeader::SubSpecificationType > getOptionalSubSpec(OutputSpec const &spec)
Get the subspec, if available.
static bool match(InputSpec const &spec, ConcreteDataMatcher const &target)
static DeploymentMode deploymentMode()
static std::string inputChannel2String(const InputChannelSpec &channel)
Helper to provide the channel configuration string for an input channel.
static std::string outputChannel2String(const OutputChannelSpec &channel)
Helper to provide the channel configuration string for an output channel.
std::vector< ForwardRoute > forwards
Definition DeviceSpec.h:64
Running state information of a given device.
Definition DeviceState.h:34
static ForwardingPolicy createDefaultForwardingPolicy()
std::string binding
A mnemonic name for the input spec.
Definition InputSpec.h:66
a BaseHeader with state information from the source
the main header struct
Definition DataHeader.h:618
TFCounterType tfCounter
Definition DataHeader.h:679
SerializationMethod payloadSerializationMethod
Definition DataHeader.h:651
TForbitType firstTForbit
Definition DataHeader.h:674
DataDescription dataDescription
Definition DataHeader.h:636
SubSpecificationType subSpecification
Definition DataHeader.h:656
PayloadSizeType payloadSize
Definition DataHeader.h:666
RunNumberType runNumber
Definition DataHeader.h:684
a move-only header stack with serialized headers This is the flat buffer where all the headers in a m...
Definition Stack.h:33
LOG(info)<< "Compressed in "<< sw.CpuTime()<< " s"
std::vector< ChannelData > channels