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