Project
Loading...
Searching...
No Matches
runDataProcessing.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.
11#include <memory>
13#define BOOST_BIND_GLOBAL_PLACEHOLDERS
14#include <stdexcept>
38#include "DeviceStateHelpers.h"
41#include "Framework/DebugGUI.h"
44#include "Framework/Logger.h"
48#include "Framework/Signpost.h"
69#include "DriverServerContext.h"
70#include "HTTPParser.h"
71#include "DPLWebSocket.h"
72#include "ArrowSupport.h"
74
77#include "DDSConfigHelpers.h"
78#include "O2ControlHelpers.h"
79#include "DeviceSpecHelpers.h"
80#include "GraphvizHelpers.h"
81#include "MermaidHelpers.h"
82#include "PropertyTreeHelpers.h"
85
86#include <Configuration/ConfigurationInterface.h>
87#include <Configuration/ConfigurationFactory.h>
88#include <Monitoring/MonitoringFactory.h>
90
91#include <fairmq/Device.h>
92#include <fairmq/DeviceRunner.h>
93#include <fairmq/shmem/Monitor.h>
94#include <fairmq/ProgOptions.h>
95
96#include <boost/program_options.hpp>
97#include <boost/program_options/options_description.hpp>
98#include <boost/program_options/variables_map.hpp>
99#include <boost/exception/diagnostic_information.hpp>
100#include <boost/property_tree/json_parser.hpp>
101
102#include <uv.h>
103#include <TEnv.h>
104#include <TSystem.h>
105
106#include <cinttypes>
107#include <cstdint>
108#include <cstdio>
109#include <cstdlib>
110#include <cstring>
111#include <csignal>
112#include <iostream>
113#include <map>
114#include <regex>
115#include <set>
116#include <string>
117#include <type_traits>
118#include <tuple>
119#include <chrono>
120#include <utility>
121#include <numeric>
122#include <functional>
123
124#include <fcntl.h>
125#include <netinet/ip.h>
126#include <sys/resource.h>
127#include <sys/select.h>
128#include <sys/socket.h>
129#include <sys/stat.h>
130#include <sys/time.h>
131#include <sys/types.h>
132#include <sys/un.h>
133#include <sys/wait.h>
134#include <unistd.h>
135#include <execinfo.h>
136#include <cfenv>
137#if defined(__linux__) && __has_include(<sched.h>)
138#include <sched.h>
139#elif __has_include(<linux/getcpu.h>)
140#include <linux/getcpu.h>
141#elif __has_include(<cpuid.h>) && (__x86_64__ || __i386__)
142#include <cpuid.h>
143#define CPUID(INFO, LEAF, SUBLEAF) __cpuid_count(LEAF, SUBLEAF, INFO[0], INFO[1], INFO[2], INFO[3])
144#define GETCPU(CPU) \
145 { \
146 uint32_t CPUInfo[4]; \
147 CPUID(CPUInfo, 1, 0); \
148 /* CPUInfo[1] is EBX, bits 24-31 are APIC ID */ \
149 if ((CPUInfo[3] & (1 << 9)) == 0) { \
150 CPU = -1; /* no APIC on chip */ \
151 } else { \
152 CPU = (unsigned)CPUInfo[1] >> 24; \
153 } \
154 if (CPU < 0) \
155 CPU = 0; \
156 }
157#endif
158
159using namespace o2::monitoring;
160using namespace o2::configuration;
161
162using namespace o2::framework;
163namespace bpo = boost::program_options;
164using DataProcessorInfos = std::vector<DataProcessorInfo>;
165using DeviceExecutions = std::vector<DeviceExecution>;
166using DeviceSpecs = std::vector<DeviceSpec>;
167using DeviceInfos = std::vector<DeviceInfo>;
168using DataProcessingStatesInfos = std::vector<DataProcessingStates>;
169using DeviceControls = std::vector<DeviceControl>;
170using DataProcessorSpecs = std::vector<DataProcessorSpec>;
171
172std::vector<DeviceMetricsInfo> gDeviceMetricsInfos;
173
174// FIXME: probably find a better place
175// these are the device options added by the framework, but they can be
176// overloaded in the config spec
177bpo::options_description gHiddenDeviceOptions("Hidden child options");
178
181
182void doBoostException(boost::exception& e, const char*);
184void doUnknownException(std::string const& s, char const*);
185
186char* getIdString(int argc, char** argv)
187{
188 for (int argi = 0; argi < argc; argi++) {
189 if (strcmp(argv[argi], "--id") == 0 && argi + 1 < argc) {
190 return argv[argi + 1];
191 }
192 }
193 return nullptr;
194}
195
196int callMain(int argc, char** argv, int (*mainNoCatch)(int, char**))
197{
198 static bool noCatch = getenv("O2_NO_CATCHALL_EXCEPTIONS") && strcmp(getenv("O2_NO_CATCHALL_EXCEPTIONS"), "0");
199 int result = 1;
200 if (noCatch) {
201 try {
202 result = mainNoCatch(argc, argv);
204 doDPLException(ref, argv[0]);
205 throw;
206 }
207 } else {
208 try {
209 // The 0 here is an int, therefore having the template matching in the
210 // SFINAE expression above fit better the version which invokes user code over
211 // the default one.
212 // The default policy is a catch all pub/sub setup to be consistent with the past.
213 result = mainNoCatch(argc, argv);
214 } catch (boost::exception& e) {
215 doBoostException(e, argv[0]);
216 throw;
217 } catch (std::exception const& error) {
218 doUnknownException(error.what(), argv[0]);
219 throw;
221 doDPLException(ref, argv[0]);
222 throw;
223 } catch (...) {
224 doUnknownException("", argv[0]);
225 throw;
226 }
227 }
228 return result;
229}
230
231// Read from a given fd and print it.
232// return true if we can still read from it,
233// return false if we need to close the input pipe.
234//
235// FIXME: We should really print full lines.
236void getChildData(int infd, DeviceInfo& outinfo)
237{
238 char buffer[1024 * 16];
239 int bytes_read;
240 // NOTE: do not quite understand read ends up blocking if I read more than
241 // once. Oh well... Good enough for now.
242 int64_t total_bytes_read = 0;
243 int64_t count = 0;
244 bool once = false;
245 while (true) {
246 bytes_read = read(infd, buffer, 1024 * 16);
247 if (bytes_read == 0) {
248 return;
249 }
250 if (!once) {
251 once = true;
252 }
253 if (bytes_read < 0) {
254 return;
255 }
256 assert(bytes_read > 0);
257 outinfo.unprinted.append(buffer, bytes_read);
258 count++;
259 }
260}
261
265bool checkIfCanExit(std::vector<DeviceInfo> const& infos)
266{
267 if (infos.empty()) {
268 return false;
269 }
270 for (auto& info : infos) {
271 if (info.readyToQuit == false) {
272 return false;
273 }
274 }
275 return true;
276}
277
278// Kill all the active children. Exit code
279// is != 0 if any of the children had an error.
280void killChildren(std::vector<DeviceInfo>& infos, int sig)
281{
282 for (auto& info : infos) {
283 if (info.active == true) {
284 kill(info.pid, sig);
285 }
286 }
287}
288
290bool areAllChildrenGone(std::vector<DeviceInfo>& infos)
291{
292 for (auto& info : infos) {
293 if ((info.pid != 0) && info.active) {
294 return false;
295 }
296 }
297 return true;
298}
299
301namespace
302{
303int calculateExitCode(DriverInfo& driverInfo, DeviceSpecs& deviceSpecs, DeviceInfos& infos)
304{
305 std::regex regexp(R"(^\[([\d+:]*)\]\[\w+\] )");
306 if (!driverInfo.lastError.empty()) {
307 LOGP(error, "SEVERE: DPL driver encountered an error while running.\n{}",
308 driverInfo.lastError);
309 return 1;
310 }
311 for (size_t di = 0; di < deviceSpecs.size(); ++di) {
312 auto& info = infos[di];
313 auto& spec = deviceSpecs[di];
314 if (info.maxLogLevel >= driverInfo.minFailureLevel) {
315 LOGP(error, "SEVERE: Device {} ({}) had at least one message above severity {}: {}",
316 spec.name,
317 info.pid,
318 (int)info.minFailureLevel,
319 std::regex_replace(info.firstSevereError, regexp, ""));
320 return 1;
321 }
322 if (info.exitStatus != 0) {
323 LOGP(error, "SEVERE: Device {} ({}) returned with {}",
324 spec.name,
325 info.pid,
326 info.exitStatus);
327 return info.exitStatus;
328 }
329 }
330 return 0;
331}
332} // namespace
333
334void createPipes(int* pipes)
335{
336 auto p = pipe(pipes);
337
338 if (p == -1) {
339 std::cerr << "Unable to create PIPE: ";
340 switch (errno) {
341 case EFAULT:
342 assert(false && "EFAULT while reading from pipe");
343 break;
344 case EMFILE:
345 std::cerr << "Too many active descriptors";
346 break;
347 case ENFILE:
348 std::cerr << "System file table is full";
349 break;
350 default:
351 std::cerr << "Unknown PIPE" << std::endl;
352 };
353 // Kill immediately both the parent and all the children
354 kill(-1 * getpid(), SIGKILL);
355 }
356}
357
358// We don't do anything in the signal handler but
359// we simply note down the fact a signal arrived.
360// All the processing is done by the state machine.
361volatile sig_atomic_t graceful_exit = false;
362volatile sig_atomic_t forceful_exit = false;
363volatile sig_atomic_t sigchld_requested = false;
364volatile sig_atomic_t double_sigint = false;
365
366static void handle_sigint(int)
367{
368 if (graceful_exit == false) {
369 graceful_exit = true;
370 } else {
371 forceful_exit = true;
372 // We keep track about forceful exiting via
373 // a double SIGINT, so that we do not print
374 // any extra message. This means that if the
375 // forceful_exit is set by the timer, we will
376 // get an error message about each child which
377 // did not gracefully exited.
378 double_sigint = true;
379 }
380}
381
383void cleanupSHM(std::string const& uniqueWorkflowId)
384{
385 using namespace fair::mq::shmem;
386 fair::mq::shmem::Monitor::Cleanup(SessionId{"dpl_" + uniqueWorkflowId}, false);
387}
388
389static void handle_sigchld(int) { sigchld_requested = true; }
390
392 std::string const&,
393 DeviceSpec const& spec,
396 DeviceInfos& deviceInfos,
397 DataProcessingStatesInfos& allStates)
398{
399 LOG(info) << "Starting " << spec.id << " as remote device";
400 DeviceInfo info{
401 .pid = 0,
402 .historyPos = 0,
403 .historySize = 1000,
404 .maxLogLevel = LogParsingHelpers::LogLevel::Debug,
405 .active = true,
406 .readyToQuit = false,
407 .inputChannelMetricsViewIndex = Metric2DViewIndex{"oldest_possible_timeslice", 0, 0, {}},
408 .outputChannelMetricsViewIndex = Metric2DViewIndex{"oldest_possible_output", 0, 0, {}},
409 .lastSignal = uv_hrtime() - 10000000};
410
411 deviceInfos.emplace_back(info);
412 timespec now;
413 clock_gettime(CLOCK_REALTIME, &now);
414 uint64_t offset = now.tv_sec * 1000 - uv_now(loop);
415 allStates.emplace_back(TimingHelpers::defaultRealtimeBaseConfigurator(offset, loop),
417 // Let's add also metrics information for the given device
419}
420
426
427void log_callback(uv_poll_t* handle, int status, int events)
428{
429 O2_SIGNPOST_ID_FROM_POINTER(sid, driver, handle->loop);
430 auto* logContext = reinterpret_cast<DeviceLogContext*>(handle->data);
431 std::vector<DeviceInfo>* infos = logContext->serverContext->infos;
432 DeviceInfo& info = infos->at(logContext->index);
433
434 if (status < 0) {
435 info.active = false;
436 }
437 if (events & UV_READABLE) {
438 getChildData(logContext->fd, info);
439 }
440 if (events & UV_DISCONNECT) {
441 info.active = false;
442 }
443 O2_SIGNPOST_EVENT_EMIT(driver, sid, "loop", "log_callback invoked by poller for device %{xcode:pid}d which is %{public}s%{public}s",
444 info.pid, info.active ? "active" : "inactive",
445 info.active ? " and still has data to read." : ".");
446 if (info.active == false) {
447 uv_poll_stop(handle);
448 }
449 uv_async_send(logContext->serverContext->asyncLogProcessing);
450}
451
453{
454 O2_SIGNPOST_ID_FROM_POINTER(sid, driver, handle->loop);
455 O2_SIGNPOST_EVENT_EMIT(driver, sid, "mainloop", "close_websocket");
456 delete (WSDPLHandler*)handle->data;
457}
458
459void websocket_callback(uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf)
460{
461 O2_SIGNPOST_ID_FROM_POINTER(sid, driver, stream->loop);
462 O2_SIGNPOST_EVENT_EMIT(driver, sid, "mainloop", "websocket_callback");
463 auto* handler = (WSDPLHandler*)stream->data;
464 if (nread == 0) {
465 return;
466 }
467 if (nread == UV_EOF) {
468 if (buf->base) {
469 free(buf->base);
470 }
471 uv_read_stop(stream);
473 return;
474 }
475 if (nread < 0) {
476 // FIXME: should I close?
477 LOG(error) << "websocket_callback: Error while reading from websocket";
478 if (buf->base) {
479 free(buf->base);
480 }
481 uv_read_stop(stream);
483 return;
484 }
485 try {
486 LOG(debug3) << "Parsing request with " << handler << " with " << nread << " bytes";
487 parse_http_request(buf->base, nread, handler);
488 if (buf->base) {
489 free(buf->base);
490 }
491 } catch (WSError& e) {
492 LOG(error) << "Error while parsing request: " << e.message;
493 handler->error(e.code, e.message.c_str());
494 free(buf->base);
495 }
496}
497
498static void my_alloc_cb(uv_handle_t*, size_t suggested_size, uv_buf_t* buf)
499{
500 buf->base = (char*)malloc(suggested_size);
501 buf->len = suggested_size;
502}
503
505void ws_connect_callback(uv_stream_t* server, int status)
506{
507 O2_SIGNPOST_ID_FROM_POINTER(sid, driver, server->loop);
508 O2_SIGNPOST_EVENT_EMIT(driver, sid, "mainloop", "websocket_callback");
509 auto* serverContext = reinterpret_cast<DriverServerContext*>(server->data);
510 if (status < 0) {
511 LOGF(error, "New connection error %s\n", uv_strerror(status));
512 // error!
513 return;
514 }
515
516 auto* client = (uv_tcp_t*)malloc(sizeof(uv_tcp_t));
517 uv_tcp_init(serverContext->loop, client);
518 if (uv_accept(server, (uv_stream_t*)client) == 0) {
519 client->data = new WSDPLHandler((uv_stream_t*)client, serverContext);
520 uv_read_start((uv_stream_t*)client, (uv_alloc_cb)my_alloc_cb, websocket_callback);
521 } else {
522 uv_close((uv_handle_t*)client, nullptr);
523 }
524}
525
527 std::string configuration;
528 int fd;
529};
530
531void stream_config(uv_work_t* req)
532{
533 auto* context = (StreamConfigContext*)req->data;
534 size_t result = write(context->fd, context->configuration.data(), context->configuration.size());
535 if (result != context->configuration.size()) {
536 LOG(error) << "Unable to pass configuration to children";
537 }
538 {
539 auto error = fsync(context->fd);
540 switch (error) {
541 case EBADF:
542 LOGP(error, "EBADF while flushing child stdin");
543 break;
544 case EINVAL:
545 LOGP(error, "EINVAL while flushing child stdin");
546 break;
547 case EINTR:
548 LOGP(error, "EINTR while flushing child stdin");
549 break;
550 case EIO:
551 LOGP(error, "EIO while flushing child stdin");
552 break;
553 default:;
554 }
555 }
556 {
557 auto error = close(context->fd); // Not allowing further communication...
558 switch (error) {
559 case EBADF:
560 LOGP(error, "EBADF while closing child stdin");
561 break;
562 case EINTR:
563 LOGP(error, "EINTR while closing child stdin");
564 break;
565 case EIO:
566 LOGP(error, "EIO while closing child stdin");
567 break;
568 default:;
569 }
570 }
571}
572
573struct DeviceRef {
574 int index;
575};
576
580};
581
583{
584 struct sigaction sa_handle_int;
585 sa_handle_int.sa_handler = handle_sigint;
586 sigemptyset(&sa_handle_int.sa_mask);
587 sa_handle_int.sa_flags = SA_RESTART;
588 if (sigaction(SIGINT, &sa_handle_int, nullptr) == -1) {
589 perror("Unable to install signal handler");
590 exit(1);
591 }
592 struct sigaction sa_handle_term;
593 sa_handle_term.sa_handler = handle_sigint;
594 sigemptyset(&sa_handle_term.sa_mask);
595 sa_handle_term.sa_flags = SA_RESTART;
596 if (sigaction(SIGTERM, &sa_handle_int, nullptr) == -1) {
597 perror("Unable to install signal handler");
598 exit(1);
599 }
600}
601
603 std::string const& forwardedStdin,
604 std::vector<DeviceStdioContext>& childFds,
605 std::vector<uv_poll_t*>& handles)
606{
607 for (size_t i = 0; i < childFds.size(); ++i) {
608 auto& childstdin = childFds[i].childstdin;
609 auto& childstdout = childFds[i].childstdout;
610
611 auto* req = (uv_work_t*)malloc(sizeof(uv_work_t));
612 req->data = new StreamConfigContext{forwardedStdin, childstdin[1]};
613 uv_queue_work(serverContext->loop, req, stream_config, nullptr);
614
615 // Setting them to non-blocking to avoid haing the driver hang when
616 // reading from child.
617 int resultCode = fcntl(childstdout[0], F_SETFL, O_NONBLOCK);
618 if (resultCode == -1) {
619 LOGP(error, "Error while setting the socket to non-blocking: {}", strerror(errno));
620 }
621
623 auto addPoller = [&handles, &serverContext](int index, int fd) {
624 auto* context = new DeviceLogContext{};
625 context->index = index;
626 context->fd = fd;
627 context->serverContext = serverContext;
628 handles.push_back((uv_poll_t*)malloc(sizeof(uv_poll_t)));
629 auto handle = handles.back();
630 handle->data = context;
631 uv_poll_init(serverContext->loop, handle, fd);
632 uv_poll_start(handle, UV_READABLE, log_callback);
633 };
634
635 addPoller(i, childstdout[0]);
636 }
637}
638
639void handle_crash(int sig)
640{
641 // dump demangled stack trace
642 void* array[1024];
643 int size = backtrace(array, 1024);
644
645 {
646 char buffer[1024];
647 char const* msg = "*** Program crashed (%s)\nBacktrace by DPL:\n";
648 snprintf(buffer, 1024, msg, strsignal(sig));
649 if (sig == SIGFPE) {
650 if (std::fetestexcept(FE_DIVBYZERO)) {
651 snprintf(buffer, 1024, msg, "FLOATING POINT EXCEPTION - DIVISION BY ZERO");
652 } else if (std::fetestexcept(FE_INVALID)) {
653 snprintf(buffer, 1024, msg, "FLOATING POINT EXCEPTION - INVALID RESULT");
654 } else {
655 snprintf(buffer, 1024, msg, "FLOATING POINT EXCEPTION - UNKNOWN REASON");
656 }
657 }
658 auto retVal = write(STDERR_FILENO, buffer, strlen(buffer));
659 (void)retVal;
660 }
662 {
663 char const* msg = "Backtrace complete.\n";
664 int len = strlen(msg); /* the byte length of the string */
665
666 auto retVal = write(STDERR_FILENO, msg, len);
667 (void)retVal;
668 fsync(STDERR_FILENO);
669 }
670 _exit(1);
671}
672
677 std::vector<DeviceSpec> const& specs,
678 DriverInfo& driverInfo,
679 std::vector<DeviceControl>&,
680 std::vector<DeviceExecution>& executions,
681 std::vector<DeviceInfo>& deviceInfos,
682 std::vector<DataProcessingStates>& allStates,
683 ServiceRegistryRef serviceRegistry,
684 boost::program_options::variables_map& varmap,
685 std::vector<DeviceStdioContext>& childFds,
686 unsigned parentCPU,
687 unsigned parentNode)
688{
689 // FIXME: this might not work when more than one DPL driver on the same
690 // machine. Hopefully we do not care.
691 // Not how the first port is actually used to broadcast clients.
692 auto& spec = specs[ref.index];
693 auto& execution = executions[ref.index];
694
695 for (auto& service : spec.services) {
696 if (service.preFork != nullptr) {
697 service.preFork(serviceRegistry, DeviceConfig{varmap});
698 }
699 }
700 // If we have a framework id, it means we have already been respawned
701 // and that we are in a child. If not, we need to fork and re-exec, adding
702 // the framework-id as one of the options.
703 pid_t id = 0;
704 id = fork();
705 // We are the child: prepare options and reexec.
706 if (id == 0) {
707 // We allow being debugged and do not terminate on SIGTRAP
708 signal(SIGTRAP, SIG_IGN);
709 // We immediately ignore SIGUSR1 and SIGUSR2 so that we do not
710 // get killed by the parent trying to force stepping children.
711 // We will re-enable them later on, when it is actually safe to
712 // do so.
713 signal(SIGUSR1, SIG_IGN);
714 signal(SIGUSR2, SIG_IGN);
715
716 // This is the child.
717 // For stdout / stderr, we close the read part of the pipe, the
718 // old descriptor, and then replace it with the write part of the pipe.
719 // For stdin, we close the write part of the pipe, the old descriptor,
720 // and then we replace it with the read part of the pipe.
721 // We also close all the filedescriptors for our sibilings.
722 struct rlimit rlim;
723 getrlimit(RLIMIT_NOFILE, &rlim);
724 // We close all FD, but the one which are actually
725 // used to communicate with the driver. This is a bad
726 // idea in the first place, because rlim_cur could be huge
727 // FIXME: I should understand which one is really to be closed and use
728 // CLOEXEC on it.
729 int rlim_cur = std::min((int)rlim.rlim_cur, 10000);
730 for (int i = 0; i < rlim_cur; ++i) {
731 if (childFds[ref.index].childstdin[0] == i) {
732 continue;
733 }
734 if (childFds[ref.index].childstdout[1] == i) {
735 continue;
736 }
737 close(i);
738 }
739 dup2(childFds[ref.index].childstdin[0], STDIN_FILENO);
740 dup2(childFds[ref.index].childstdout[1], STDOUT_FILENO);
741 dup2(childFds[ref.index].childstdout[1], STDERR_FILENO);
742
743 for (auto& service : spec.services) {
744 if (service.postForkChild != nullptr) {
745 service.postForkChild(serviceRegistry);
746 }
747 }
748 for (auto& env : execution.environ) {
749 putenv(strdup(DeviceSpecHelpers::reworkTimeslicePlaceholder(env, spec).data()));
750 }
751 execvp(execution.args[0], execution.args.data());
752 } else {
753 O2_SIGNPOST_ID_GENERATE(sid, driver);
754 O2_SIGNPOST_EVENT_EMIT(driver, sid, "spawnDevice", "New child at %{pid}d", id);
755 }
756 close(childFds[ref.index].childstdin[0]);
757 close(childFds[ref.index].childstdout[1]);
758 if (varmap.count("post-fork-command")) {
759 auto templateCmd = varmap["post-fork-command"];
760 auto cmd = fmt::format(fmt::runtime(templateCmd.as<std::string>()),
761 fmt::arg("pid", id),
762 fmt::arg("id", spec.id),
763 fmt::arg("cpu", parentCPU),
764 fmt::arg("node", parentNode),
765 fmt::arg("name", spec.name),
766 fmt::arg("timeslice0", spec.inputTimesliceId),
767 fmt::arg("timeslice1", spec.inputTimesliceId + 1),
768 fmt::arg("rank0", spec.rank),
769 fmt::arg("maxRank0", spec.nSlots));
770 int err = system(cmd.c_str());
771 if (err) {
772 LOG(error) << "Post fork command `" << cmd << "` returned with status " << err;
773 }
774 LOG(debug) << "Successfully executed `" << cmd;
775 }
776 // This is the parent. We close the write end of
777 // the child pipe and and keep track of the fd so
778 // that we can later select on it.
779 for (auto& service : spec.services) {
780 if (service.postForkParent != nullptr) {
781 service.postForkParent(serviceRegistry);
782 }
783 }
784
785 LOG(info) << "Starting " << spec.id << " on pid " << id;
786 deviceInfos.push_back({.pid = id,
787 .historyPos = 0,
788 .historySize = 1000,
789 .maxLogLevel = LogParsingHelpers::LogLevel::Debug,
790 .minFailureLevel = driverInfo.minFailureLevel,
791 .active = true,
792 .readyToQuit = false,
793 .inputChannelMetricsViewIndex = Metric2DViewIndex{"oldest_possible_timeslice", 0, 0, {}},
794 .outputChannelMetricsViewIndex = Metric2DViewIndex{"oldest_possible_output", 0, 0, {}},
795 .lastSignal = uv_hrtime() - 10000000});
796 // create the offset using uv_hrtime
797 timespec now;
798 clock_gettime(CLOCK_REALTIME, &now);
799 uint64_t offset = now.tv_sec * 1000 - uv_now(loop);
800 allStates.emplace_back(
803
804 allStates.back().registerState(DataProcessingStates::StateSpec{
805 .name = "data_queries",
806 .stateId = (short)ProcessingStateId::DATA_QUERIES,
807 .sendInitialValue = true,
808 });
809 allStates.back().registerState(DataProcessingStates::StateSpec{
810 .name = "output_matchers",
811 .stateId = (short)ProcessingStateId::OUTPUT_MATCHERS,
812 .sendInitialValue = true,
813 });
814
815 for (size_t i = 0; i < DefaultsHelpers::pipelineLength(); ++i) {
816 allStates.back().registerState(DataProcessingStates::StateSpec{
817 .name = fmt::format("matcher_variables/{}", i),
818 .stateId = static_cast<short>((short)(ProcessingStateId::CONTEXT_VARIABLES_BASE) + i),
819 .minPublishInterval = 200, // if we publish too often we flood the GUI and we are not able to read it in any case
820 .sendInitialValue = true,
821 });
822 }
823
824 for (size_t i = 0; i < DefaultsHelpers::pipelineLength(); ++i) {
825 allStates.back().registerState(DataProcessingStates::StateSpec{
826 .name = fmt::format("data_relayer/{}", i),
827 .stateId = static_cast<short>((short)(ProcessingStateId::DATA_RELAYER_BASE) + i),
828 .minPublishInterval = 200, // if we publish too often we flood the GUI and we are not able to read it in any case
829 .sendInitialValue = true,
830 });
831 }
832
833 // Let's add also metrics information for the given device
835}
836
838 DriverInfo& driverInfo,
839 DeviceInfos& infos,
840 DeviceSpecs const& specs,
841 DeviceControls& controls)
842{
843 // Display part. All you need to display should actually be in
844 // `infos`.
845 // TODO: split at \n
846 // TODO: update this only once per 1/60 of a second or
847 // things like this.
848 // TODO: have multiple display modes
849 // TODO: graphical view of the processing?
850 assert(infos.size() == controls.size());
851 ParsedMetricMatch metricMatch;
852
853 int processed = 0;
854 for (size_t di = 0, de = infos.size(); di < de; ++di) {
855 DeviceInfo& info = infos[di];
856 DeviceControl& control = controls[di];
857 assert(specs.size() == infos.size());
858 DeviceSpec const& spec = specs[di];
859
860 if (info.unprinted.empty()) {
861 continue;
862 }
863 processed++;
864
865 O2_SIGNPOST_ID_FROM_POINTER(sid, driver, &info);
866 O2_SIGNPOST_START(driver, sid, "bytes_processed", "bytes processed by %{xcode:pid}d", info.pid);
867
868 std::string_view s = info.unprinted;
869 size_t pos = 0;
870 info.history.resize(info.historySize);
871 info.historyLevel.resize(info.historySize);
872
873 while ((pos = s.find("\n")) != std::string::npos) {
874 std::string_view token{s.substr(0, pos)};
875 auto logLevel = LogParsingHelpers::parseTokenLevel(token);
876
877 // Check if the token is a metric from SimpleMetricsService
878 // if yes, we do not print it out and simply store it to be displayed
879 // in the GUI.
880 // Then we check if it is part of our Poor man control system
881 // if yes, we execute the associated command.
882 if (!control.quiet && (token.find(control.logFilter) != std::string::npos) && logLevel >= info.logLevel) {
883 assert(info.historyPos >= 0);
884 assert(info.historyPos < info.history.size());
885 info.history[info.historyPos] = token;
886 info.historyLevel[info.historyPos] = logLevel;
887 info.historyPos = (info.historyPos + 1) % info.history.size();
888 fmt::print("[{}:{}]: {}\n", info.pid, spec.id, token);
889 }
890 // We keep track of the maximum log error a
891 // device has seen.
892 bool maxLogLevelIncreased = false;
893 if (logLevel > info.maxLogLevel && logLevel > LogParsingHelpers::LogLevel::Info &&
894 logLevel != LogParsingHelpers::LogLevel::Unknown) {
895 info.maxLogLevel = logLevel;
896 maxLogLevelIncreased = true;
897 }
898 if (logLevel >= driverInfo.minFailureLevel) {
899 info.lastError = token;
900 if (info.firstSevereError.empty() || maxLogLevelIncreased) {
901 info.firstSevereError = token;
902 }
903 }
904 // +1 is to skip the \n
905 s.remove_prefix(pos + 1);
906 }
907 size_t oldSize = info.unprinted.size();
908 info.unprinted = std::string(s);
909 int64_t bytesProcessed = oldSize - info.unprinted.size();
910 O2_SIGNPOST_END(driver, sid, "bytes_processed", "bytes processed by %{xcode:network-size-in-bytes}" PRIi64, bytesProcessed);
911 }
912 if (processed == 0) {
913 O2_SIGNPOST_ID_FROM_POINTER(lid, driver, loop);
914 O2_SIGNPOST_EVENT_EMIT(driver, lid, "mainloop", "processChildrenOutput invoked for nothing!");
915 }
916}
917
918// Process all the sigchld which are pending
919// @return wether or not a given child exited with an error condition.
921{
922 bool hasError = false;
923 while (true) {
924 int status;
925 pid_t pid = waitpid((pid_t)(-1), &status, WNOHANG);
926 if (pid > 0) {
927 // Normal exit
928 int es = WEXITSTATUS(status);
929 if (WIFEXITED(status) == false || es != 0) {
930 // Look for the name associated to the pid in the infos
931 std::string id = "unknown";
932 assert(specs.size() == infos.size());
933 for (size_t ii = 0; ii < infos.size(); ++ii) {
934 if (infos[ii].pid == pid) {
935 id = specs[ii].id;
936 }
937 }
938 // No need to print anything if the user
939 // force quitted doing a double Ctrl-C.
940 if (double_sigint) {
941 } else if (forceful_exit) {
942 LOGP(error, "pid {} ({}) was forcefully terminated after being requested to quit", pid, id);
943 } else {
944 if (WIFSIGNALED(status)) {
945 int exitSignal = WTERMSIG(status);
946 es = exitSignal + 128;
947 LOGP(error, "Workflow crashed - PID {} ({}) was killed abnormally with {} and exited code was set to {}.", pid, id, strsignal(exitSignal), es);
948 } else {
949 es = 128;
950 LOGP(error, "Workflow crashed - PID {} ({}) did not exit correctly however it's not clear why. Exit code forced to {}.", pid, id, es);
951 }
952 }
953 hasError |= true;
954 }
955 for (auto& info : infos) {
956 if (info.pid == pid) {
957 info.active = false;
958 info.exitStatus = es;
959 }
960 }
961 continue;
962 } else {
963 break;
964 }
965 }
966 return hasError;
967}
968
969void doDPLException(RuntimeErrorRef& e, char const* processName)
970{
971 auto& err = o2::framework::error_from_ref(e);
972 if (err.maxBacktrace != 0) {
973 LOGP(fatal,
974 "Unhandled o2::framework::runtime_error reached the top of main of {}, device shutting down."
975 " Reason: {}",
976 processName, err.what);
977 LOGP(error, "Backtrace follow:");
978 BacktraceHelpers::demangled_backtrace_symbols(err.backtrace, err.maxBacktrace, STDERR_FILENO);
979 } else {
980 LOGP(fatal,
981 "Unhandled o2::framework::runtime_error reached the top of main of {}, device shutting down."
982 " Reason: {}",
983 processName, err.what);
984 LOGP(error, "Recompile with DPL_ENABLE_BACKTRACE=1 to get more information.");
985 }
986}
987
988void doUnknownException(std::string const& s, char const* processName)
989{
990 if (s.empty()) {
991 LOGP(fatal, "unknown error while setting up workflow in {}.", processName);
992 } else {
993 LOGP(fatal, "error while setting up workflow in {}: {}", processName, s);
994 }
995}
996
997[[maybe_unused]] AlgorithmSpec dryRun(DeviceSpec const& spec)
998{
1000 [&routes = spec.outputs](DataAllocator& outputs) {
1001 LOG(info) << "Dry run enforced. Creating dummy messages to simulate computation happended";
1002 for (auto& route : routes) {
1003 auto concrete = DataSpecUtils::asConcreteDataMatcher(route.matcher);
1004 outputs.make<int>(Output{concrete.origin, concrete.description, concrete.subSpec}, 2);
1005 }
1006 })};
1007}
1008
1010{
1011 // LOG(info) << "Process " << getpid() << " is exiting.";
1012}
1013
1014int doChild(int argc, char** argv, ServiceRegistry& serviceRegistry,
1015 RunningWorkflowInfo const& runningWorkflow,
1017 DriverConfig const& driverConfig,
1018 ProcessingPolicies processingPolicies,
1019 std::string const& defaultDriverClient,
1020 uv_loop_t* loop)
1021{
1022 fair::Logger::SetConsoleColor(false);
1023 fair::Logger::OnFatal([]() { throw runtime_error("Fatal error"); });
1024 DeviceSpec const& spec = runningWorkflow.devices[ref.index];
1025 LOG(info) << "Spawing new device " << spec.id << " in process with pid " << getpid();
1026
1027 fair::mq::DeviceRunner runner{argc, argv};
1028
1029 // Populate options from the command line. Notice that only the options
1030 // declared in the workflow definition are allowed.
1031 runner.AddHook<fair::mq::hooks::SetCustomCmdLineOptions>([&spec, driverConfig, defaultDriverClient](fair::mq::DeviceRunner& r) {
1032 std::string defaultExitTransitionTimeout = "0";
1033 std::string defaultDataProcessingTimeout = "0";
1034 std::string defaultInfologgerMode = "";
1036 if (deploymentMode == o2::framework::DeploymentMode::OnlineDDS) {
1037 defaultExitTransitionTimeout = "40";
1038 defaultDataProcessingTimeout = "20";
1039 defaultInfologgerMode = "infoLoggerD";
1040 } else if (deploymentMode == o2::framework::DeploymentMode::OnlineECS) {
1041 defaultExitTransitionTimeout = "40";
1042 defaultDataProcessingTimeout = "20";
1043 }
1044 boost::program_options::options_description optsDesc;
1046 char const* defaultSignposts = getenv("DPL_SIGNPOSTS");
1047 optsDesc.add_options()("monitoring-backend", bpo::value<std::string>()->default_value("default"), "monitoring backend info") //
1048 ("dpl-stats-min-online-publishing-interval", bpo::value<std::string>()->default_value("0"), "minimum flushing interval for online metrics (in s)") //
1049 ("driver-client-backend", bpo::value<std::string>()->default_value(defaultDriverClient), "backend for device -> driver communicataon: stdout://: use stdout, ws://: use websockets") //
1050 ("infologger-severity", bpo::value<std::string>()->default_value(""), "minimum FairLogger severity to send to InfoLogger") //
1051 ("dpl-tracing-flags", bpo::value<std::string>()->default_value(""), "pipe `|` separate list of events to be traced") //
1052 ("signposts", bpo::value<std::string>()->default_value(defaultSignposts ? defaultSignposts : ""), "comma separated list of signposts to enable") //
1053 ("expected-region-callbacks", bpo::value<std::string>()->default_value("0"), "how many region callbacks we are expecting") //
1054 ("exit-transition-timeout", bpo::value<std::string>()->default_value(defaultExitTransitionTimeout), "how many second to wait before switching from RUN to READY") //
1055 ("data-processing-timeout", bpo::value<std::string>()->default_value(defaultDataProcessingTimeout), "how many second to wait before stopping data processing and allowing data calibration") //
1056 ("timeframes-rate-limit", bpo::value<std::string>()->default_value("0"), "how many timeframe can be in fly at the same moment (0 disables)") //
1057 ("configuration,cfg", bpo::value<std::string>()->default_value("command-line"), "configuration backend") //
1058 ("infologger-mode", bpo::value<std::string>()->default_value(defaultInfologgerMode), "O2_INFOLOGGER_MODE override");
1059 r.fConfig.AddToCmdLineOptions(optsDesc, true);
1060 });
1061
1062 // This is to control lifetime. All these services get destroyed
1063 // when the runner is done.
1064 std::unique_ptr<SimpleRawDeviceService> simpleRawDeviceService;
1065 std::unique_ptr<DeviceState> deviceState;
1066 std::unique_ptr<ComputingQuotaEvaluator> quotaEvaluator;
1067 std::unique_ptr<FairMQDeviceProxy> deviceProxy;
1068 std::unique_ptr<DeviceContext> deviceContext;
1069
1070 auto afterConfigParsingCallback = [&simpleRawDeviceService,
1071 &runningWorkflow,
1072 ref,
1073 &spec,
1074 &quotaEvaluator,
1075 &serviceRegistry,
1076 &deviceState,
1077 &deviceProxy,
1078 &processingPolicies,
1079 &deviceContext,
1080 &driverConfig,
1081 &loop](fair::mq::DeviceRunner& r) {
1082 ServiceRegistryRef serviceRef = {serviceRegistry};
1083 simpleRawDeviceService = std::make_unique<SimpleRawDeviceService>(nullptr, spec);
1084 serviceRef.registerService(ServiceRegistryHelpers::handleForService<RawDeviceService>(simpleRawDeviceService.get()));
1085
1086 deviceState = std::make_unique<DeviceState>();
1087 deviceState->loop = loop;
1088 deviceState->tracingFlags = DeviceStateHelpers::parseTracingFlags(r.fConfig.GetPropertyAsString("dpl-tracing-flags"));
1089 serviceRef.registerService(ServiceRegistryHelpers::handleForService<DeviceState>(deviceState.get()));
1090
1091 quotaEvaluator = std::make_unique<ComputingQuotaEvaluator>(serviceRef);
1092 serviceRef.registerService(ServiceRegistryHelpers::handleForService<ComputingQuotaEvaluator>(quotaEvaluator.get()));
1093
1094 deviceContext = std::make_unique<DeviceContext>();
1095 serviceRef.registerService(ServiceRegistryHelpers::handleForService<DeviceSpec const>(&spec));
1096 serviceRef.registerService(ServiceRegistryHelpers::handleForService<RunningWorkflowInfo const>(&runningWorkflow));
1097 serviceRef.registerService(ServiceRegistryHelpers::handleForService<DeviceContext>(deviceContext.get()));
1098 serviceRef.registerService(ServiceRegistryHelpers::handleForService<DriverConfig const>(&driverConfig));
1099
1100 auto device = std::make_unique<DataProcessingDevice>(ref, serviceRegistry, processingPolicies);
1101
1102 serviceRef.get<RawDeviceService>().setDevice(device.get());
1103 r.fDevice = std::move(device);
1104 fair::Logger::SetConsoleColor(false);
1105
1107 for (auto& service : spec.services) {
1108 LOG(debug) << "Declaring service " << service.name;
1109 serviceRegistry.declareService(service, *deviceState.get(), r.fConfig);
1110 }
1111 if (ResourcesMonitoringHelper::isResourcesMonitoringEnabled(spec.resourceMonitoringInterval)) {
1112 serviceRef.get<Monitoring>().enableProcessMonitoring(spec.resourceMonitoringInterval, {PmMeasurement::Cpu, PmMeasurement::Mem, PmMeasurement::Smaps});
1113 }
1114 };
1115
1116 runner.AddHook<fair::mq::hooks::InstantiateDevice>(afterConfigParsingCallback);
1117
1118 auto result = runner.Run();
1119 ServiceRegistryRef serviceRef = {serviceRegistry};
1120 auto& context = serviceRef.get<DataProcessorContext>();
1121 DataProcessorContext::preExitCallbacks(context.preExitHandles, serviceRef);
1122 return result;
1123}
1124
1126 std::string executable;
1127 std::vector<std::string> args;
1128 std::vector<ConfigParamSpec> options;
1129};
1130
1131void gui_callback(uv_timer_s* ctx)
1132{
1133 auto* gui = reinterpret_cast<GuiCallbackContext*>(ctx->data);
1134 if (gui->plugin == nullptr) {
1135 // The gui is not there. Why are we here?
1136 O2_SIGNPOST_ID_FROM_POINTER(sid, driver, ctx->loop);
1137 O2_SIGNPOST_EVENT_EMIT_ERROR(driver, sid, "gui", "GUI timer callback invoked without a GUI plugin.");
1138 uv_timer_stop(ctx);
1139 return;
1140 }
1141 *gui->guiTimerExpired = true;
1142 static int counter = 0;
1143 if ((counter++ % 6000) == 0) {
1144 O2_SIGNPOST_ID_FROM_POINTER(sid, driver, ctx->loop);
1145 O2_SIGNPOST_EVENT_EMIT(driver, sid, "gui", "The GUI callback got called %d times.", counter);
1146 *gui->guiTimerExpired = false;
1147 }
1148 // One interval per GUI invocation, using the loop as anchor.
1149 O2_SIGNPOST_ID_FROM_POINTER(sid, gui, ctx->loop);
1150 O2_SIGNPOST_START(gui, sid, "gui", "gui_callback");
1151
1152 // New version which allows deferred closure of windows
1153 if (gui->plugin->supportsDeferredClose()) {
1154 // For now, there is nothing for which we want to defer the close
1155 // so if the flag is set, we simply exit
1156 if (*(gui->guiQuitRequested)) {
1157 O2_SIGNPOST_END(gui, sid, "gui", "Quit requested by the GUI.");
1158 return;
1159 }
1160 void* draw_data = nullptr;
1161 uint64_t frameStart = uv_hrtime();
1162 uint64_t frameLatency = frameStart - gui->frameLast;
1163
1164 // if less than 15ms have passed reuse old frame
1165 if (frameLatency / 1000000 <= 15) {
1166 draw_data = gui->lastFrame;
1167 O2_SIGNPOST_END(gui, sid, "gui", "Reusing old frame.");
1168 return;
1169 }
1170 // The result of the pollGUIPreRender is used to determine if we
1171 // should quit the GUI, however, the rendering is started in any
1172 // case, so we should complete it.
1173 if (!gui->plugin->pollGUIPreRender(gui->window, (float)frameLatency / 1000000000.0f)) {
1174 *(gui->guiQuitRequested) = true;
1175 }
1176 draw_data = gui->plugin->pollGUIRender(gui->callback);
1177 gui->plugin->pollGUIPostRender(gui->window, draw_data);
1178
1179 uint64_t frameEnd = uv_hrtime();
1180 *(gui->frameCost) = (frameEnd - frameStart) / 1000000.f;
1181 *(gui->frameLatency) = frameLatency / 1000000.f;
1182 gui->frameLast = frameStart;
1183 } else {
1184 void* draw_data = nullptr;
1185
1186 uint64_t frameStart = uv_hrtime();
1187 uint64_t frameLatency = frameStart - gui->frameLast;
1188
1189 // if less than 15ms have passed reuse old frame
1190 if (frameLatency / 1000000 > 15) {
1191 if (!gui->plugin->pollGUIPreRender(gui->window, (float)frameLatency / 1000000000.0f)) {
1192 *(gui->guiQuitRequested) = true;
1193 O2_SIGNPOST_END(gui, sid, "gui", "Reusing old frame.");
1194 return;
1195 }
1196 draw_data = gui->plugin->pollGUIRender(gui->callback);
1197 gui->plugin->pollGUIPostRender(gui->window, draw_data);
1198 } else {
1199 draw_data = gui->lastFrame;
1200 }
1201
1202 if (frameLatency / 1000000 > 15) {
1203 uint64_t frameEnd = uv_hrtime();
1204 *(gui->frameCost) = (frameEnd - frameStart) / 1000000.f;
1205 *(gui->frameLatency) = frameLatency / 1000000.f;
1206 gui->frameLast = frameStart;
1207 }
1208 }
1209 O2_SIGNPOST_END(gui, sid, "gui", "Gui redrawn.");
1210}
1211
1213void single_step_callback(uv_timer_s* ctx)
1214{
1215 auto* infos = reinterpret_cast<DeviceInfos*>(ctx->data);
1216 killChildren(*infos, SIGUSR1);
1217}
1218
1219void force_exit_callback(uv_timer_s* ctx)
1220{
1221 auto* infos = reinterpret_cast<DeviceInfos*>(ctx->data);
1222 killChildren(*infos, SIGKILL);
1223}
1224
1225std::vector<std::regex> getDumpableMetrics()
1226{
1227 auto performanceMetrics = o2::monitoring::ProcessMonitor::getAvailableMetricsNames();
1228 auto dumpableMetrics = std::vector<std::regex>{};
1229 for (const auto& metric : performanceMetrics) {
1230 dumpableMetrics.emplace_back(metric);
1231 }
1232 dumpableMetrics.emplace_back("^arrow-bytes-delta$");
1233 dumpableMetrics.emplace_back("^aod-bytes-read-uncompressed$");
1234 dumpableMetrics.emplace_back("^aod-bytes-read-compressed$");
1235 dumpableMetrics.emplace_back("^aod-file-read-info$");
1236 dumpableMetrics.emplace_back("^aod-largest-object-written$");
1237 dumpableMetrics.emplace_back("^table-bytes-.*");
1238 dumpableMetrics.emplace_back("^total-timeframes.*");
1239 dumpableMetrics.emplace_back("^device_state.*");
1240 dumpableMetrics.emplace_back("^total_wall_time_ms$");
1241 return dumpableMetrics;
1242}
1243
1245{
1246 auto* context = (DriverServerContext*)handle->data;
1247
1248 static auto performanceMetrics = getDumpableMetrics();
1250 context->driver->metrics, *(context->specs), performanceMetrics);
1251}
1252
1253void dumpRunSummary(DriverServerContext& context, DriverInfo const& driverInfo, DeviceInfos const& infos, DeviceSpecs const& specs)
1254{
1255 if (infos.empty()) {
1256 return;
1257 }
1258 LOGP(info, "## Processes completed. Run summary:");
1259 LOGP(info, "### Devices started: {}", infos.size());
1260 for (size_t di = 0; di < infos.size(); ++di) {
1261 auto& info = infos[di];
1262 auto& spec = specs[di];
1263 if (info.exitStatus) {
1264 LOGP(error, " - Device {}: pid {} (exit {})", spec.name, info.pid, info.exitStatus);
1265 } else {
1266 LOGP(info, " - Device {}: pid {} (exit {})", spec.name, info.pid, info.exitStatus);
1267 }
1268 if (info.exitStatus != 0 && info.firstSevereError.empty() == false) {
1269 LOGP(info, " - First error: {}", info.firstSevereError);
1270 }
1271 if (info.exitStatus != 0 && info.lastError != info.firstSevereError) {
1272 LOGP(info, " - Last error: {}", info.lastError);
1273 }
1274 }
1275 for (auto& summary : *context.summaryCallbacks) {
1276 summary(ServiceMetricsInfo{*context.metrics, *context.specs, *context.infos, context.driver->metrics, driverInfo});
1277 }
1278}
1279
1280auto bindGUIPort = [](DriverInfo& driverInfo, DriverServerContext& serverContext, std::string frameworkId) {
1281 uv_tcp_init(serverContext.loop, &serverContext.serverHandle);
1282
1283 driverInfo.port = 8080 + (getpid() % 30000);
1284
1285 if (getenv("DPL_REMOTE_GUI_PORT")) {
1286 try {
1287 driverInfo.port = stoi(std::string(getenv("DPL_REMOTE_GUI_PORT")));
1288 } catch (std::invalid_argument) {
1289 LOG(error) << "DPL_REMOTE_GUI_PORT not a valid integer";
1290 } catch (std::out_of_range) {
1291 LOG(error) << "DPL_REMOTE_GUI_PORT out of range (integer)";
1292 }
1293 if (driverInfo.port < 1024 || driverInfo.port > 65535) {
1294 LOG(error) << "DPL_REMOTE_GUI_PORT out of range (1024-65535)";
1295 }
1296 }
1297
1298 int result = 0;
1299 struct sockaddr_in* serverAddr = nullptr;
1300
1301 // Do not offer websocket endpoint for devices
1302 // FIXME: this was blocking david's workflows. For now
1303 // there is no point in any case to have devices
1304 // offering a web based API, but it might make sense in
1305 // the future to inspect them via some web based interface.
1306 if (serverContext.isDriver) {
1307 do {
1308 free(serverAddr);
1309 if (driverInfo.port > 64000) {
1310 throw runtime_error_f("Unable to find a free port for the driver. Last attempt returned %d", result);
1311 }
1312 serverAddr = (sockaddr_in*)malloc(sizeof(sockaddr_in));
1313 uv_ip4_addr("0.0.0.0", driverInfo.port, serverAddr);
1314 auto bindResult = uv_tcp_bind(&serverContext.serverHandle, (const struct sockaddr*)serverAddr, 0);
1315 if (bindResult != 0) {
1316 driverInfo.port++;
1317 usleep(1000);
1318 continue;
1319 }
1320 result = uv_listen((uv_stream_t*)&serverContext.serverHandle, 100, ws_connect_callback);
1321 if (result != 0) {
1322 driverInfo.port++;
1323 usleep(1000);
1324 continue;
1325 }
1326 } while (result != 0);
1327 } else if (getenv("DPL_DEVICE_REMOTE_GUI") && !serverContext.isDriver) {
1328 do {
1329 free(serverAddr);
1330 if (driverInfo.port > 64000) {
1331 throw runtime_error_f("Unable to find a free port for the driver. Last attempt returned %d", result);
1332 }
1333 serverAddr = (sockaddr_in*)malloc(sizeof(sockaddr_in));
1334 uv_ip4_addr("0.0.0.0", driverInfo.port, serverAddr);
1335 auto bindResult = uv_tcp_bind(&serverContext.serverHandle, (const struct sockaddr*)serverAddr, 0);
1336 if (bindResult != 0) {
1337 driverInfo.port++;
1338 usleep(1000);
1339 continue;
1340 }
1341 result = uv_listen((uv_stream_t*)&serverContext.serverHandle, 100, ws_connect_callback);
1342 if (result != 0) {
1343 driverInfo.port++;
1344 usleep(1000);
1345 continue;
1346 }
1347 LOG(info) << "Device GUI port: " << driverInfo.port << " " << frameworkId;
1348 } while (result != 0);
1349 }
1350};
1351
1352// This is the handler for the parent inner loop.
1354 WorkflowInfo const& workflowInfo,
1355 DataProcessorInfos const& previousDataProcessorInfos,
1356 CommandInfo const& commandInfo,
1357 DriverControl& driverControl,
1358 DriverInfo& driverInfo,
1359 DriverConfig& driverConfig,
1360 std::vector<DeviceMetricsInfo>& metricsInfos,
1361 std::vector<ConfigParamSpec> const& detectedParams,
1362 boost::program_options::variables_map& varmap,
1363 std::vector<ServiceSpec>& driverServices,
1364 std::string frameworkId)
1365{
1366 RunningWorkflowInfo runningWorkflow{
1367 .uniqueWorkflowId = driverInfo.uniqueWorkflowId,
1368 .shmSegmentId = (int16_t)atoi(varmap["shm-segment-id"].as<std::string>().c_str())};
1369 DeviceInfos infos;
1370 DeviceControls controls;
1371 DataProcessingStatesInfos allStates;
1372 auto* devicesManager = new DevicesManager{.controls = controls, .infos = infos, .specs = runningWorkflow.devices, .messages = {}};
1373 DeviceExecutions deviceExecutions;
1374 DataProcessorInfos dataProcessorInfos = previousDataProcessorInfos;
1375
1376 std::vector<uv_poll_t*> pollHandles;
1377 std::vector<DeviceStdioContext> childFds;
1378
1379 std::vector<ComputingResource> resources;
1380
1381 if (driverInfo.resources != "") {
1382 resources = ComputingResourceHelpers::parseResources(driverInfo.resources);
1383 } else {
1385 }
1386
1387 auto resourceManager = std::make_unique<SimpleResourceManager>(resources);
1388
1389 DebugGUI* debugGUI = nullptr;
1390 void* window = nullptr;
1391 decltype(debugGUI->getGUIDebugger(infos, runningWorkflow.devices, allStates, dataProcessorInfos, metricsInfos, driverInfo, controls, driverControl)) debugGUICallback;
1392
1393 // An empty frameworkId means this is the driver, so we initialise the GUI
1394 auto initDebugGUI = []() -> DebugGUI* {
1395 uv_lib_t supportLib;
1396 int result = 0;
1397#ifdef __APPLE__
1398 result = uv_dlopen("libO2FrameworkGUISupport.dylib", &supportLib);
1399#else
1400 result = uv_dlopen("libO2FrameworkGUISupport.so", &supportLib);
1401#endif
1402 if (result == -1) {
1403 LOG(error) << uv_dlerror(&supportLib);
1404 return nullptr;
1405 }
1406 DPLPluginHandle* (*dpl_plugin_callback)(DPLPluginHandle*);
1407
1408 result = uv_dlsym(&supportLib, "dpl_plugin_callback", (void**)&dpl_plugin_callback);
1409 if (result == -1) {
1410 LOG(error) << uv_dlerror(&supportLib);
1411 return nullptr;
1412 }
1413 DPLPluginHandle* pluginInstance = dpl_plugin_callback(nullptr);
1414 return PluginManager::getByName<DebugGUI>(pluginInstance, "ImGUIDebugGUI");
1415 };
1416
1417 // We initialise this in the driver, because different drivers might have
1418 // different versions of the service
1419 ServiceRegistry serviceRegistry;
1420
1421 if ((driverConfig.batch == false || getenv("DPL_DRIVER_REMOTE_GUI") != nullptr) && frameworkId.empty()) {
1422 debugGUI = initDebugGUI();
1423 if (debugGUI) {
1424 if (driverConfig.batch == false) {
1425 window = debugGUI->initGUI("O2 Framework debug GUI", serviceRegistry);
1426 } else {
1427 window = debugGUI->initGUI(nullptr, serviceRegistry);
1428 }
1429 }
1430 } else if (getenv("DPL_DEVICE_REMOTE_GUI") && !frameworkId.empty()) {
1431 debugGUI = initDebugGUI();
1432 // We never run the GUI on desktop for devices. All
1433 // you can do is to connect to the remote version.
1434 // this is done to avoid having a proliferation of
1435 // GUIs popping up when the variable is set globally.
1436 // FIXME: maybe this is not what we want, but it should
1437 // be ok for now.
1438 if (debugGUI) {
1439 window = debugGUI->initGUI(nullptr, serviceRegistry);
1440 }
1441 }
1442 if (driverConfig.batch == false && window == nullptr && frameworkId.empty()) {
1443 LOG(warn) << "Could not create GUI. Switching to batch mode. Do you have GLFW on your system?";
1444 driverConfig.batch = true;
1445 if (varmap["error-policy"].defaulted()) {
1446 driverInfo.processingPolicies.error = TerminationPolicy::QUIT;
1447 }
1448 }
1449 bool guiQuitRequested = false;
1450 bool hasError = false;
1451
1452 // FIXME: I should really have some way of exiting the
1453 // parent..
1454 DriverState current;
1455 DriverState previous;
1456
1457 uv_loop_t* loop = uv_loop_new();
1458
1459 uv_timer_t* gui_timer = nullptr;
1460
1461 if (!driverConfig.batch) {
1462 gui_timer = (uv_timer_t*)malloc(sizeof(uv_timer_t));
1463 uv_timer_init(loop, gui_timer);
1464 }
1465
1466 std::vector<ServiceMetricHandling> metricProcessingCallbacks;
1467 std::vector<ServiceSummaryHandling> summaryCallbacks;
1468 std::vector<ServicePreSchedule> preScheduleCallbacks;
1469 std::vector<ServicePostSchedule> postScheduleCallbacks;
1470 std::vector<ServiceDriverInit> driverInitCallbacks;
1471 for (auto& service : driverServices) {
1472 if (service.driverStartup == nullptr) {
1473 continue;
1474 }
1475 service.driverStartup(serviceRegistry, DeviceConfig{varmap});
1476 }
1477
1478 ServiceRegistryRef ref{serviceRegistry};
1479 ref.registerService(ServiceRegistryHelpers::handleForService<DevicesManager>(devicesManager));
1480
1481 bool guiTimerExpired = false;
1482 GuiCallbackContext guiContext;
1483 guiContext.plugin = debugGUI;
1484 guiContext.frameLast = uv_hrtime();
1485 guiContext.frameLatency = &driverInfo.frameLatency;
1486 guiContext.frameCost = &driverInfo.frameCost;
1487 guiContext.guiQuitRequested = &guiQuitRequested;
1488 guiContext.guiTimerExpired = &guiTimerExpired;
1489
1490 // This is to make sure we can process metrics, commands, configuration
1491 // changes coming from websocket (or even via any standard uv_stream_t, I guess).
1492 DriverServerContext serverContext{
1493 .registry = {serviceRegistry},
1494 .loop = loop,
1495 .controls = &controls,
1496 .infos = &infos,
1497 .states = &allStates,
1498 .specs = &runningWorkflow.devices,
1499 .metrics = &metricsInfos,
1500 .metricProcessingCallbacks = &metricProcessingCallbacks,
1501 .summaryCallbacks = &summaryCallbacks,
1502 .driver = &driverInfo,
1503 .gui = &guiContext,
1504 .isDriver = frameworkId.empty(),
1505 };
1506
1507 serverContext.serverHandle.data = &serverContext;
1508
1509 uv_timer_t force_step_timer;
1510 uv_timer_init(loop, &force_step_timer);
1511 uv_timer_t force_exit_timer;
1512 uv_timer_init(loop, &force_exit_timer);
1513
1514 bool guiDeployedOnce = false;
1515 bool once = false;
1516
1517 uv_timer_t metricDumpTimer;
1518 metricDumpTimer.data = &serverContext;
1519 bool allChildrenGone = false;
1520 guiContext.allChildrenGone = &allChildrenGone;
1521 O2_SIGNPOST_ID_FROM_POINTER(sid, driver, loop);
1522 O2_SIGNPOST_START(driver, sid, "driver", "Starting driver loop");
1523
1524 // Async callback to process the output of the children, if needed.
1525 serverContext.asyncLogProcessing = (uv_async_t*)malloc(sizeof(uv_async_t));
1526 serverContext.asyncLogProcessing->data = &serverContext;
1527 uv_async_init(loop, serverContext.asyncLogProcessing, [](uv_async_t* handle) {
1528 auto* context = (DriverServerContext*)handle->data;
1529 processChildrenOutput(context->loop, *context->driver, *context->infos, *context->specs, *context->controls);
1530 });
1531
1532 while (true) {
1533 // If control forced some transition on us, we push it to the queue.
1534 if (driverControl.forcedTransitions.empty() == false) {
1535 for (auto transition : driverControl.forcedTransitions) {
1536 driverInfo.states.push_back(transition);
1537 }
1538 driverControl.forcedTransitions.resize(0);
1539 }
1540 // In case a timeout was requested, we check if we are running
1541 // for more than the timeout duration and exit in case that's the case.
1542 {
1543 auto currentTime = uv_hrtime();
1544 uint64_t diff = (currentTime - driverInfo.startTime) / 1000000000LL;
1545 if ((graceful_exit == false) && (driverInfo.timeout > 0) && (diff > driverInfo.timeout)) {
1546 LOG(info) << "Timout ellapsed. Requesting to quit.";
1547 graceful_exit = true;
1548 }
1549 }
1550 // Move to exit loop if sigint was sent we execute this only once.
1551 if (graceful_exit == true && driverInfo.sigintRequested == false) {
1552 driverInfo.sigintRequested = true;
1553 driverInfo.states.resize(0);
1554 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
1555 }
1556 // If one of the children dies and sigint was not requested
1557 // we should decide what to do.
1558 if (sigchld_requested == true && driverInfo.sigchldRequested == false) {
1559 driverInfo.sigchldRequested = true;
1560 driverInfo.states.push_back(DriverState::HANDLE_CHILDREN);
1561 }
1562 if (driverInfo.states.empty() == false) {
1563 previous = current;
1564 current = driverInfo.states.back();
1565 } else {
1566 current = DriverState::UNKNOWN;
1567 }
1568 driverInfo.states.pop_back();
1569 switch (current) {
1570 case DriverState::BIND_GUI_PORT:
1571 bindGUIPort(driverInfo, serverContext, frameworkId);
1572 break;
1573 case DriverState::INIT:
1574 LOGP(info, "Initialising O2 Data Processing Layer. Driver PID: {}.", getpid());
1575 LOGP(info, "Driver listening on port: {}", driverInfo.port);
1576
1577 // Install signal handler for quitting children.
1578 driverInfo.sa_handle_child.sa_handler = &handle_sigchld;
1579 sigemptyset(&driverInfo.sa_handle_child.sa_mask);
1580 driverInfo.sa_handle_child.sa_flags = SA_RESTART | SA_NOCLDSTOP;
1581 if (sigaction(SIGCHLD, &driverInfo.sa_handle_child, nullptr) == -1) {
1582 perror(nullptr);
1583 exit(1);
1584 }
1585
1588 if (driverInfo.noSHMCleanup) {
1589 LOGP(warning, "Not cleaning up shared memory.");
1590 } else {
1591 cleanupSHM(driverInfo.uniqueWorkflowId);
1592 }
1597 for (auto& callback : driverInitCallbacks) {
1598 callback(serviceRegistry, {varmap});
1599 }
1600 driverInfo.states.push_back(DriverState::RUNNING);
1601 // driverInfo.states.push_back(DriverState::REDEPLOY_GUI);
1602 LOG(info) << "O2 Data Processing Layer initialised. We brake for nobody.";
1603#ifdef NDEBUG
1604 LOGF(info, "Optimised build. O2DEBUG / LOG(debug) / LOGF(debug) / assert statement will not be shown.");
1605#endif
1606 break;
1607 case DriverState::IMPORT_CURRENT_WORKFLOW:
1608 // This state is needed to fill the metadata structure
1609 // which contains how to run the current workflow
1610 dataProcessorInfos = previousDataProcessorInfos;
1611 for (auto const& device : runningWorkflow.devices) {
1612 auto exists = std::find_if(dataProcessorInfos.begin(),
1613 dataProcessorInfos.end(),
1614 [id = device.id](DataProcessorInfo const& info) -> bool { return info.name == id; });
1615 if (exists != dataProcessorInfos.end()) {
1616 continue;
1617 }
1618 std::vector<std::string> channels;
1619 for (auto channel : device.inputChannels) {
1620 channels.push_back(channel.name);
1621 }
1622 for (auto channel : device.outputChannels) {
1623 channels.push_back(channel.name);
1624 }
1625 dataProcessorInfos.push_back(
1627 device.id,
1628 workflowInfo.executable,
1629 workflowInfo.args,
1630 workflowInfo.options,
1631 channels});
1632 }
1633 break;
1634 case DriverState::MATERIALISE_WORKFLOW:
1635 try {
1636 auto workflowState = WorkflowHelpers::verifyWorkflow(workflow);
1637 if (driverConfig.batch == true && varmap["dds"].as<std::string>().empty() && !varmap["dump-workflow"].as<bool>() && workflowState == WorkflowParsingState::Empty) {
1638 LOGP(error, "Empty workflow provided while running in batch mode.");
1639 return 1;
1640 }
1641
1644 auto altered_workflow = workflow;
1645
1646 auto confNameFromParam = [](std::string const& paramName) {
1647 std::regex name_regex(R"(^control:([\w-]+)\/(\w+))");
1648 auto match = std::sregex_token_iterator(paramName.begin(), paramName.end(), name_regex, 0);
1649 if (match == std::sregex_token_iterator()) {
1650 throw runtime_error_f("Malformed process control spec: %s", paramName.c_str());
1651 }
1652 std::string task = std::sregex_token_iterator(paramName.begin(), paramName.end(), name_regex, 1)->str();
1653 std::string conf = std::sregex_token_iterator(paramName.begin(), paramName.end(), name_regex, 2)->str();
1654 return std::pair{task, conf};
1655 };
1656 bool altered = false;
1657 for (auto& device : altered_workflow) {
1658 // ignore internal devices
1659 if (device.name.find("internal") != std::string::npos) {
1660 continue;
1661 }
1662 // ignore devices with no inputs
1663 if (device.inputs.empty() == true) {
1664 continue;
1665 }
1666 // ignore devices with no metadata in inputs
1667 auto hasMetadata = std::any_of(device.inputs.begin(), device.inputs.end(), [](InputSpec const& spec) {
1668 return spec.metadata.empty() == false;
1669 });
1670 if (!hasMetadata) {
1671 continue;
1672 }
1673 // ignore devices with no control options
1674 auto hasControls = std::any_of(device.inputs.begin(), device.inputs.end(), [](InputSpec const& spec) {
1675 return std::any_of(spec.metadata.begin(), spec.metadata.end(), [](ConfigParamSpec const& param) {
1676 return param.type == VariantType::Bool && param.name.find("control:") != std::string::npos;
1677 });
1678 });
1679 if (!hasControls) {
1680 continue;
1681 }
1682
1683 LOGP(debug, "Adjusting device {}", device.name.c_str());
1684
1685 auto configStore = DeviceConfigurationHelpers::getConfiguration(serviceRegistry, device.name.c_str(), device.options);
1686 if (configStore != nullptr) {
1687 auto reg = std::make_unique<ConfigParamRegistry>(std::move(configStore));
1688 for (auto& input : device.inputs) {
1689 for (auto& param : input.metadata) {
1690 if (param.type == VariantType::Bool && param.name.find("control:") != std::string::npos) {
1691 if (param.name != "control:default" && param.name != "control:spawn" && param.name != "control:build" && param.name != "control:define") {
1692 auto confName = confNameFromParam(param.name).second;
1693 param.defaultValue = reg->get<bool>(confName.c_str());
1694 }
1695 }
1696 }
1697 }
1698 }
1700 LOGP(debug, "Original inputs: ");
1701 for (auto& input : device.inputs) {
1702 LOGP(debug, "-> {}", input.binding);
1703 }
1704 auto end = device.inputs.end();
1705 auto new_end = std::remove_if(device.inputs.begin(), device.inputs.end(), [](InputSpec& input) {
1706 auto requested = false;
1707 auto hasControls = false;
1708 for (auto& param : input.metadata) {
1709 if (param.type != VariantType::Bool) {
1710 continue;
1711 }
1712 if (param.name.find("control:") != std::string::npos) {
1713 hasControls = true;
1714 if (param.defaultValue.get<bool>() == true) {
1715 requested = true;
1716 break;
1717 }
1718 }
1719 }
1720 if (hasControls) {
1721 return !requested;
1722 }
1723 return false;
1724 });
1725 device.inputs.erase(new_end, end);
1726 LOGP(debug, "Adjusted inputs: ");
1727 for (auto& input : device.inputs) {
1728 LOGP(debug, "-> {}", input.binding);
1729 }
1730 altered = true;
1731 }
1732 WorkflowHelpers::adjustTopology(altered_workflow, *driverInfo.configContext);
1733 if (altered) {
1734 WorkflowSpecNode node{altered_workflow};
1735 for (auto& service : driverServices) {
1736 if (service.adjustTopology == nullptr) {
1737 continue;
1738 }
1739 service.adjustTopology(node, *driverInfo.configContext);
1740 }
1741 }
1742
1743 // These allow services customization via an environment variable
1744 OverrideServiceSpecs overrides = ServiceSpecHelpers::parseOverrides(getenv("DPL_OVERRIDE_SERVICES"));
1745 DeviceSpecHelpers::validate(altered_workflow);
1747 driverInfo.channelPolicies,
1748 driverInfo.completionPolicies,
1749 driverInfo.dispatchPolicies,
1750 driverInfo.resourcePolicies,
1751 driverInfo.callbacksPolicies,
1752 driverInfo.sendingPolicies,
1753 driverInfo.forwardingPolicies,
1754 runningWorkflow.devices,
1755 *resourceManager,
1756 driverInfo.uniqueWorkflowId,
1757 *driverInfo.configContext,
1758 !varmap["no-IPC"].as<bool>(),
1759 driverInfo.resourcesMonitoringInterval,
1760 varmap["channel-prefix"].as<std::string>(),
1761 overrides);
1762 metricProcessingCallbacks.clear();
1763 std::vector<std::string> matchingServices;
1764
1765 // FIXME: once moving to C++20, we can use templated lambdas.
1766 matchingServices.clear();
1767 for (auto& device : runningWorkflow.devices) {
1768 for (auto& service : device.services) {
1769 // If a service with the same name is already registered, skip it
1770 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1771 continue;
1772 }
1773 if (service.metricHandling) {
1774 metricProcessingCallbacks.push_back(service.metricHandling);
1775 matchingServices.push_back(service.name);
1776 }
1777 }
1778 }
1779
1780 // FIXME: once moving to C++20, we can use templated lambdas.
1781 matchingServices.clear();
1782 for (auto& device : runningWorkflow.devices) {
1783 for (auto& service : device.services) {
1784 // If a service with the same name is already registered, skip it
1785 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1786 continue;
1787 }
1788 if (service.summaryHandling) {
1789 summaryCallbacks.push_back(service.summaryHandling);
1790 matchingServices.push_back(service.name);
1791 }
1792 }
1793 }
1794
1795 preScheduleCallbacks.clear();
1796 matchingServices.clear();
1797 for (auto& device : runningWorkflow.devices) {
1798 for (auto& service : device.services) {
1799 // If a service with the same name is already registered, skip it
1800 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1801 continue;
1802 }
1803 if (service.preSchedule) {
1804 preScheduleCallbacks.push_back(service.preSchedule);
1805 }
1806 }
1807 }
1808 postScheduleCallbacks.clear();
1809 matchingServices.clear();
1810 for (auto& device : runningWorkflow.devices) {
1811 for (auto& service : device.services) {
1812 // If a service with the same name is already registered, skip it
1813 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1814 continue;
1815 }
1816 if (service.postSchedule) {
1817 postScheduleCallbacks.push_back(service.postSchedule);
1818 }
1819 }
1820 }
1821 driverInitCallbacks.clear();
1822 matchingServices.clear();
1823 for (auto& device : runningWorkflow.devices) {
1824 for (auto& service : device.services) {
1825 // If a service with the same name is already registered, skip it
1826 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1827 continue;
1828 }
1829 if (service.driverInit) {
1830 driverInitCallbacks.push_back(service.driverInit);
1831 }
1832 }
1833 }
1834
1835 // This should expand nodes so that we can build a consistent DAG.
1836
1837 // This updates the options in the runningWorkflow.devices
1838 for (auto& device : runningWorkflow.devices) {
1839 // ignore internal devices
1840 if (device.name.find("internal") != std::string::npos) {
1841 continue;
1842 }
1843 auto configStore = DeviceConfigurationHelpers::getConfiguration(serviceRegistry, device.name.c_str(), device.options);
1844 if (configStore != nullptr) {
1845 auto reg = std::make_unique<ConfigParamRegistry>(std::move(configStore));
1846 for (auto& option : device.options) {
1847 const char* name = option.name.c_str();
1848 switch (option.type) {
1849 case VariantType::Int:
1850 option.defaultValue = reg->get<int32_t>(name);
1851 break;
1852 case VariantType::Int8:
1853 option.defaultValue = reg->get<int8_t>(name);
1854 break;
1855 case VariantType::Int16:
1856 option.defaultValue = reg->get<int16_t>(name);
1857 break;
1858 case VariantType::UInt8:
1859 option.defaultValue = reg->get<uint8_t>(name);
1860 break;
1861 case VariantType::UInt16:
1862 option.defaultValue = reg->get<uint16_t>(name);
1863 break;
1864 case VariantType::UInt32:
1865 option.defaultValue = reg->get<uint32_t>(name);
1866 break;
1867 case VariantType::UInt64:
1868 option.defaultValue = reg->get<uint64_t>(name);
1869 break;
1870 case VariantType::Int64:
1871 option.defaultValue = reg->get<int64_t>(name);
1872 break;
1873 case VariantType::Float:
1874 option.defaultValue = reg->get<float>(name);
1875 break;
1876 case VariantType::Double:
1877 option.defaultValue = reg->get<double>(name);
1878 break;
1879 case VariantType::String:
1880 option.defaultValue = reg->get<std::string>(name);
1881 break;
1882 case VariantType::Bool:
1883 option.defaultValue = reg->get<bool>(name);
1884 break;
1885 case VariantType::ArrayInt:
1886 option.defaultValue = reg->get<std::vector<int>>(name);
1887 break;
1888 case VariantType::ArrayFloat:
1889 option.defaultValue = reg->get<std::vector<float>>(name);
1890 break;
1891 case VariantType::ArrayDouble:
1892 option.defaultValue = reg->get<std::vector<double>>(name);
1893 break;
1894 case VariantType::ArrayString:
1895 option.defaultValue = reg->get<std::vector<std::string>>(name);
1896 break;
1897 case VariantType::Array2DInt:
1898 option.defaultValue = reg->get<Array2D<int>>(name);
1899 break;
1900 case VariantType::Array2DFloat:
1901 option.defaultValue = reg->get<Array2D<float>>(name);
1902 break;
1903 case VariantType::Array2DDouble:
1904 option.defaultValue = reg->get<Array2D<double>>(name);
1905 break;
1906 case VariantType::LabeledArrayInt:
1907 option.defaultValue = reg->get<LabeledArray<int>>(name);
1908 break;
1909 case VariantType::LabeledArrayFloat:
1910 option.defaultValue = reg->get<LabeledArray<float>>(name);
1911 break;
1912 case VariantType::LabeledArrayDouble:
1913 option.defaultValue = reg->get<LabeledArray<double>>(name);
1914 break;
1915 case VariantType::LabeledArrayString:
1916 option.defaultValue = reg->get<LabeledArray<std::string>>(name);
1917 break;
1918 default:
1919 break;
1920 }
1921 }
1922 }
1923 }
1924 } catch (std::runtime_error& e) {
1925 LOGP(error, "invalid workflow in {}: {}", driverInfo.argv[0], e.what());
1926 return 1;
1929#ifdef DPL_ENABLE_BACKTRACE
1930 BacktraceHelpers::demangled_backtrace_symbols(err.backtrace, err.maxBacktrace, STDERR_FILENO);
1931#endif
1932 LOGP(error, "invalid workflow in {}: {}", driverInfo.argv[0], err.what);
1933 return 1;
1934 } catch (...) {
1935 LOGP(error, "invalid workflow in {}: Unknown error while materialising workflow", driverInfo.argv[0]);
1936 return 1;
1937 }
1938 break;
1939 case DriverState::DO_CHILD:
1940 // We do not start the process if by default we are stopped.
1941 if (driverControl.defaultStopped) {
1942 kill(getpid(), SIGSTOP);
1943 }
1944 for (size_t di = 0; di < runningWorkflow.devices.size(); di++) {
1946 if (runningWorkflow.devices[di].id == frameworkId) {
1947 return doChild(driverInfo.argc, driverInfo.argv,
1948 serviceRegistry,
1949 runningWorkflow, ref,
1950 driverConfig,
1951 driverInfo.processingPolicies,
1952 driverInfo.defaultDriverClient,
1953 loop);
1954 }
1955 }
1956 {
1957 std::ostringstream ss;
1958 for (auto& processor : workflow) {
1959 ss << " - " << processor.name << "\n";
1960 }
1961 for (auto& spec : runningWorkflow.devices) {
1962 ss << " - " << spec.name << "(" << spec.id << ")"
1963 << "\n";
1964 }
1965 driverInfo.lastError = fmt::format(
1966 "Unable to find component with id {}."
1967 " Available options:\n{}",
1968 frameworkId, ss.str());
1969 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
1970 }
1971 break;
1972 case DriverState::REDEPLOY_GUI:
1973 // The callback for the GUI needs to be recalculated every time
1974 // the deployed configuration changes, e.g. a new device
1975 // has been added to the topology.
1976 // We need to recreate the GUI callback every time we reschedule
1977 // because getGUIDebugger actually recreates the GUI state.
1978 // Notice also that we need the actual gui_timer only for the
1979 // case the GUI runs in interactive mode, however we deploy the
1980 // GUI in both interactive and non-interactive mode, if the
1981 // DPL_DRIVER_REMOTE_GUI environment variable is set.
1982 if (!driverConfig.batch || getenv("DPL_DRIVER_REMOTE_GUI")) {
1983 if (gui_timer) {
1984 uv_timer_stop(gui_timer);
1985 }
1986
1987 auto callback = debugGUI->getGUIDebugger(infos, runningWorkflow.devices, allStates, dataProcessorInfos, metricsInfos, driverInfo, controls, driverControl);
1988 guiContext.callback = [&serviceRegistry, &driverServices, &debugGUI, &infos, &runningWorkflow, &dataProcessorInfos, &metricsInfos, &driverInfo, &controls, &driverControl, callback]() {
1989 callback();
1990 for (auto& service : driverServices) {
1991 if (service.postRenderGUI) {
1992 service.postRenderGUI(serviceRegistry);
1993 }
1994 }
1995 };
1996 guiContext.window = window;
1997
1998 if (gui_timer) {
1999 gui_timer->data = &guiContext;
2000 uv_timer_start(gui_timer, gui_callback, 0, 20);
2001 }
2002 guiDeployedOnce = true;
2003 }
2004 break;
2005 case DriverState::MERGE_CONFIGS: {
2006 try {
2007 controls.resize(runningWorkflow.devices.size());
2010 if (varmap.count("dpl-tracing-flags")) {
2011 for (auto& control : controls) {
2012 auto tracingFlags = DeviceStateHelpers::parseTracingFlags(varmap["dpl-tracing-flags"].as<std::string>());
2013 control.tracingFlags = tracingFlags;
2014 }
2015 }
2016 deviceExecutions.resize(runningWorkflow.devices.size());
2017
2018 // Options which should be uniform across all
2019 // the subworkflow invokations.
2020 const auto uniformOptions = {
2021 "--aod-file",
2022 "--aod-memory-rate-limit",
2023 "--aod-writer-json",
2024 "--aod-writer-ntfmerge",
2025 "--aod-writer-resdir",
2026 "--aod-writer-resfile",
2027 "--aod-writer-resmode",
2028 "--aod-writer-maxfilesize",
2029 "--aod-writer-keep",
2030 "--aod-max-io-rate",
2031 "--aod-parent-access-level",
2032 "--aod-parent-base-path-replacement",
2033 "--driver-client-backend",
2034 "--fairmq-ipc-prefix",
2035 "--readers",
2036 "--resources-monitoring",
2037 "--resources-monitoring-dump-interval",
2038 "--time-limit",
2039 };
2040
2041 for (auto& option : uniformOptions) {
2042 DeviceSpecHelpers::reworkHomogeneousOption(dataProcessorInfos, option, nullptr);
2043 }
2044
2045 DeviceSpecHelpers::reworkShmSegmentSize(dataProcessorInfos);
2047 driverControl.defaultStopped,
2048 driverInfo.processingPolicies.termination == TerminationPolicy::WAIT,
2049 driverInfo.port,
2050 driverConfig,
2051 dataProcessorInfos,
2052 runningWorkflow.devices,
2053 deviceExecutions,
2054 controls,
2055 detectedParams,
2056 driverInfo.uniqueWorkflowId);
2059 LOGP(error, "unable to merge configurations in {}: {}", driverInfo.argv[0], err.what);
2060#ifdef DPL_ENABLE_BACKTRACE
2061 std::cerr << "\nStacktrace follows:\n\n";
2062 BacktraceHelpers::demangled_backtrace_symbols(err.backtrace, err.maxBacktrace, STDERR_FILENO);
2063#endif
2064 return 1;
2065 }
2066 } break;
2067 case DriverState::SCHEDULE: {
2068 // FIXME: for the moment modifying the topology means we rebuild completely
2069 // all the devices and we restart them. This is also what DDS does at
2070 // a larger scale. In principle one could try to do a delta and only
2071 // restart the data processors which need to be restarted.
2072 LOG(info) << "Redeployment of configuration asked.";
2073 std::ostringstream forwardedStdin;
2074 WorkflowSerializationHelpers::dump(forwardedStdin, workflow, dataProcessorInfos, commandInfo);
2075 infos.reserve(runningWorkflow.devices.size());
2076
2077 // This is guaranteed to be a single CPU.
2078 unsigned parentCPU = -1;
2079 unsigned parentNode = -1;
2080#if defined(__linux__) && __has_include(<sched.h>)
2081 parentCPU = sched_getcpu();
2082#elif __has_include(<linux/getcpu.h>)
2083 getcpu(&parentCPU, &parentNode, nullptr);
2084#elif __has_include(<cpuid.h>) && (__x86_64__ || __i386__)
2085 // FIXME: this is a last resort as it is apparently buggy
2086 // on some Intel CPUs.
2087 GETCPU(parentCPU);
2088#endif
2089 for (auto& callback : preScheduleCallbacks) {
2090 callback(serviceRegistry, {varmap});
2091 }
2092 childFds.resize(runningWorkflow.devices.size());
2093 for (int di = 0; di < (int)runningWorkflow.devices.size(); ++di) {
2094 auto& context = childFds[di];
2095 createPipes(context.childstdin);
2096 createPipes(context.childstdout);
2097 if (driverInfo.mode == DriverMode::EMBEDDED || runningWorkflow.devices[di].resource.hostname != driverInfo.deployHostname) {
2098 spawnRemoteDevice(loop, forwardedStdin.str(),
2099 runningWorkflow.devices[di], controls[di], deviceExecutions[di], infos, allStates);
2100 } else {
2101 DeviceRef ref{di};
2102 spawnDevice(loop,
2103 ref,
2104 runningWorkflow.devices, driverInfo,
2105 controls, deviceExecutions, infos,
2106 allStates,
2107 serviceRegistry, varmap,
2108 childFds, parentCPU, parentNode);
2109 }
2110 }
2111 handleSignals();
2112 handleChildrenStdio(&serverContext, forwardedStdin.str(), childFds, pollHandles);
2113 for (auto& callback : postScheduleCallbacks) {
2114 callback(serviceRegistry, {varmap});
2115 }
2116 assert(infos.empty() == false);
2117
2118 // In case resource monitoring is requested, we dump metrics to disk
2119 // every 3 minutes.
2120 if (driverInfo.resourcesMonitoringDumpInterval && ResourcesMonitoringHelper::isResourcesMonitoringEnabled(driverInfo.resourcesMonitoringInterval)) {
2121 uv_timer_init(loop, &metricDumpTimer);
2122 uv_timer_start(&metricDumpTimer, dumpMetricsCallback,
2123 driverInfo.resourcesMonitoringDumpInterval * 1000,
2124 driverInfo.resourcesMonitoringDumpInterval * 1000);
2125 }
2127 for (const auto& processorInfo : dataProcessorInfos) {
2128 const auto& cmdLineArgs = processorInfo.cmdLineArgs;
2129 if (std::find(cmdLineArgs.begin(), cmdLineArgs.end(), "--severity") != cmdLineArgs.end()) {
2130 for (size_t counter = 0; const auto& spec : runningWorkflow.devices) {
2131 if (spec.name.compare(processorInfo.name) == 0) {
2132 auto& info = infos[counter];
2133 const auto logLevelIt = std::find(cmdLineArgs.begin(), cmdLineArgs.end(), "--severity") + 1;
2134 if ((*logLevelIt).compare("debug") == 0) {
2135 info.logLevel = LogParsingHelpers::LogLevel::Debug;
2136 } else if ((*logLevelIt).compare("detail") == 0) {
2137 info.logLevel = LogParsingHelpers::LogLevel::Debug;
2138 } else if ((*logLevelIt).compare("info") == 0) {
2139 info.logLevel = LogParsingHelpers::LogLevel::Info;
2140 } else if ((*logLevelIt).compare("warning") == 0) {
2141 info.logLevel = LogParsingHelpers::LogLevel::Warning;
2142 } else if ((*logLevelIt).compare("error") == 0) {
2143 info.logLevel = LogParsingHelpers::LogLevel::Error;
2144 } else if ((*logLevelIt).compare("important") == 0) {
2145 info.logLevel = LogParsingHelpers::LogLevel::Info;
2146 } else if ((*logLevelIt).compare("alarm") == 0) {
2147 info.logLevel = LogParsingHelpers::LogLevel::Alarm;
2148 } else if ((*logLevelIt).compare("critical") == 0) {
2149 info.logLevel = LogParsingHelpers::LogLevel::Critical;
2150 } else if ((*logLevelIt).compare("fatal") == 0) {
2151 info.logLevel = LogParsingHelpers::LogLevel::Fatal;
2152 }
2153 break;
2154 }
2155 ++counter;
2156 }
2157 }
2158 }
2159 LOG(info) << "Redeployment of configuration done.";
2160 } break;
2161 case DriverState::RUNNING:
2162 // Run any pending libUV event loop, block if
2163 // any, so that we do not consume CPU time when the driver is
2164 // idle.
2165 devicesManager->flush();
2166 // We print the event loop for the gui only once every
2167 // 6000 iterations (i.e. ~2 minutes). To avoid spamming, while still
2168 // being able to see the event loop in case of a deadlock / systematic failure.
2169 if (guiTimerExpired == false) {
2170 O2_SIGNPOST_EVENT_EMIT(driver, sid, "mainloop", "Entering event loop with %{public}s", once ? "UV_RUN_ONCE" : "UV_RUN_NOWAIT");
2171 }
2172 uv_run(loop, once ? UV_RUN_ONCE : UV_RUN_NOWAIT);
2173 once = true;
2174 // Calculate what we should do next and eventually
2175 // show the GUI
2176 if (guiQuitRequested ||
2177 (driverInfo.processingPolicies.termination == TerminationPolicy::QUIT && (checkIfCanExit(infos) == true))) {
2178 // Something requested to quit. This can be a user
2179 // interaction with the GUI or (if --completion-policy=quit)
2180 // it could mean that the workflow does not have anything else to do.
2181 // Let's update the GUI one more time and then EXIT.
2182 LOG(info) << "Quitting";
2183 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
2184 } else if (infos.size() != runningWorkflow.devices.size()) {
2185 // If the number of devices is different from
2186 // the DeviceInfos it means the speicification
2187 // does not match what is running, so we need to do
2188 // further scheduling.
2189 driverInfo.states.push_back(DriverState::RUNNING);
2190 driverInfo.states.push_back(DriverState::REDEPLOY_GUI);
2191 driverInfo.states.push_back(DriverState::SCHEDULE);
2192 driverInfo.states.push_back(DriverState::MERGE_CONFIGS);
2193 } else if (runningWorkflow.devices.empty() && driverConfig.batch == true) {
2194 LOG(info) << "No device resulting from the workflow. Quitting.";
2195 // If there are no deviceSpecs, we exit.
2196 driverInfo.states.push_back(DriverState::EXIT);
2197 } else if (runningWorkflow.devices.empty() && driverConfig.batch == false && !guiDeployedOnce) {
2198 // In case of an empty workflow, we need to deploy the GUI at least once.
2199 driverInfo.states.push_back(DriverState::RUNNING);
2200 driverInfo.states.push_back(DriverState::REDEPLOY_GUI);
2201 } else {
2202 driverInfo.states.push_back(DriverState::RUNNING);
2203 }
2204 break;
2205 case DriverState::QUIT_REQUESTED:
2206 LOG(info) << "QUIT_REQUESTED";
2207 guiQuitRequested = true;
2208 // We send SIGCONT to make sure stopped children are resumed
2209 killChildren(infos, SIGCONT);
2210 // We send SIGTERM to make sure we do the STOP transition in FairMQ
2211 killChildren(infos, SIGTERM);
2212 // We have a timer to send SIGUSR1 to make sure we advance all devices
2213 // in a timely manner.
2214 force_step_timer.data = &infos;
2215 uv_timer_start(&force_step_timer, single_step_callback, 0, 300);
2216 driverInfo.states.push_back(DriverState::HANDLE_CHILDREN);
2217 break;
2218 case DriverState::HANDLE_CHILDREN: {
2219 // Run any pending libUV event loop, block if
2220 // any, so that we do not consume CPU time when the driver is
2221 // idle.
2222 uv_run(loop, once ? UV_RUN_ONCE : UV_RUN_NOWAIT);
2223 once = true;
2224 // I allow queueing of more sigchld only when
2225 // I process the previous call
2226 if (forceful_exit == true) {
2227 static bool forcefulExitMessage = true;
2228 if (forcefulExitMessage) {
2229 LOG(info) << "Forceful exit requested.";
2230 forcefulExitMessage = false;
2231 }
2232 killChildren(infos, SIGCONT);
2233 killChildren(infos, SIGKILL);
2234 }
2235 sigchld_requested = false;
2236 driverInfo.sigchldRequested = false;
2237 processChildrenOutput(loop, driverInfo, infos, runningWorkflow.devices, controls);
2238 hasError = processSigChild(infos, runningWorkflow.devices);
2239 allChildrenGone = areAllChildrenGone(infos);
2240 bool canExit = checkIfCanExit(infos);
2241 bool supposedToQuit = (guiQuitRequested || canExit || graceful_exit);
2242
2243 if (allChildrenGone && (supposedToQuit || driverInfo.processingPolicies.termination == TerminationPolicy::QUIT)) {
2244 // We move to the exit, regardless of where we were
2245 driverInfo.states.resize(0);
2246 driverInfo.states.push_back(DriverState::EXIT);
2247 } else if (hasError && driverInfo.processingPolicies.error == TerminationPolicy::QUIT && !supposedToQuit) {
2248 graceful_exit = 1;
2249 force_exit_timer.data = &infos;
2250 static bool forceful_timer_started = false;
2251 if (forceful_timer_started == false) {
2252 forceful_timer_started = true;
2253 uv_timer_start(&force_exit_timer, force_exit_callback, 15000, 3000);
2254 }
2255 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
2256 } else if (allChildrenGone == false && supposedToQuit) {
2257 driverInfo.states.push_back(DriverState::HANDLE_CHILDREN);
2258 } else {
2259 }
2260 } break;
2261 case DriverState::EXIT: {
2262 if (ResourcesMonitoringHelper::isResourcesMonitoringEnabled(driverInfo.resourcesMonitoringInterval)) {
2263 if (driverInfo.resourcesMonitoringDumpInterval) {
2264 uv_timer_stop(&metricDumpTimer);
2265 }
2266 LOG(info) << "Dumping performance metrics to performanceMetrics.json file";
2267 dumpMetricsCallback(&metricDumpTimer);
2268 }
2269 dumpRunSummary(serverContext, driverInfo, infos, runningWorkflow.devices);
2270 // This is a clean exit. Before we do so, if required,
2271 // we dump the configuration of all the devices so that
2272 // we can reuse it. Notice we do not dump anything if
2273 // the workflow was not really run.
2274 // NOTE: is this really what we want? should we run
2275 // SCHEDULE and dump the full configuration as well?
2276 if (infos.empty()) {
2277 return 0;
2278 }
2279 boost::property_tree::ptree finalConfig;
2280 assert(infos.size() == runningWorkflow.devices.size());
2281 for (size_t di = 0; di < infos.size(); ++di) {
2282 auto& info = infos[di];
2283 auto& spec = runningWorkflow.devices[di];
2284 finalConfig.put_child(spec.name, info.currentConfig);
2285 }
2286 LOG(info) << "Dumping used configuration in dpl-config.json";
2287
2288 std::ofstream outDPLConfigFile("dpl-config.json", std::ios::out);
2289 if (outDPLConfigFile.is_open()) {
2290 boost::property_tree::write_json(outDPLConfigFile, finalConfig);
2291 } else {
2292 LOGP(warning, "Could not write out final configuration file. Read only run folder?");
2293 }
2294 if (driverInfo.noSHMCleanup) {
2295 LOGP(warning, "Not cleaning up shared memory.");
2296 } else {
2297 cleanupSHM(driverInfo.uniqueWorkflowId);
2298 }
2299 return calculateExitCode(driverInfo, runningWorkflow.devices, infos);
2300 }
2301 case DriverState::PERFORM_CALLBACKS:
2302 for (auto& callback : driverControl.callbacks) {
2303 callback(workflow, runningWorkflow.devices, deviceExecutions, dataProcessorInfos, commandInfo);
2304 }
2305 driverControl.callbacks.clear();
2306 break;
2307 default:
2308 LOG(error) << "Driver transitioned in an unknown state("
2309 << "current: " << (int)current
2310 << ", previous: " << (int)previous
2311 << "). Shutting down.";
2312 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
2313 }
2314 }
2315 O2_SIGNPOST_END(driver, sid, "driver", "End driver loop");
2316}
2317
2318// Print help
2319void printHelp(bpo::variables_map const& varmap,
2320 bpo::options_description const& executorOptions,
2321 std::vector<DataProcessorSpec> const& physicalWorkflow,
2322 std::vector<ConfigParamSpec> const& currentWorkflowOptions)
2323{
2324 auto mode = varmap["help"].as<std::string>();
2325 bpo::options_description helpOptions;
2326 if (mode == "full" || mode == "short" || mode == "executor") {
2327 helpOptions.add(executorOptions);
2328 }
2329 // this time no veto is applied, so all the options are added for printout
2330 if (mode == "executor") {
2331 // nothing more
2332 } else if (mode == "workflow") {
2333 // executor options and workflow options, skip the actual workflow
2334 o2::framework::WorkflowSpec emptyWorkflow;
2335 helpOptions.add(ConfigParamsHelper::prepareOptionDescriptions(emptyWorkflow, currentWorkflowOptions));
2336 } else if (mode == "full" || mode == "short") {
2337 helpOptions.add(ConfigParamsHelper::prepareOptionDescriptions(physicalWorkflow, currentWorkflowOptions,
2338 bpo::options_description(),
2339 mode));
2340 } else {
2341 helpOptions.add(ConfigParamsHelper::prepareOptionDescriptions(physicalWorkflow, {},
2342 bpo::options_description(),
2343 mode));
2344 }
2345 if (helpOptions.options().size() == 0) {
2346 // the specified argument is invalid, add at leat the executor options
2347 mode += " is an invalid argument, please use correct argument for";
2348 helpOptions.add(executorOptions);
2349 }
2350 std::cout << "ALICE O2 DPL workflow driver" //
2351 << " (" << mode << " help)" << std::endl //
2352 << helpOptions << std::endl; //
2353}
2354
2355// Helper to find out if stdout is actually attached to a pipe.
2357{
2358 struct stat s;
2359 fstat(STDOUT_FILENO, &s);
2360 return ((s.st_mode & S_IFIFO) != 0);
2361}
2362
2364{
2365 struct stat s;
2366 int r = fstat(STDIN_FILENO, &s);
2367 // If stdin cannot be statted, we assume the shell is some sort of
2368 // non-interactive container thing
2369 if (r < 0) {
2370 return false;
2371 }
2372 // If stdin is a pipe or a file, we try to fetch configuration from there
2373 return ((s.st_mode & S_IFIFO) != 0 || (s.st_mode & S_IFREG) != 0);
2374}
2375
2377{
2378 struct CloningSpec {
2379 std::string templateMatcher;
2380 std::string cloneName;
2381 };
2382 auto s = ctx.options().get<std::string>("clone");
2383 std::vector<CloningSpec> specs;
2384 std::string delimiter = ",";
2385
2386 while (s.empty() == false) {
2387 auto newPos = s.find(delimiter);
2388 auto token = s.substr(0, newPos);
2389 auto split = token.find(":");
2390 if (split == std::string::npos) {
2391 throw std::runtime_error("bad clone definition. Syntax <template-processor>:<clone-name>");
2392 }
2393 auto key = token.substr(0, split);
2394 token.erase(0, split + 1);
2395 size_t error;
2396 std::string value = "";
2397 try {
2398 auto numValue = std::stoll(token, &error, 10);
2399 if (token[error] != '\0') {
2400 throw std::runtime_error("bad name for clone:" + token);
2401 }
2402 value = key + "_c" + std::to_string(numValue);
2403 } catch (std::invalid_argument& e) {
2404 value = token;
2405 }
2406 specs.push_back({key, value});
2407 s.erase(0, newPos + (newPos == std::string::npos ? 0 : 1));
2408 }
2409 if (s.empty() == false && specs.empty() == true) {
2410 throw std::runtime_error("bad pipeline definition. Syntax <processor>:<pipeline>");
2411 }
2412
2413 std::vector<DataProcessorSpec> extraSpecs;
2414 for (auto& spec : specs) {
2415 for (auto& processor : workflow) {
2416 if (processor.name == spec.templateMatcher) {
2417 auto clone = processor;
2418 clone.name = spec.cloneName;
2419 extraSpecs.push_back(clone);
2420 }
2421 }
2422 }
2423 workflow.insert(workflow.end(), extraSpecs.begin(), extraSpecs.end());
2424}
2425
2427{
2428 struct PipelineSpec {
2429 std::string matcher;
2430 int64_t pipeline;
2431 };
2432 auto s = ctx.options().get<std::string>("pipeline");
2433 std::vector<PipelineSpec> specs;
2434 std::string delimiter = ",";
2435
2436 while (s.empty() == false) {
2437 auto newPos = s.find(delimiter);
2438 auto token = s.substr(0, newPos);
2439 auto split = token.find(":");
2440 if (split == std::string::npos) {
2441 throw std::runtime_error("bad pipeline definition. Syntax <processor>:<pipeline>");
2442 }
2443 auto key = token.substr(0, split);
2444 token.erase(0, split + 1);
2445 size_t error;
2446 auto value = std::stoll(token, &error, 10);
2447 if (token[error] != '\0') {
2448 throw std::runtime_error("Bad pipeline definition. Expecting integer");
2449 }
2450 specs.push_back({key, value});
2451 s.erase(0, newPos + (newPos == std::string::npos ? 0 : 1));
2452 }
2453 if (s.empty() == false && specs.empty() == true) {
2454 throw std::runtime_error("bad pipeline definition. Syntax <processor>:<pipeline>");
2455 }
2456
2457 for (auto& spec : specs) {
2458 for (auto& processor : workflow) {
2459 if (processor.name == spec.matcher) {
2460 processor.maxInputTimeslices = spec.pipeline;
2461 }
2462 }
2463 }
2464}
2465
2467{
2468 struct LabelsSpec {
2469 std::string_view matcher;
2470 std::vector<std::string> labels;
2471 };
2472 std::vector<LabelsSpec> specs;
2473
2474 auto labelsString = ctx.options().get<std::string>("labels");
2475 if (labelsString.empty()) {
2476 return;
2477 }
2478 std::string_view sv{labelsString};
2479
2480 size_t specStart = 0;
2481 size_t specEnd = 0;
2482 constexpr char specDelim = ',';
2483 constexpr char labelDelim = ':';
2484 do {
2485 specEnd = sv.find(specDelim, specStart);
2486 auto token = sv.substr(specStart, specEnd == std::string_view::npos ? std::string_view::npos : specEnd - specStart);
2487 if (token.empty()) {
2488 throw std::runtime_error("bad labels definition. Syntax <processor>:<label>[:<label>][,<processor>:<label>[:<label>]");
2489 }
2490
2491 size_t labelDelimPos = token.find(labelDelim);
2492 if (labelDelimPos == 0 || labelDelimPos == std::string_view::npos) {
2493 throw std::runtime_error("bad labels definition. Syntax <processor>:<label>[:<label>][,<processor>:<label>[:<label>]");
2494 }
2495 LabelsSpec spec{.matcher = token.substr(0, labelDelimPos), .labels = {}};
2496
2497 size_t labelEnd = labelDelimPos + 1;
2498 do {
2499 size_t labelStart = labelDelimPos + 1;
2500 labelEnd = token.find(labelDelim, labelStart);
2501 auto label = labelEnd == std::string_view::npos ? token.substr(labelStart) : token.substr(labelStart, labelEnd - labelStart);
2502 if (label.empty()) {
2503 throw std::runtime_error("bad labels definition. Syntax <processor>:<label>[:<label>][,<processor>:<label>[:<label>]");
2504 }
2505 spec.labels.emplace_back(label);
2506 labelDelimPos = labelEnd;
2507 } while (labelEnd != std::string_view::npos);
2508
2509 specs.push_back(spec);
2510 specStart = specEnd + 1;
2511 } while (specEnd != std::string_view::npos);
2512
2513 if (labelsString.empty() == false && specs.empty() == true) {
2514 throw std::runtime_error("bad labels definition. Syntax <processor>:<label>[:<label>][,<processor>:<label>[:<label>]");
2515 }
2516
2517 for (auto& spec : specs) {
2518 for (auto& processor : workflow) {
2519 if (processor.name == spec.matcher) {
2520 for (const auto& label : spec.labels) {
2521 if (std::find_if(processor.labels.begin(), processor.labels.end(),
2522 [label](const auto& procLabel) { return procLabel.value == label; }) == processor.labels.end()) {
2523 processor.labels.push_back({label});
2524 }
2525 }
2526 }
2527 }
2528 }
2529}
2530
2532void initialiseDriverControl(bpo::variables_map const& varmap,
2533 DriverInfo& driverInfo,
2534 DriverControl& control)
2535{
2536 // Control is initialised outside the main loop because
2537 // command line options are really affecting control.
2538 control.defaultQuiet = varmap["quiet"].as<bool>();
2539 control.defaultStopped = varmap["stop"].as<bool>();
2540
2541 if (varmap["single-step"].as<bool>()) {
2542 control.state = DriverControlState::STEP;
2543 } else {
2544 control.state = DriverControlState::PLAY;
2545 }
2546
2547 if (varmap["graphviz"].as<bool>()) {
2548 // Dump a graphviz representation of what I will do.
2549 control.callbacks = {[](WorkflowSpec const&,
2550 DeviceSpecs const& specs,
2551 DeviceExecutions const&,
2553 CommandInfo const&) {
2555 }};
2556 control.forcedTransitions = {
2557 DriverState::EXIT, //
2558 DriverState::PERFORM_CALLBACKS, //
2559 DriverState::MERGE_CONFIGS, //
2560 DriverState::IMPORT_CURRENT_WORKFLOW, //
2561 DriverState::MATERIALISE_WORKFLOW //
2562 };
2563 } else if (!varmap["dds"].as<std::string>().empty()) {
2564 // Dump a DDS representation of what I will do.
2565 // Notice that compared to DDS we need to schedule things,
2566 // because DDS needs to be able to have actual Executions in
2567 // order to provide a correct configuration.
2568 control.callbacks = {[filename = varmap["dds"].as<std::string>(),
2569 workflowSuffix = varmap["dds-workflow-suffix"],
2570 driverMode = driverInfo.mode](WorkflowSpec const& workflow,
2571 DeviceSpecs const& specs,
2572 DeviceExecutions const& executions,
2573 DataProcessorInfos& dataProcessorInfos,
2574 CommandInfo const& commandInfo) {
2575 if (filename == "-") {
2576 DDSConfigHelpers::dumpDeviceSpec2DDS(std::cout, driverMode, workflowSuffix.as<std::string>(), workflow, dataProcessorInfos, specs, executions, commandInfo);
2577 } else {
2578 std::ofstream out(filename);
2579 DDSConfigHelpers::dumpDeviceSpec2DDS(out, driverMode, workflowSuffix.as<std::string>(), workflow, dataProcessorInfos, specs, executions, commandInfo);
2580 }
2581 }};
2582 control.forcedTransitions = {
2583 DriverState::EXIT, //
2584 DriverState::PERFORM_CALLBACKS, //
2585 DriverState::MERGE_CONFIGS, //
2586 DriverState::IMPORT_CURRENT_WORKFLOW, //
2587 DriverState::MATERIALISE_WORKFLOW //
2588 };
2589 } else if (!varmap["o2-control"].as<std::string>().empty() or !varmap["mermaid"].as<std::string>().empty()) {
2590 // Dump the workflow in o2-control and/or mermaid format
2591 control.callbacks = {[filename = varmap["mermaid"].as<std::string>(),
2592 workflowName = varmap["o2-control"].as<std::string>()](WorkflowSpec const&,
2593 DeviceSpecs const& specs,
2594 DeviceExecutions const& executions,
2596 CommandInfo const& commandInfo) {
2597 if (!workflowName.empty()) {
2598 dumpDeviceSpec2O2Control(workflowName, specs, executions, commandInfo);
2599 }
2600 if (!filename.empty()) {
2601 if (filename == "-") {
2603 } else {
2604 std::ofstream output(filename);
2606 }
2607 }
2608 }};
2609 control.forcedTransitions = {
2610 DriverState::EXIT, //
2611 DriverState::PERFORM_CALLBACKS, //
2612 DriverState::MERGE_CONFIGS, //
2613 DriverState::IMPORT_CURRENT_WORKFLOW, //
2614 DriverState::MATERIALISE_WORKFLOW //
2615 };
2616
2617 } else if (varmap.count("id")) {
2618 // Add our own stacktrace dumping
2619 if (getenv("O2_NO_CATCHALL_EXCEPTIONS") != nullptr && strcmp(getenv("O2_NO_CATCHALL_EXCEPTIONS"), "0") != 0) {
2620 LOGP(info, "Not instrumenting crash signals because O2_NO_CATCHALL_EXCEPTIONS is set");
2621 gEnv->SetValue("Root.Stacktrace", "no");
2622 gSystem->ResetSignal(kSigSegmentationViolation, kTRUE);
2623 rlimit limit;
2624 if (getrlimit(RLIMIT_CORE, &limit) == 0) {
2625 LOGP(info, "Core limit: {} {}", limit.rlim_cur, limit.rlim_max);
2626 }
2627 }
2628 if (varmap["stacktrace-on-signal"].as<std::string>() == "simple" && (getenv("O2_NO_CATCHALL_EXCEPTIONS") == nullptr || strcmp(getenv("O2_NO_CATCHALL_EXCEPTIONS"), "0") == 0)) {
2629 LOGP(info, "Instrumenting crash signals");
2630 signal(SIGSEGV, handle_crash);
2631 signal(SIGABRT, handle_crash);
2632 signal(SIGBUS, handle_crash);
2633 signal(SIGILL, handle_crash);
2634 signal(SIGFPE, handle_crash);
2635 }
2636 // FIXME: for the time being each child needs to recalculate the workflow,
2637 // so that it can understand what it needs to do. This is obviously
2638 // a bad idea. In the future we should have the client be pushed
2639 // it's own configuration by the driver.
2640 control.forcedTransitions = {
2641 DriverState::DO_CHILD, //
2642 DriverState::BIND_GUI_PORT, //
2643 DriverState::MERGE_CONFIGS, //
2644 DriverState::IMPORT_CURRENT_WORKFLOW, //
2645 DriverState::MATERIALISE_WORKFLOW //
2646 };
2647 } else if ((varmap["dump-workflow"].as<bool>() == true) || (varmap["run"].as<bool>() == false && varmap.count("id") == 0 && isOutputToPipe())) {
2648 control.callbacks = {[filename = varmap["dump-workflow-file"].as<std::string>()](WorkflowSpec const& workflow,
2649 DeviceSpecs const&,
2650 DeviceExecutions const&,
2651 DataProcessorInfos& dataProcessorInfos,
2652 CommandInfo const& commandInfo) {
2653 if (filename == "-") {
2654 WorkflowSerializationHelpers::dump(std::cout, workflow, dataProcessorInfos, commandInfo);
2655 // FIXME: this is to avoid trailing garbage..
2656 exit(0);
2657 } else {
2658 std::ofstream output(filename);
2659 WorkflowSerializationHelpers::dump(output, workflow, dataProcessorInfos, commandInfo);
2660 }
2661 }};
2662 control.forcedTransitions = {
2663 DriverState::EXIT, //
2664 DriverState::PERFORM_CALLBACKS, //
2665 DriverState::MERGE_CONFIGS, //
2666 DriverState::IMPORT_CURRENT_WORKFLOW, //
2667 DriverState::MATERIALISE_WORKFLOW //
2668 };
2669 } else {
2670 // By default we simply start the main loop of the driver.
2671 control.forcedTransitions = {
2672 DriverState::INIT, //
2673 DriverState::BIND_GUI_PORT, //
2674 DriverState::IMPORT_CURRENT_WORKFLOW, //
2675 DriverState::MATERIALISE_WORKFLOW //
2676 };
2677 }
2678}
2679
2681void conflicting_options(const boost::program_options::variables_map& vm,
2682 const std::string& opt1, const std::string& opt2)
2683{
2684 if (vm.count(opt1) && !vm[opt1].defaulted() &&
2685 vm.count(opt2) && !vm[opt2].defaulted()) {
2686 throw std::logic_error(std::string("Conflicting options '") +
2687 opt1 + "' and '" + opt2 + "'.");
2688 }
2689}
2690
2691template <typename T>
2693 std::vector<T>& v,
2694 std::vector<int>& indices)
2695{
2696 using std::swap; // to permit Koenig lookup
2697 for (int i = 0; i < (int)indices.size(); i++) {
2698 auto current = i;
2699 while (i != indices[current]) {
2700 auto next = indices[current];
2701 swap(v[current], v[next]);
2702 indices[current] = current;
2703 current = next;
2704 }
2705 indices[current] = current;
2706 }
2707}
2708
2709// Check if the workflow is resiliant to failures
2710void checkNonResiliency(std::vector<DataProcessorSpec> const& specs,
2711 std::vector<std::pair<int, int>> const& edges)
2712{
2713 auto checkExpendable = [](DataProcessorLabel const& label) {
2714 return label.value == "expendable";
2715 };
2716 auto checkResilient = [](DataProcessorLabel const& label) {
2717 return label.value == "resilient" || label.value == "expendable";
2718 };
2719
2720 for (auto& edge : edges) {
2721 auto& src = specs[edge.first];
2722 auto& dst = specs[edge.second];
2723 if (std::none_of(src.labels.begin(), src.labels.end(), checkExpendable)) {
2724 continue;
2725 }
2726 if (std::any_of(dst.labels.begin(), dst.labels.end(), checkResilient)) {
2727 continue;
2728 }
2729 throw std::runtime_error("Workflow is not resiliant to failures. Processor " + dst.name + " gets inputs from expendable devices, but is not marked as expendable or resilient itself.");
2730 }
2731}
2732
2733std::string debugTopoInfo(std::vector<DataProcessorSpec> const& specs,
2734 std::vector<TopoIndexInfo> const& infos,
2735 std::vector<std::pair<int, int>> const& edges)
2736{
2737 std::ostringstream out;
2738
2739 out << "\nTopological info:\n";
2740 for (auto& ti : infos) {
2741 out << specs[ti.index].name << " (index: " << ti.index << ", layer: " << ti.layer << ")\n";
2742 out << " Inputs:\n";
2743 for (auto& ii : specs[ti.index].inputs) {
2744 out << " - " << DataSpecUtils::describe(ii) << "\n";
2745 }
2746 out << "\n Outputs:\n";
2747 for (auto& ii : specs[ti.index].outputs) {
2748 out << " - " << DataSpecUtils::describe(ii) << "\n";
2749 }
2750 }
2751 out << "\nEdges values:\n";
2752 for (auto& e : edges) {
2753 out << specs[e.second].name << " depends on " << specs[e.first].name << "\n";
2754 }
2755 for (auto& d : specs) {
2756 out << "- " << d.name << std::endl;
2757 }
2759 return out.str();
2760}
2761
2762void enableSignposts(std::string const& signpostsToEnable)
2763{
2764 static pid_t pid = getpid();
2765 if (signpostsToEnable.empty() == true) {
2766 auto printAllSignposts = [](char const* name, void* l, void* context) {
2767 auto* log = (_o2_log_t*)l;
2768 LOGP(detail, "Signpost stream {} disabled. Enable it with o2-log -p {} -a {}", name, pid, (void*)&log->stacktrace);
2769 return true;
2770 };
2771 o2_walk_logs(printAllSignposts, nullptr);
2772 return;
2773 }
2774 auto matchingLogEnabler = [](char const* name, void* l, void* context) {
2775 auto* log = (_o2_log_t*)l;
2776 auto* selectedName = (char const*)context;
2777 std::string prefix = "ch.cern.aliceo2.";
2778 auto* last = strchr(selectedName, ':');
2779 int maxDepth = 1;
2780 if (last) {
2781 char* err;
2782 maxDepth = strtol(last + 1, &err, 10);
2783 if (*(last + 1) == '\0' || *err != '\0') {
2784 maxDepth = 1;
2785 }
2786 }
2787
2788 auto fullName = prefix + std::string{selectedName, last ? last - selectedName : strlen(selectedName)};
2789 if (fullName == name) {
2790 LOGP(info, "Enabling signposts for stream \"{}\" with depth {}.", fullName, maxDepth);
2791 _o2_log_set_stacktrace(log, maxDepth);
2792 return false;
2793 } else {
2794 LOGP(info, "Signpost stream \"{}\" disabled. Enable it with o2-log -p {} -a {}", name, pid, (void*)&log->stacktrace);
2795 }
2796 return true;
2797 };
2798 // Split signpostsToEnable by comma using strtok_r
2799 char* saveptr;
2800 char* src = const_cast<char*>(signpostsToEnable.data());
2801 auto* token = strtok_r(src, ",", &saveptr);
2802 while (token) {
2803 o2_walk_logs(matchingLogEnabler, token);
2804 token = strtok_r(nullptr, ",", &saveptr);
2805 }
2806}
2807
2808void overrideAll(o2::framework::ConfigContext& ctx, std::vector<o2::framework::DataProcessorSpec>& workflow)
2809{
2810 overrideCloning(ctx, workflow);
2811 overridePipeline(ctx, workflow);
2812 overrideLabels(ctx, workflow);
2813}
2814
2815o2::framework::ConfigContext createConfigContext(std::unique_ptr<ConfigParamRegistry>& workflowOptionsRegistry,
2816 o2::framework::ServiceRegistry& configRegistry,
2817 std::vector<o2::framework::ConfigParamSpec>& workflowOptions,
2818 std::vector<o2::framework::ConfigParamSpec>& extraOptions, int argc, char** argv)
2819{
2820 std::vector<std::unique_ptr<o2::framework::ParamRetriever>> retrievers;
2821 std::unique_ptr<o2::framework::ParamRetriever> retriever{new o2::framework::BoostOptionsRetriever(true, argc, argv)};
2822 retrievers.emplace_back(std::move(retriever));
2823 auto workflowOptionsStore = std::make_unique<o2::framework::ConfigParamStore>(workflowOptions, std::move(retrievers));
2824 workflowOptionsStore->preload();
2825 workflowOptionsStore->activate();
2826 workflowOptionsRegistry = std::make_unique<ConfigParamRegistry>(std::move(workflowOptionsStore));
2827 extraOptions = o2::framework::ConfigParamDiscovery::discover(*workflowOptionsRegistry, argc, argv);
2828 for (auto& extra : extraOptions) {
2829 workflowOptions.push_back(extra);
2830 }
2831
2832 return o2::framework::ConfigContext(*workflowOptionsRegistry, o2::framework::ServiceRegistryRef{configRegistry}, argc, argv);
2833}
2834
2835std::unique_ptr<o2::framework::ServiceRegistry> createRegistry()
2836{
2837 return std::make_unique<o2::framework::ServiceRegistry>();
2838}
2839
2840// This is a toy executor for the workflow spec
2841// What it needs to do is:
2842//
2843// - Print the properties of each DataProcessorSpec
2844// - Fork one process per DataProcessorSpec
2845// - Parent -> wait for all the children to complete (eventually
2846// killing them all on ctrl-c).
2847// - Child, pick the data-processor ID and start a O2DataProcessorDevice for
2848// each DataProcessorSpec
2849int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& workflow,
2850 std::vector<ChannelConfigurationPolicy> const& channelPolicies,
2851 std::vector<CompletionPolicy> const& completionPolicies,
2852 std::vector<DispatchPolicy> const& dispatchPolicies,
2853 std::vector<ResourcePolicy> const& resourcePolicies,
2854 std::vector<CallbacksPolicy> const& callbacksPolicies,
2855 std::vector<SendingPolicy> const& sendingPolicies,
2856 std::vector<ConfigParamSpec> const& currentWorkflowOptions,
2857 std::vector<ConfigParamSpec> const& detectedParams,
2858 o2::framework::ConfigContext& configContext)
2859{
2860 // Peek very early in the driver options and look for
2861 // signposts, so the we can enable it without going through the whole dance
2862 if (getenv("DPL_DRIVER_SIGNPOSTS")) {
2863 enableSignposts(getenv("DPL_DRIVER_SIGNPOSTS"));
2864 }
2865
2866 std::vector<std::string> currentArgs;
2867 std::vector<PluginInfo> plugins;
2868 std::vector<ForwardingPolicy> forwardingPolicies = ForwardingPolicy::createDefaultPolicies();
2869
2870 for (int ai = 1; ai < argc; ++ai) {
2871 currentArgs.emplace_back(argv[ai]);
2872 }
2873
2874 WorkflowInfo currentWorkflow{
2875 argv[0],
2876 currentArgs,
2877 currentWorkflowOptions};
2878
2879 ProcessingPolicies processingPolicies;
2880 enum LogParsingHelpers::LogLevel minFailureLevel;
2881 bpo::options_description executorOptions("Executor options");
2882 const char* helpDescription = "print help: short, full, executor, or processor name";
2883 enum DriverMode driverMode;
2884 executorOptions.add_options() //
2885 ("help,h", bpo::value<std::string>()->implicit_value("short"), helpDescription) // //
2886 ("quiet,q", bpo::value<bool>()->zero_tokens()->default_value(false), "quiet operation") // //
2887 ("stop,s", bpo::value<bool>()->zero_tokens()->default_value(false), "stop before device start") // //
2888 ("single-step", bpo::value<bool>()->zero_tokens()->default_value(false), "start in single step mode") // //
2889 ("batch,b", bpo::value<std::vector<std::string>>()->zero_tokens()->composing(), "batch processing mode") // //
2890 ("no-batch", bpo::value<bool>()->zero_tokens(), "force gui processing mode") // //
2891 ("no-cleanup", bpo::value<bool>()->zero_tokens()->default_value(false), "do not cleanup the shm segment") // //
2892 ("hostname", bpo::value<std::string>()->default_value("localhost"), "hostname to deploy") // //
2893 ("resources", bpo::value<std::string>()->default_value(""), "resources allocated for the workflow") // //
2894 ("start-port,p", bpo::value<unsigned short>()->default_value(22000), "start port to allocate") // //
2895 ("port-range,pr", bpo::value<unsigned short>()->default_value(1000), "ports in range") // //
2896 ("completion-policy,c", bpo::value<TerminationPolicy>(&processingPolicies.termination)->default_value(TerminationPolicy::QUIT), // //
2897 "what to do when processing is finished: quit, wait") // //
2898 ("error-policy", bpo::value<TerminationPolicy>(&processingPolicies.error)->default_value(TerminationPolicy::QUIT), // //
2899 "what to do when a device has an error: quit, wait") // //
2900 ("min-failure-level", bpo::value<LogParsingHelpers::LogLevel>(&minFailureLevel)->default_value(LogParsingHelpers::LogLevel::Fatal), // //
2901 "minimum message level which will be considered as fatal and exit with 1") // //
2902 ("graphviz,g", bpo::value<bool>()->zero_tokens()->default_value(false), "produce graphviz output") // //
2903 ("mermaid", bpo::value<std::string>()->default_value(""), "produce graph output in mermaid format in file under specified name or on stdout if argument is \"-\"") // //
2904 ("timeout,t", bpo::value<uint64_t>()->default_value(0), "forced exit timeout (in seconds)") // //
2905 ("dds,D", bpo::value<std::string>()->default_value(""), "create DDS configuration") // //
2906 ("dds-workflow-suffix,D", bpo::value<std::string>()->default_value(""), "suffix for DDS names") // //
2907 ("dump-workflow,dump", bpo::value<bool>()->zero_tokens()->default_value(false), "dump workflow as JSON") // //
2908 ("dump-workflow-file", bpo::value<std::string>()->default_value("-"), "file to which do the dump") // //
2909 ("driver-mode", bpo::value<DriverMode>(&driverMode)->default_value(DriverMode::STANDALONE), R"(how to run the driver. default: "standalone". Valid: "embedded")") // //
2910 ("run", bpo::value<bool>()->zero_tokens()->default_value(false), "run workflow merged so far. It implies --batch. Use --no-batch to see the GUI") // //
2911 ("no-IPC", bpo::value<bool>()->zero_tokens()->default_value(false), "disable IPC topology optimization") // //
2912 ("o2-control,o2", bpo::value<std::string>()->default_value(""), "dump O2 Control workflow configuration under the specified name") //
2913 ("resources-monitoring", bpo::value<unsigned short>()->default_value(0), "enable cpu/memory monitoring for provided interval in seconds") //
2914 ("resources-monitoring-dump-interval", bpo::value<unsigned short>()->default_value(0), "dump monitoring information to disk every provided seconds"); //
2915 // some of the options must be forwarded by default to the device
2916 executorOptions.add(DeviceSpecHelpers::getForwardedDeviceOptions());
2917
2918 gHiddenDeviceOptions.add_options() //
2919 ("id,i", bpo::value<std::string>(), "device id for child spawning") //
2920 ("channel-config", bpo::value<std::vector<std::string>>(), "channel configuration") //
2921 ("control", "control plugin") //
2922 ("log-color", "logging color scheme")("color", "logging color scheme");
2923
2924 bpo::options_description visibleOptions;
2925 visibleOptions.add(executorOptions);
2926
2927 auto physicalWorkflow = workflow;
2928 std::map<std::string, size_t> rankIndex;
2929 // We remove the duplicates because for the moment child get themself twice:
2930 // once from the actual definition in the child, a second time from the
2931 // configuration they get passed by their parents.
2932 // Notice that we do not know in which order we will get the workflows, so
2933 // while we keep the order of DataProcessors we reshuffle them based on
2934 // some hopefully unique hash.
2935 size_t workflowHashA = 0;
2936 std::hash<std::string> hash_fn;
2937
2938 for (auto& dp : workflow) {
2939 workflowHashA += hash_fn(dp.name);
2940 }
2941
2942 for (auto& dp : workflow) {
2943 rankIndex.insert(std::make_pair(dp.name, workflowHashA));
2944 }
2945
2946 std::vector<DataProcessorInfo> dataProcessorInfos;
2947 CommandInfo commandInfo{};
2948
2949 if (isatty(STDIN_FILENO) == false && isInputConfig()) {
2950 std::vector<DataProcessorSpec> importedWorkflow;
2951 bool previousWorked = WorkflowSerializationHelpers::import(std::cin, importedWorkflow, dataProcessorInfos, commandInfo);
2952 if (previousWorked == false) {
2953 exit(1);
2954 }
2955
2956 size_t workflowHashB = 0;
2957 for (auto& dp : importedWorkflow) {
2958 workflowHashB += hash_fn(dp.name);
2959 }
2960
2961 // FIXME: Streamline...
2962 // We remove the duplicates because for the moment child get themself twice:
2963 // once from the actual definition in the child, a second time from the
2964 // configuration they get passed by their parents.
2965 for (auto& dp : importedWorkflow) {
2966 auto found = std::find_if(physicalWorkflow.begin(), physicalWorkflow.end(),
2967 [&name = dp.name](DataProcessorSpec const& spec) { return spec.name == name; });
2968 if (found == physicalWorkflow.end()) {
2969 physicalWorkflow.push_back(dp);
2970 rankIndex.insert(std::make_pair(dp.name, workflowHashB));
2971 }
2972 }
2973 }
2974
2979 for (auto& dp : physicalWorkflow) {
2980 auto isExpendable = [](DataProcessorLabel const& label) { return label.value == "expendable" || label.value == "non-critical"; };
2981 if (std::find_if(dp.labels.begin(), dp.labels.end(), isExpendable) != dp.labels.end()) {
2982 for (auto& output : dp.outputs) {
2983 if (output.lifetime == Lifetime::Timeframe) {
2984 output.lifetime = Lifetime::Sporadic;
2985 }
2986 }
2987 }
2988 }
2989
2991 OverrideServiceSpecs driverServicesOverride = ServiceSpecHelpers::parseOverrides(getenv("DPL_DRIVER_OVERRIDE_SERVICES"));
2993 // We insert the hash for the internal devices.
2994 WorkflowHelpers::injectServiceDevices(physicalWorkflow, configContext);
2995 auto reader = std::find_if(physicalWorkflow.begin(), physicalWorkflow.end(), [](DataProcessorSpec& spec) { return spec.name == "internal-dpl-aod-reader"; });
2996 if (reader != physicalWorkflow.end()) {
2997 driverServices.push_back(ArrowSupport::arrowBackendSpec());
2998 }
2999 for (auto& service : driverServices) {
3000 if (service.injectTopology == nullptr) {
3001 continue;
3002 }
3003 WorkflowSpecNode node{physicalWorkflow};
3004 service.injectTopology(node, configContext);
3005 }
3006 for (auto& dp : physicalWorkflow) {
3007 if (dp.name.rfind("internal-", 0) == 0) {
3008 rankIndex.insert(std::make_pair(dp.name, hash_fn("internal")));
3009 }
3010 }
3011
3012 // We sort dataprocessors and Inputs / outputs by name, so that the edges are
3013 // always in the same order.
3014 std::stable_sort(physicalWorkflow.begin(), physicalWorkflow.end(), [](DataProcessorSpec const& a, DataProcessorSpec const& b) {
3015 return a.name < b.name;
3016 });
3017
3018 for (auto& dp : physicalWorkflow) {
3019 std::stable_sort(dp.inputs.begin(), dp.inputs.end(),
3020 [](InputSpec const& a, InputSpec const& b) { return DataSpecUtils::describe(a) < DataSpecUtils::describe(b); });
3021 std::stable_sort(dp.outputs.begin(), dp.outputs.end(),
3022 [](OutputSpec const& a, OutputSpec const& b) { return DataSpecUtils::describe(a) < DataSpecUtils::describe(b); });
3023 }
3024
3025 // Create a list of all the edges, so that we can do a topological sort
3026 // before we create the graph.
3027 std::vector<std::pair<int, int>> edges;
3028
3029 if (physicalWorkflow.size() > 1) {
3030 edges = TopologyPolicyHelpers::buildEdges(physicalWorkflow);
3031
3032 auto topoInfos = WorkflowHelpers::topologicalSort(physicalWorkflow.size(), &edges[0].first, &edges[0].second, sizeof(std::pair<int, int>), edges.size());
3033 if (topoInfos.size() != physicalWorkflow.size()) {
3034 // Check missing resilincy of one of the tasks
3035 checkNonResiliency(physicalWorkflow, edges);
3036 throw std::runtime_error("Unable to do topological sort of the resulting workflow. Do you have loops?\n" + debugTopoInfo(physicalWorkflow, topoInfos, edges));
3037 }
3038 // Sort by layer and then by name, to ensure stability.
3039 std::stable_sort(topoInfos.begin(), topoInfos.end(), [&workflow = physicalWorkflow](TopoIndexInfo const& a, TopoIndexInfo const& b) {
3040 auto aRank = std::make_tuple(a.layer, -workflow.at(a.index).outputs.size(), workflow.at(a.index).name);
3041 auto bRank = std::make_tuple(b.layer, -workflow.at(b.index).outputs.size(), workflow.at(b.index).name);
3042 return aRank < bRank;
3043 });
3044 // Reverse index and apply the result
3045 std::vector<int> dataProcessorOrder;
3046 dataProcessorOrder.resize(topoInfos.size());
3047 for (size_t i = 0; i < topoInfos.size(); ++i) {
3048 dataProcessorOrder[topoInfos[i].index] = i;
3049 }
3050 std::vector<int> newLocations;
3051 newLocations.resize(dataProcessorOrder.size());
3052 for (size_t i = 0; i < dataProcessorOrder.size(); ++i) {
3053 newLocations[dataProcessorOrder[i]] = i;
3054 }
3055 apply_permutation(physicalWorkflow, newLocations);
3056 }
3057
3058 // Use the hidden options as veto, all config specs matching a definition
3059 // in the hidden options are skipped in order to avoid duplicate definitions
3060 // in the main parser. Note: all config specs are forwarded to devices
3061 visibleOptions.add(ConfigParamsHelper::prepareOptionDescriptions(physicalWorkflow, currentWorkflowOptions, gHiddenDeviceOptions));
3062
3063 bpo::options_description od;
3064 od.add(visibleOptions);
3065 od.add(gHiddenDeviceOptions);
3066
3067 // FIXME: decide about the policy for handling unrecognized arguments
3068 // command_line_parser with option allow_unregistered() can be used
3069 using namespace bpo::command_line_style;
3070 auto style = (allow_short | short_allow_adjacent | short_allow_next | allow_long | long_allow_adjacent | long_allow_next | allow_sticky | allow_dash_for_short);
3071 bpo::variables_map varmap;
3072 try {
3073 bpo::store(
3074 bpo::command_line_parser(argc, argv)
3075 .options(od)
3076 .style(style)
3077 .run(),
3078 varmap);
3079 } catch (std::exception const& e) {
3080 LOGP(error, "error parsing options of {}: {}", argv[0], e.what());
3081 exit(1);
3082 }
3083 conflicting_options(varmap, "dds", "o2-control");
3084 conflicting_options(varmap, "dds", "dump-workflow");
3085 conflicting_options(varmap, "dds", "run");
3086 conflicting_options(varmap, "dds", "graphviz");
3087 conflicting_options(varmap, "o2-control", "dump-workflow");
3088 conflicting_options(varmap, "o2-control", "run");
3089 conflicting_options(varmap, "o2-control", "graphviz");
3090 conflicting_options(varmap, "run", "dump-workflow");
3091 conflicting_options(varmap, "run", "graphviz");
3092 conflicting_options(varmap, "run", "mermaid");
3093 conflicting_options(varmap, "dump-workflow", "graphviz");
3094 conflicting_options(varmap, "no-batch", "batch");
3095
3096 if (varmap.count("help")) {
3097 printHelp(varmap, executorOptions, physicalWorkflow, currentWorkflowOptions);
3098 exit(0);
3099 }
3103 if (varmap.count("severity")) {
3104 auto logLevel = varmap["severity"].as<std::string>();
3105 if (logLevel == "debug") {
3106 fair::Logger::SetConsoleSeverity(fair::Severity::debug);
3107 } else if (logLevel == "detail") {
3108 fair::Logger::SetConsoleSeverity(fair::Severity::detail);
3109 } else if (logLevel == "info") {
3110 fair::Logger::SetConsoleSeverity(fair::Severity::info);
3111 } else if (logLevel == "warning") {
3112 fair::Logger::SetConsoleSeverity(fair::Severity::warning);
3113 } else if (logLevel == "error") {
3114 fair::Logger::SetConsoleSeverity(fair::Severity::error);
3115 } else if (logLevel == "important") {
3116 fair::Logger::SetConsoleSeverity(fair::Severity::important);
3117 } else if (logLevel == "alarm") {
3118 fair::Logger::SetConsoleSeverity(fair::Severity::alarm);
3119 } else if (logLevel == "critical") {
3120 fair::Logger::SetConsoleSeverity(fair::Severity::critical);
3121 } else if (logLevel == "fatal") {
3122 fair::Logger::SetConsoleSeverity(fair::Severity::fatal);
3123 } else {
3124 LOGP(error, "Invalid log level '{}'", logLevel);
3125 exit(1);
3126 }
3127 }
3128
3129 enableSignposts(varmap["signposts"].as<std::string>());
3130
3131 auto evaluateBatchOption = [&varmap]() -> bool {
3132 if (varmap.count("no-batch") > 0) {
3133 return false;
3134 }
3135 if (varmap.count("batch") == 0) {
3136 // default value
3137 return isatty(fileno(stdout)) == 0;
3138 }
3139 // FIXME: should actually use the last value, but for some reason the
3140 // values are not filled into the vector, even if specifying `-b true`
3141 // need to find out why the boost program options example is not working
3142 // in our case. Might depend on the parser options
3143 // auto value = varmap["batch"].as<std::vector<std::string>>();
3144 return true;
3145 };
3146 DriverInfo driverInfo{
3147 .sendingPolicies = sendingPolicies,
3148 .forwardingPolicies = forwardingPolicies,
3149 .callbacksPolicies = callbacksPolicies};
3150 driverInfo.states.reserve(10);
3151 driverInfo.sigintRequested = false;
3152 driverInfo.sigchldRequested = false;
3153 driverInfo.channelPolicies = channelPolicies;
3154 driverInfo.completionPolicies = completionPolicies;
3155 driverInfo.dispatchPolicies = dispatchPolicies;
3156 driverInfo.resourcePolicies = resourcePolicies;
3157 driverInfo.argc = argc;
3158 driverInfo.argv = argv;
3159 driverInfo.noSHMCleanup = varmap["no-cleanup"].as<bool>();
3160 driverInfo.processingPolicies.termination = varmap["completion-policy"].as<TerminationPolicy>();
3161 driverInfo.processingPolicies.earlyForward = varmap["early-forward-policy"].as<EarlyForwardPolicy>();
3162 driverInfo.mode = varmap["driver-mode"].as<DriverMode>();
3163
3164 auto batch = evaluateBatchOption();
3165 DriverConfig driverConfig{
3166 .batch = batch,
3167 .driverHasGUI = (batch == false) || getenv("DPL_DRIVER_REMOTE_GUI") != nullptr,
3168 };
3169
3170 if (varmap["error-policy"].defaulted() && driverConfig.batch == false) {
3171 driverInfo.processingPolicies.error = TerminationPolicy::WAIT;
3172 } else {
3173 driverInfo.processingPolicies.error = varmap["error-policy"].as<TerminationPolicy>();
3174 }
3175 driverInfo.minFailureLevel = varmap["min-failure-level"].as<LogParsingHelpers::LogLevel>();
3176 driverInfo.startTime = uv_hrtime();
3177 driverInfo.startTimeMsFromEpoch = std::chrono::duration_cast<std::chrono::milliseconds>(
3178 std::chrono::system_clock::now().time_since_epoch())
3179 .count();
3180 driverInfo.timeout = varmap["timeout"].as<uint64_t>();
3181 driverInfo.deployHostname = varmap["hostname"].as<std::string>();
3182 driverInfo.resources = varmap["resources"].as<std::string>();
3183 driverInfo.resourcesMonitoringInterval = varmap["resources-monitoring"].as<unsigned short>();
3184 driverInfo.resourcesMonitoringDumpInterval = varmap["resources-monitoring-dump-interval"].as<unsigned short>();
3185
3186 // FIXME: should use the whole dataProcessorInfos, actually...
3187 driverInfo.processorInfo = dataProcessorInfos;
3188 driverInfo.configContext = &configContext;
3189
3190 DriverControl driverControl;
3191 initialiseDriverControl(varmap, driverInfo, driverControl);
3192
3193 commandInfo.merge(CommandInfo(argc, argv));
3194
3195 std::string frameworkId;
3196 // If the id is set, this means this is a device,
3197 // otherwise this is the driver.
3198 if (varmap.count("id")) {
3199 // The framework id does not want to know anything about DDS template expansion
3200 // so we simply drop it. Notice that the "id" Property is still the same as the
3201 // original --id option.
3202 frameworkId = std::regex_replace(varmap["id"].as<std::string>(), std::regex{"_dds.*"}, "");
3203 driverInfo.uniqueWorkflowId = fmt::format("{}", getppid());
3204 driverInfo.defaultDriverClient = "stdout://";
3205 } else {
3206 driverInfo.uniqueWorkflowId = fmt::format("{}", getpid());
3207 driverInfo.defaultDriverClient = "ws://";
3208 }
3209 return runStateMachine(physicalWorkflow,
3210 currentWorkflow,
3211 dataProcessorInfos,
3212 commandInfo,
3213 driverControl,
3214 driverInfo,
3215 driverConfig,
3217 detectedParams,
3218 varmap,
3219 driverServices,
3220 frameworkId);
3221}
3222
3223void doBoostException(boost::exception&, char const* processName)
3224{
3225 LOGP(error, "error while setting up workflow in {}: {}",
3226 processName, boost::current_exception_diagnostic_information(true));
3227}
3228#pragma GCC diagnostic push
struct uv_timer_s uv_timer_t
struct uv_async_s uv_async_t
struct uv_handle_s uv_handle_t
struct uv_poll_s uv_poll_t
struct uv_loop_s uv_loop_t
std::vector< OutputRoute > routes
int32_t i
int32_t retVal
void output(const std::map< std::string, ChannelStat > &channels)
Definition rawdump.cxx:197
o2::phos::PHOSEnergySlot es
uint16_t pos
Definition RawData.h:3
uint16_t pid
Definition RawData.h:2
#define O2_SIGNPOST_EVENT_EMIT_ERROR(log, id, name, format,...)
Definition Signpost.h:553
o2_log_handle_t * o2_walk_logs(bool(*callback)(char const *name, void *log, void *context), void *context=nullptr)
#define O2_DECLARE_DYNAMIC_LOG(name)
Definition Signpost.h:489
#define O2_SIGNPOST_ID_FROM_POINTER(name, log, pointer)
Definition Signpost.h:505
#define O2_SIGNPOST_END(log, id, name, format,...)
Definition Signpost.h:608
void _o2_log_set_stacktrace(_o2_log_t *log, int stacktrace)
#define O2_SIGNPOST_ID_GENERATE(name, log)
Definition Signpost.h:506
#define O2_SIGNPOST_EVENT_EMIT(log, id, name, format,...)
Definition Signpost.h:522
#define O2_SIGNPOST_START(log, id, name, format,...)
Definition Signpost.h:602
std::ostringstream debug
o2::monitoring::Monitoring Monitoring
StringRef key
ConfigParamRegistry & options() const
T get(uint32_t y, uint32_t x) const
Definition Array2D.h:199
void registerService(ServiceTypeHash typeHash, void *service, ServiceKind kind, char const *name=nullptr) const
bool match(const std::vector< std::string > &queries, const char *pattern)
Definition dcs-ccdb.cxx:229
GLenum mode
Definition glcorearb.h:266
GLenum src
Definition glcorearb.h:1767
GLint GLsizei count
Definition glcorearb.h:399
GLuint64EXT * result
Definition glcorearb.h:5662
GLuint buffer
Definition glcorearb.h:655
GLsizeiptr size
Definition glcorearb.h:659
GLuint GLuint end
Definition glcorearb.h:469
const GLdouble * v
Definition glcorearb.h:832
GLenum array
Definition glcorearb.h:4274
GLuint index
Definition glcorearb.h:781
GLuint const GLchar * name
Definition glcorearb.h:781
GLboolean GLboolean GLboolean b
Definition glcorearb.h:1233
GLsizei const GLfloat * value
Definition glcorearb.h:819
GLenum GLenum dst
Definition glcorearb.h:1767
GLboolean * data
Definition glcorearb.h:298
GLintptr offset
Definition glcorearb.h:660
GLuint GLsizei const GLchar * label
Definition glcorearb.h:2519
GLsizei GLenum const void * indices
Definition glcorearb.h:400
typedef void(APIENTRYP PFNGLCULLFACEPROC)(GLenum mode)
GLboolean r
Definition glcorearb.h:1233
GLenum GLenum GLsizei len
Definition glcorearb.h:4232
GLenum GLfloat param
Definition glcorearb.h:271
GLboolean GLboolean GLboolean GLboolean a
Definition glcorearb.h:1233
GLuint GLuint stream
Definition glcorearb.h:1806
GLint ref
Definition glcorearb.h:291
GLenum GLuint GLenum GLsizei const GLchar * buf
Definition glcorearb.h:2514
GLuint id
Definition glcorearb.h:650
GLuint counter
Definition glcorearb.h:3987
Defining PrimaryVertex explicitly as messageable.
Definition TFIDInfo.h:20
std::vector< ServiceSpec > ServiceSpecs
RuntimeErrorRef runtime_error(const char *)
EarlyForwardPolicy
When to enable the early forwarding optimization:
std::vector< OverrideServiceSpec > OverrideServiceSpecs
void parse_http_request(char *start, size_t size, HTTPParser *parser)
RuntimeError & error_from_ref(RuntimeErrorRef)
std::vector< DataProcessorSpec > WorkflowSpec
AlgorithmSpec::ProcessCallback adaptStateless(LAMBDA l)
RuntimeErrorRef runtime_error_f(const char *,...)
void dumpDeviceSpec2O2Control(std::string workflowName, std::vector< DeviceSpec > const &specs, std::vector< DeviceExecution > const &executions, CommandInfo const &commandInfo)
Dumps the AliECS compatible workflow and task templates for a DPL workflow.
std::string to_string(gsl::span< T, Size > span)
Definition common.h:52
std::string filename()
void empty(int)
std::vector< std::string > split(const std::string &str, char delimiter=',')
int runStateMachine(DataProcessorSpecs const &workflow, WorkflowInfo const &workflowInfo, DataProcessorInfos const &previousDataProcessorInfos, CommandInfo const &commandInfo, DriverControl &driverControl, DriverInfo &driverInfo, DriverConfig &driverConfig, std::vector< DeviceMetricsInfo > &metricsInfos, std::vector< ConfigParamSpec > const &detectedParams, boost::program_options::variables_map &varmap, std::vector< ServiceSpec > &driverServices, std::string frameworkId)
AlgorithmSpec dryRun(DeviceSpec const &spec)
auto bindGUIPort
void getChildData(int infd, DeviceInfo &outinfo)
void overrideLabels(ConfigContext &ctx, WorkflowSpec &workflow)
void apply_permutation(std::vector< T > &v, std::vector< int > &indices)
int doMain(int argc, char **argv, o2::framework::WorkflowSpec const &workflow, std::vector< ChannelConfigurationPolicy > const &channelPolicies, std::vector< CompletionPolicy > const &completionPolicies, std::vector< DispatchPolicy > const &dispatchPolicies, std::vector< ResourcePolicy > const &resourcePolicies, std::vector< CallbacksPolicy > const &callbacksPolicies, std::vector< SendingPolicy > const &sendingPolicies, std::vector< ConfigParamSpec > const &currentWorkflowOptions, std::vector< ConfigParamSpec > const &detectedParams, o2::framework::ConfigContext &configContext)
void overridePipeline(ConfigContext &ctx, WorkflowSpec &workflow)
void enableSignposts(std::string const &signpostsToEnable)
void spawnDevice(uv_loop_t *loop, DeviceRef ref, std::vector< DeviceSpec > const &specs, DriverInfo &driverInfo, std::vector< DeviceControl > &, std::vector< DeviceExecution > &executions, std::vector< DeviceInfo > &deviceInfos, std::vector< DataProcessingStates > &allStates, ServiceRegistryRef serviceRegistry, boost::program_options::variables_map &varmap, std::vector< DeviceStdioContext > &childFds, unsigned parentCPU, unsigned parentNode)
void killChildren(std::vector< DeviceInfo > &infos, int sig)
void ws_connect_callback(uv_stream_t *server, int status)
A callback for the rest engine.
std::vector< DataProcessingStates > DataProcessingStatesInfos
void createPipes(int *pipes)
void doDPLException(o2::framework::RuntimeErrorRef &ref, char const *)
std::vector< DeviceExecution > DeviceExecutions
void overrideAll(o2::framework::ConfigContext &ctx, std::vector< o2::framework::DataProcessorSpec > &workflow)
std::vector< DeviceMetricsInfo > gDeviceMetricsInfos
void force_exit_callback(uv_timer_s *ctx)
std::string debugTopoInfo(std::vector< DataProcessorSpec > const &specs, std::vector< TopoIndexInfo > const &infos, std::vector< std::pair< int, int > > const &edges)
void overrideCloning(ConfigContext &ctx, WorkflowSpec &workflow)
void doBoostException(boost::exception &e, const char *)
bool processSigChild(DeviceInfos &infos, DeviceSpecs &specs)
void checkNonResiliency(std::vector< DataProcessorSpec > const &specs, std::vector< std::pair< int, int > > const &edges)
std::vector< std::regex > getDumpableMetrics()
void stream_config(uv_work_t *req)
std::vector< DataProcessorSpec > DataProcessorSpecs
void dumpRunSummary(DriverServerContext &context, DriverInfo const &driverInfo, DeviceInfos const &infos, DeviceSpecs const &specs)
void conflicting_options(const boost::program_options::variables_map &vm, const std::string &opt1, const std::string &opt2)
Helper to to detect conflicting options.
int doChild(int argc, char **argv, ServiceRegistry &serviceRegistry, RunningWorkflowInfo const &runningWorkflow, RunningDeviceRef ref, DriverConfig const &driverConfig, ProcessingPolicies processingPolicies, std::string const &defaultDriverClient, uv_loop_t *loop)
void doDefaultWorkflowTerminationHook()
bool checkIfCanExit(std::vector< DeviceInfo > const &infos)
volatile sig_atomic_t sigchld_requested
bool isOutputToPipe()
void handleSignals()
std::vector< DeviceSpec > DeviceSpecs
std::vector< DataProcessorInfo > DataProcessorInfos
volatile sig_atomic_t forceful_exit
bool areAllChildrenGone(std::vector< DeviceInfo > &infos)
Check the state of the children.
std::vector< DeviceControl > DeviceControls
volatile sig_atomic_t double_sigint
void close_websocket(uv_handle_t *handle)
void handleChildrenStdio(DriverServerContext *serverContext, std::string const &forwardedStdin, std::vector< DeviceStdioContext > &childFds, std::vector< uv_poll_t * > &handles)
char * getIdString(int argc, char **argv)
bool isInputConfig()
std::unique_ptr< o2::framework::ServiceRegistry > createRegistry()
void log_callback(uv_poll_t *handle, int status, int events)
void processChildrenOutput(uv_loop_t *loop, DriverInfo &driverInfo, DeviceInfos &infos, DeviceSpecs const &specs, DeviceControls &controls)
volatile sig_atomic_t graceful_exit
void single_step_callback(uv_timer_s *ctx)
Force single stepping of the children.
bpo::options_description gHiddenDeviceOptions("Hidden child options")
void doUnknownException(std::string const &s, char const *)
o2::framework::ConfigContext createConfigContext(std::unique_ptr< ConfigParamRegistry > &workflowOptionsRegistry, o2::framework::ServiceRegistry &configRegistry, std::vector< o2::framework::ConfigParamSpec > &workflowOptions, std::vector< o2::framework::ConfigParamSpec > &extraOptions, int argc, char **argv)
int callMain(int argc, char **argv, int(*mainNoCatch)(int, char **))
void handle_crash(int sig)
void dumpMetricsCallback(uv_timer_t *handle)
void cleanupSHM(std::string const &uniqueWorkflowId)
Helper to invoke shared memory cleanup.
void initialiseDriverControl(bpo::variables_map const &varmap, DriverInfo &driverInfo, DriverControl &control)
Helper function to initialise the controller from the command line options.
void gui_callback(uv_timer_s *ctx)
std::vector< DeviceInfo > DeviceInfos
void printHelp(bpo::variables_map const &varmap, bpo::options_description const &executorOptions, std::vector< DataProcessorSpec > const &physicalWorkflow, std::vector< ConfigParamSpec > const &currentWorkflowOptions)
void spawnRemoteDevice(uv_loop_t *loop, std::string const &, DeviceSpec const &spec, DeviceControl &, DeviceExecution &, DeviceInfos &deviceInfos, DataProcessingStatesInfos &allStates)
void websocket_callback(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf)
int mainNoCatch(int argc, char **argv)
DriverServerContext * serverContext
std::vector< ConfigParamSpec > options
std::vector< std::string > args
static ServiceSpec arrowBackendSpec()
static void demangled_backtrace_symbols(void **backtrace, unsigned int total, int fd)
static std::vector< ServiceSpec > defaultServices()
static std::vector< ComputingResource > parseResources(std::string const &resourceString)
static std::vector< ConfigParamSpec > discover(ConfigParamRegistry &, int, char **)
static boost::program_options::options_description prepareOptionDescriptions(ContainerType const &workflow, std::vector< ConfigParamSpec > const &currentWorkflowOptions, options_description vetos=options_description(), std::string mode="full")
populate boost program options for a complete workflow
static void populateBoostProgramOptions(options_description &options, const std::vector< ConfigParamSpec > &specs, options_description vetos=options_description())
static void dumpDeviceSpec2DDS(std::ostream &out, DriverMode mode, std::string const &workflowSuffix, std::vector< DataProcessorSpec > const &workflow, std::vector< DataProcessorInfo > const &metadata, std::vector< DeviceSpec > const &specs, std::vector< DeviceExecution > const &executions, CommandInfo const &commandInfo)
static void preExitCallbacks(std::vector< ServiceExitHandle >, ServiceRegistryRef)
Invoke callback to be executed on exit, in reverse order.
A label that can be associated to a DataProcessorSpec.
static std::string describe(InputSpec const &spec)
Plugin interface for DPL GUIs.
Definition DebugGUI.h:30
virtual void * initGUI(char const *windowTitle, ServiceRegistry &registry)=0
virtual std::function< void(void)> getGUIDebugger(std::vector< o2::framework::DeviceInfo > const &infos, std::vector< o2::framework::DeviceSpec > const &devices, std::vector< o2::framework::DataProcessingStates > const &allStates, std::vector< o2::framework::DataProcessorInfo > const &metadata, std::vector< o2::framework::DeviceMetricsInfo > const &metricsInfos, o2::framework::DriverInfo const &driverInfo, std::vector< o2::framework::DeviceControl > &controls, o2::framework::DriverControl &driverControl)=0
static DeploymentMode deploymentMode()
static unsigned int pipelineLength()
get max number of timeslices in the queue
static std::unique_ptr< ConfigParamStore > getConfiguration(ServiceRegistryRef registry, const char *name, std::vector< ConfigParamSpec > const &options)
char logFilter[MAX_USER_FILTER_SIZE]
Lines in the log should match this to be displayed.
bool quiet
wether we should be capturing device output.
std::vector< std::string > history
Definition DeviceInfo.h:56
bool active
Whether the device is active (running) or not.
Definition DeviceInfo.h:65
size_t historySize
The size of the history circular buffer.
Definition DeviceInfo.h:45
std::vector< LogParsingHelpers::LogLevel > historyLevel
Definition DeviceInfo.h:59
pid_t pid
The pid of the device associated to this device.
Definition DeviceInfo.h:36
std::string unprinted
An unterminated string which is not ready to be printed yet.
Definition DeviceInfo.h:63
LogParsingHelpers::LogLevel logLevel
The minimum log level for log messages sent/displayed by this device.
Definition DeviceInfo.h:49
LogParsingHelpers::LogLevel maxLogLevel
The maximum log level ever seen by this device.
Definition DeviceInfo.h:47
size_t historyPos
The position inside the history circular buffer of this device.
Definition DeviceInfo.h:43
std::string firstSevereError
Definition DeviceInfo.h:60
static void validate(WorkflowSpec const &workflow)
static boost::program_options::options_description getForwardedDeviceOptions()
define the options which are forwarded to every child
static std::string reworkTimeslicePlaceholder(std::string const &str, DeviceSpec const &spec)
static void prepareArguments(bool defaultQuiet, bool defaultStopped, bool intereactive, unsigned short driverPort, DriverConfig const &driverConfig, std::vector< DataProcessorInfo > const &processorInfos, std::vector< DeviceSpec > const &deviceSpecs, std::vector< DeviceExecution > &deviceExecutions, std::vector< DeviceControl > &deviceControls, std::vector< ConfigParamSpec > const &detectedOptions, std::string const &uniqueWorkflowId)
static void reworkShmSegmentSize(std::vector< DataProcessorInfo > &infos)
static void reworkHomogeneousOption(std::vector< DataProcessorInfo > &infos, char const *name, char const *defaultValue)
static void dataProcessorSpecs2DeviceSpecs(const WorkflowSpec &workflow, std::vector< ChannelConfigurationPolicy > const &channelPolicies, std::vector< CompletionPolicy > const &completionPolicies, std::vector< DispatchPolicy > const &dispatchPolicies, std::vector< ResourcePolicy > const &resourcePolicies, std::vector< CallbacksPolicy > const &callbacksPolicies, std::vector< SendingPolicy > const &sendingPolicy, std::vector< ForwardingPolicy > const &forwardingPolicies, std::vector< DeviceSpec > &devices, ResourceManager &resourceManager, std::string const &uniqueWorkflowId, ConfigContext const &configContext, bool optimizeTopology=false, unsigned short resourcesMonitoringInterval=0, std::string const &channelPrefix="", OverrideServiceSpecs const &overrideServices={})
std::vector< OutputRoute > outputs
Definition DeviceSpec.h:63
std::string id
The id of the device, including time-pipelining and suffix.
Definition DeviceSpec.h:52
static int parseTracingFlags(std::string const &events)
std::vector< DeviceControl > & controls
bool batch
Whether the driver was started in batch mode or not.
std::vector< DriverState > forcedTransitions
std::vector< Callback > callbacks
DriverControlState state
Current state of the state machine player.
std::vector< DeviceSpec > * specs
std::vector< ServiceSummaryHandling > * summaryCallbacks
std::vector< DeviceMetricsInfo > * metrics
std::vector< DeviceInfo > * infos
static std::vector< ForwardingPolicy > createDefaultPolicies()
static void dumpDeviceSpec2Graphviz(std::ostream &, const Devices &specs)
Helper to dump a set of devices as a graphviz file.
static void dumpDataProcessorSpec2Graphviz(std::ostream &, const WorkflowSpec &specs, std::vector< std::pair< int, int > > const &edges={})
Helper to dump a workflow as a graphviz file.
std::function< void(void)> callback
LogLevel
Possible log levels for device log entries.
static LogLevel parseTokenLevel(std::string_view const s)
static void dumpDeviceSpec2Mermaid(std::ostream &, const Devices &specs)
Helper to dump a set of devices as a mermaid file.
Temporary struct to hold a metric after it has been parsed.
static bool dumpMetricsToJSON(std::vector< DeviceMetricsInfo > const &metrics, DeviceMetricsInfo const &driverMetrics, std::vector< DeviceSpec > const &specs, std::vector< std::regex > const &metricsToDump) noexcept
static bool isResourcesMonitoringEnabled(unsigned short interval) noexcept
Information about the running workflow.
void declareService(ServiceSpec const &spec, DeviceState &state, fair::mq::ProgOptions &options, ServiceRegistry::Salt salt=ServiceRegistry::globalDeviceSalt())
static OverrideServiceSpecs parseOverrides(char const *overrideString)
static ServiceSpecs filterDisabled(ServiceSpecs originals, OverrideServiceSpecs const &overrides)
static std::function< int64_t(int64_t base, int64_t offset)> defaultCPUTimeConfigurator(uv_loop_t *loop)
static std::function< void(int64_t &base, int64_t &offset)> defaultRealtimeBaseConfigurator(uint64_t offset, uv_loop_t *loop)
Helper struct to keep track of the results of the topological sort.
static auto buildEdges(WorkflowSpec &physicalWorkflow) -> std::vector< std::pair< int, int > >
static void adjustTopology(WorkflowSpec &workflow, ConfigContext const &ctx)
static void injectServiceDevices(WorkflowSpec &workflow, ConfigContext &ctx)
static WorkflowParsingState verifyWorkflow(const WorkflowSpec &workflow)
static std::vector< TopoIndexInfo > topologicalSort(size_t nodeCount, int const *edgeIn, int const *edgeOut, size_t byteStride, size_t edgesCount)
static void dump(std::ostream &o, std::vector< DataProcessorSpec > const &workflow, std::vector< DataProcessorInfo > const &metadata, CommandInfo const &commandInfo)
static bool import(std::istream &s, std::vector< DataProcessorSpec > &workflow, std::vector< DataProcessorInfo > &metadata, CommandInfo &command)
uint16_t de
LOG(info)<< "Compressed in "<< sw.CpuTime()<< " s"
std::vector< ChannelData > channels
uint64_t const void const *restrict const msg
Definition x9.h:153