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 std::time_t result = std::time(nullptr);
2207 char buffer[32];
2208 std::strncpy(buffer, std::ctime(&result), 26);
2209 O2_SIGNPOST_EVENT_EMIT_INFO(driver, sid, "mainloop", "Quit requested at %{public}s", buffer);
2210 guiQuitRequested = true;
2211 // We send SIGCONT to make sure stopped children are resumed
2212 killChildren(infos, SIGCONT);
2213 // We send SIGTERM to make sure we do the STOP transition in FairMQ
2214 killChildren(infos, SIGTERM);
2215 // We have a timer to send SIGUSR1 to make sure we advance all devices
2216 // in a timely manner.
2217 force_step_timer.data = &infos;
2218 uv_timer_start(&force_step_timer, single_step_callback, 0, 300);
2219 driverInfo.states.push_back(DriverState::HANDLE_CHILDREN);
2220 break;
2221 }
2222 case DriverState::HANDLE_CHILDREN: {
2223 // Run any pending libUV event loop, block if
2224 // any, so that we do not consume CPU time when the driver is
2225 // idle.
2226 uv_run(loop, once ? UV_RUN_ONCE : UV_RUN_NOWAIT);
2227 once = true;
2228 // I allow queueing of more sigchld only when
2229 // I process the previous call
2230 if (forceful_exit == true) {
2231 static bool forcefulExitMessage = true;
2232 if (forcefulExitMessage) {
2233 LOG(info) << "Forceful exit requested.";
2234 forcefulExitMessage = false;
2235 }
2236 killChildren(infos, SIGCONT);
2237 killChildren(infos, SIGKILL);
2238 }
2239 sigchld_requested = false;
2240 driverInfo.sigchldRequested = false;
2241 processChildrenOutput(loop, driverInfo, infos, runningWorkflow.devices, controls);
2242 hasError = processSigChild(infos, runningWorkflow.devices);
2243 allChildrenGone = areAllChildrenGone(infos);
2244 bool canExit = checkIfCanExit(infos);
2245 bool supposedToQuit = (guiQuitRequested || canExit || graceful_exit);
2246
2247 if (allChildrenGone && (supposedToQuit || driverInfo.processingPolicies.termination == TerminationPolicy::QUIT)) {
2248 // We move to the exit, regardless of where we were
2249 driverInfo.states.resize(0);
2250 driverInfo.states.push_back(DriverState::EXIT);
2251 } else if (hasError && driverInfo.processingPolicies.error == TerminationPolicy::QUIT && !supposedToQuit) {
2252 graceful_exit = 1;
2253 force_exit_timer.data = &infos;
2254 static bool forceful_timer_started = false;
2255 if (forceful_timer_started == false) {
2256 forceful_timer_started = true;
2257 uv_timer_start(&force_exit_timer, force_exit_callback, 15000, 3000);
2258 }
2259 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
2260 } else if (allChildrenGone == false && supposedToQuit) {
2261 driverInfo.states.push_back(DriverState::HANDLE_CHILDREN);
2262 } else {
2263 }
2264 } break;
2265 case DriverState::EXIT: {
2266 if (ResourcesMonitoringHelper::isResourcesMonitoringEnabled(driverInfo.resourcesMonitoringInterval)) {
2267 if (driverInfo.resourcesMonitoringDumpInterval) {
2268 uv_timer_stop(&metricDumpTimer);
2269 }
2270 LOG(info) << "Dumping performance metrics to performanceMetrics.json file";
2271 dumpMetricsCallback(&metricDumpTimer);
2272 }
2273 dumpRunSummary(serverContext, driverInfo, infos, runningWorkflow.devices);
2274 // This is a clean exit. Before we do so, if required,
2275 // we dump the configuration of all the devices so that
2276 // we can reuse it. Notice we do not dump anything if
2277 // the workflow was not really run.
2278 // NOTE: is this really what we want? should we run
2279 // SCHEDULE and dump the full configuration as well?
2280 if (infos.empty()) {
2281 return 0;
2282 }
2283 boost::property_tree::ptree finalConfig;
2284 assert(infos.size() == runningWorkflow.devices.size());
2285 for (size_t di = 0; di < infos.size(); ++di) {
2286 auto& info = infos[di];
2287 auto& spec = runningWorkflow.devices[di];
2288 finalConfig.put_child(spec.name, info.currentConfig);
2289 }
2290 LOG(info) << "Dumping used configuration in dpl-config.json";
2291
2292 std::ofstream outDPLConfigFile("dpl-config.json", std::ios::out);
2293 if (outDPLConfigFile.is_open()) {
2294 boost::property_tree::write_json(outDPLConfigFile, finalConfig);
2295 } else {
2296 LOGP(warning, "Could not write out final configuration file. Read only run folder?");
2297 }
2298 if (driverInfo.noSHMCleanup) {
2299 LOGP(warning, "Not cleaning up shared memory.");
2300 } else {
2301 cleanupSHM(driverInfo.uniqueWorkflowId);
2302 }
2303 return calculateExitCode(driverInfo, runningWorkflow.devices, infos);
2304 }
2305 case DriverState::PERFORM_CALLBACKS:
2306 for (auto& callback : driverControl.callbacks) {
2307 callback(workflow, runningWorkflow.devices, deviceExecutions, dataProcessorInfos, commandInfo);
2308 }
2309 driverControl.callbacks.clear();
2310 break;
2311 default:
2312 LOG(error) << "Driver transitioned in an unknown state("
2313 << "current: " << (int)current
2314 << ", previous: " << (int)previous
2315 << "). Shutting down.";
2316 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
2317 }
2318 }
2319 O2_SIGNPOST_END(driver, sid, "driver", "End driver loop");
2320}
2321
2322// Print help
2323void printHelp(bpo::variables_map const& varmap,
2324 bpo::options_description const& executorOptions,
2325 std::vector<DataProcessorSpec> const& physicalWorkflow,
2326 std::vector<ConfigParamSpec> const& currentWorkflowOptions)
2327{
2328 auto mode = varmap["help"].as<std::string>();
2329 bpo::options_description helpOptions;
2330 if (mode == "full" || mode == "short" || mode == "executor") {
2331 helpOptions.add(executorOptions);
2332 }
2333 // this time no veto is applied, so all the options are added for printout
2334 if (mode == "executor") {
2335 // nothing more
2336 } else if (mode == "workflow") {
2337 // executor options and workflow options, skip the actual workflow
2338 o2::framework::WorkflowSpec emptyWorkflow;
2339 helpOptions.add(ConfigParamsHelper::prepareOptionDescriptions(emptyWorkflow, currentWorkflowOptions));
2340 } else if (mode == "full" || mode == "short") {
2341 helpOptions.add(ConfigParamsHelper::prepareOptionDescriptions(physicalWorkflow, currentWorkflowOptions,
2342 bpo::options_description(),
2343 mode));
2344 } else {
2345 helpOptions.add(ConfigParamsHelper::prepareOptionDescriptions(physicalWorkflow, {},
2346 bpo::options_description(),
2347 mode));
2348 }
2349 if (helpOptions.options().size() == 0) {
2350 // the specified argument is invalid, add at leat the executor options
2351 mode += " is an invalid argument, please use correct argument for";
2352 helpOptions.add(executorOptions);
2353 }
2354 std::cout << "ALICE O2 DPL workflow driver" //
2355 << " (" << mode << " help)" << std::endl //
2356 << helpOptions << std::endl; //
2357}
2358
2359// Helper to find out if stdout is actually attached to a pipe.
2361{
2362 struct stat s;
2363 fstat(STDOUT_FILENO, &s);
2364 return ((s.st_mode & S_IFIFO) != 0);
2365}
2366
2368{
2369 struct stat s;
2370 int r = fstat(STDIN_FILENO, &s);
2371 // If stdin cannot be statted, we assume the shell is some sort of
2372 // non-interactive container thing
2373 if (r < 0) {
2374 return false;
2375 }
2376 // If stdin is a pipe or a file, we try to fetch configuration from there
2377 return ((s.st_mode & S_IFIFO) != 0 || (s.st_mode & S_IFREG) != 0);
2378}
2379
2381{
2382 struct CloningSpec {
2383 std::string templateMatcher;
2384 std::string cloneName;
2385 };
2386 auto s = ctx.options().get<std::string>("clone");
2387 std::vector<CloningSpec> specs;
2388 std::string delimiter = ",";
2389
2390 while (s.empty() == false) {
2391 auto newPos = s.find(delimiter);
2392 auto token = s.substr(0, newPos);
2393 auto split = token.find(":");
2394 if (split == std::string::npos) {
2395 throw std::runtime_error("bad clone definition. Syntax <template-processor>:<clone-name>");
2396 }
2397 auto key = token.substr(0, split);
2398 token.erase(0, split + 1);
2399 size_t error;
2400 std::string value = "";
2401 try {
2402 auto numValue = std::stoll(token, &error, 10);
2403 if (token[error] != '\0') {
2404 throw std::runtime_error("bad name for clone:" + token);
2405 }
2406 value = key + "_c" + std::to_string(numValue);
2407 } catch (std::invalid_argument& e) {
2408 value = token;
2409 }
2410 specs.push_back({key, value});
2411 s.erase(0, newPos + (newPos == std::string::npos ? 0 : 1));
2412 }
2413 if (s.empty() == false && specs.empty() == true) {
2414 throw std::runtime_error("bad pipeline definition. Syntax <processor>:<pipeline>");
2415 }
2416
2417 std::vector<DataProcessorSpec> extraSpecs;
2418 for (auto& spec : specs) {
2419 for (auto& processor : workflow) {
2420 if (processor.name == spec.templateMatcher) {
2421 auto clone = processor;
2422 clone.name = spec.cloneName;
2423 extraSpecs.push_back(clone);
2424 }
2425 }
2426 }
2427 workflow.insert(workflow.end(), extraSpecs.begin(), extraSpecs.end());
2428}
2429
2431{
2432 struct PipelineSpec {
2433 std::string matcher;
2434 int64_t pipeline;
2435 };
2436 auto s = ctx.options().get<std::string>("pipeline");
2437 std::vector<PipelineSpec> specs;
2438 std::string delimiter = ",";
2439
2440 while (s.empty() == false) {
2441 auto newPos = s.find(delimiter);
2442 auto token = s.substr(0, newPos);
2443 auto split = token.find(":");
2444 if (split == std::string::npos) {
2445 throw std::runtime_error("bad pipeline definition. Syntax <processor>:<pipeline>");
2446 }
2447 auto key = token.substr(0, split);
2448 token.erase(0, split + 1);
2449 size_t error;
2450 auto value = std::stoll(token, &error, 10);
2451 if (token[error] != '\0') {
2452 throw std::runtime_error("Bad pipeline definition. Expecting integer");
2453 }
2454 specs.push_back({key, value});
2455 s.erase(0, newPos + (newPos == std::string::npos ? 0 : 1));
2456 }
2457 if (s.empty() == false && specs.empty() == true) {
2458 throw std::runtime_error("bad pipeline definition. Syntax <processor>:<pipeline>");
2459 }
2460
2461 for (auto& spec : specs) {
2462 for (auto& processor : workflow) {
2463 if (processor.name == spec.matcher) {
2464 processor.maxInputTimeslices = spec.pipeline;
2465 }
2466 }
2467 }
2468}
2469
2471{
2472 struct LabelsSpec {
2473 std::string_view matcher;
2474 std::vector<std::string> labels;
2475 };
2476 std::vector<LabelsSpec> specs;
2477
2478 auto labelsString = ctx.options().get<std::string>("labels");
2479 if (labelsString.empty()) {
2480 return;
2481 }
2482 std::string_view sv{labelsString};
2483
2484 size_t specStart = 0;
2485 size_t specEnd = 0;
2486 constexpr char specDelim = ',';
2487 constexpr char labelDelim = ':';
2488 do {
2489 specEnd = sv.find(specDelim, specStart);
2490 auto token = sv.substr(specStart, specEnd == std::string_view::npos ? std::string_view::npos : specEnd - specStart);
2491 if (token.empty()) {
2492 throw std::runtime_error("bad labels definition. Syntax <processor>:<label>[:<label>][,<processor>:<label>[:<label>]");
2493 }
2494
2495 size_t labelDelimPos = token.find(labelDelim);
2496 if (labelDelimPos == 0 || labelDelimPos == std::string_view::npos) {
2497 throw std::runtime_error("bad labels definition. Syntax <processor>:<label>[:<label>][,<processor>:<label>[:<label>]");
2498 }
2499 LabelsSpec spec{.matcher = token.substr(0, labelDelimPos), .labels = {}};
2500
2501 size_t labelEnd = labelDelimPos + 1;
2502 do {
2503 size_t labelStart = labelDelimPos + 1;
2504 labelEnd = token.find(labelDelim, labelStart);
2505 auto label = labelEnd == std::string_view::npos ? token.substr(labelStart) : token.substr(labelStart, labelEnd - labelStart);
2506 if (label.empty()) {
2507 throw std::runtime_error("bad labels definition. Syntax <processor>:<label>[:<label>][,<processor>:<label>[:<label>]");
2508 }
2509 spec.labels.emplace_back(label);
2510 labelDelimPos = labelEnd;
2511 } while (labelEnd != std::string_view::npos);
2512
2513 specs.push_back(spec);
2514 specStart = specEnd + 1;
2515 } while (specEnd != std::string_view::npos);
2516
2517 if (labelsString.empty() == false && specs.empty() == true) {
2518 throw std::runtime_error("bad labels definition. Syntax <processor>:<label>[:<label>][,<processor>:<label>[:<label>]");
2519 }
2520
2521 for (auto& spec : specs) {
2522 for (auto& processor : workflow) {
2523 if (processor.name == spec.matcher) {
2524 for (const auto& label : spec.labels) {
2525 if (std::find_if(processor.labels.begin(), processor.labels.end(),
2526 [label](const auto& procLabel) { return procLabel.value == label; }) == processor.labels.end()) {
2527 processor.labels.push_back({label});
2528 }
2529 }
2530 }
2531 }
2532 }
2533}
2534
2536void initialiseDriverControl(bpo::variables_map const& varmap,
2537 DriverInfo& driverInfo,
2538 DriverControl& control)
2539{
2540 // Control is initialised outside the main loop because
2541 // command line options are really affecting control.
2542 control.defaultQuiet = varmap["quiet"].as<bool>();
2543 control.defaultStopped = varmap["stop"].as<bool>();
2544
2545 if (varmap["single-step"].as<bool>()) {
2546 control.state = DriverControlState::STEP;
2547 } else {
2548 control.state = DriverControlState::PLAY;
2549 }
2550
2551 if (varmap["graphviz"].as<bool>()) {
2552 // Dump a graphviz representation of what I will do.
2553 control.callbacks = {[](WorkflowSpec const&,
2554 DeviceSpecs const& specs,
2555 DeviceExecutions const&,
2557 CommandInfo const&) {
2559 }};
2560 control.forcedTransitions = {
2561 DriverState::EXIT, //
2562 DriverState::PERFORM_CALLBACKS, //
2563 DriverState::MERGE_CONFIGS, //
2564 DriverState::IMPORT_CURRENT_WORKFLOW, //
2565 DriverState::MATERIALISE_WORKFLOW //
2566 };
2567 } else if (!varmap["dds"].as<std::string>().empty()) {
2568 // Dump a DDS representation of what I will do.
2569 // Notice that compared to DDS we need to schedule things,
2570 // because DDS needs to be able to have actual Executions in
2571 // order to provide a correct configuration.
2572 control.callbacks = {[filename = varmap["dds"].as<std::string>(),
2573 workflowSuffix = varmap["dds-workflow-suffix"],
2574 driverMode = driverInfo.mode](WorkflowSpec const& workflow,
2575 DeviceSpecs const& specs,
2576 DeviceExecutions const& executions,
2577 DataProcessorInfos& dataProcessorInfos,
2578 CommandInfo const& commandInfo) {
2579 if (filename == "-") {
2580 DDSConfigHelpers::dumpDeviceSpec2DDS(std::cout, driverMode, workflowSuffix.as<std::string>(), workflow, dataProcessorInfos, specs, executions, commandInfo);
2581 } else {
2582 std::ofstream out(filename);
2583 DDSConfigHelpers::dumpDeviceSpec2DDS(out, driverMode, workflowSuffix.as<std::string>(), workflow, dataProcessorInfos, specs, executions, commandInfo);
2584 }
2585 }};
2586 control.forcedTransitions = {
2587 DriverState::EXIT, //
2588 DriverState::PERFORM_CALLBACKS, //
2589 DriverState::MERGE_CONFIGS, //
2590 DriverState::IMPORT_CURRENT_WORKFLOW, //
2591 DriverState::MATERIALISE_WORKFLOW //
2592 };
2593 } else if (!varmap["o2-control"].as<std::string>().empty() or !varmap["mermaid"].as<std::string>().empty()) {
2594 // Dump the workflow in o2-control and/or mermaid format
2595 control.callbacks = {[filename = varmap["mermaid"].as<std::string>(),
2596 workflowName = varmap["o2-control"].as<std::string>()](WorkflowSpec const&,
2597 DeviceSpecs const& specs,
2598 DeviceExecutions const& executions,
2600 CommandInfo const& commandInfo) {
2601 if (!workflowName.empty()) {
2602 dumpDeviceSpec2O2Control(workflowName, specs, executions, commandInfo);
2603 }
2604 if (!filename.empty()) {
2605 if (filename == "-") {
2607 } else {
2608 std::ofstream output(filename);
2610 }
2611 }
2612 }};
2613 control.forcedTransitions = {
2614 DriverState::EXIT, //
2615 DriverState::PERFORM_CALLBACKS, //
2616 DriverState::MERGE_CONFIGS, //
2617 DriverState::IMPORT_CURRENT_WORKFLOW, //
2618 DriverState::MATERIALISE_WORKFLOW //
2619 };
2620
2621 } else if (varmap.count("id")) {
2622 // Add our own stacktrace dumping
2623 if (getenv("O2_NO_CATCHALL_EXCEPTIONS") != nullptr && strcmp(getenv("O2_NO_CATCHALL_EXCEPTIONS"), "0") != 0) {
2624 LOGP(info, "Not instrumenting crash signals because O2_NO_CATCHALL_EXCEPTIONS is set");
2625 gEnv->SetValue("Root.Stacktrace", "no");
2626 gSystem->ResetSignal(kSigSegmentationViolation, kTRUE);
2627 rlimit limit;
2628 if (getrlimit(RLIMIT_CORE, &limit) == 0) {
2629 LOGP(info, "Core limit: {} {}", limit.rlim_cur, limit.rlim_max);
2630 }
2631 }
2632 if (varmap["stacktrace-on-signal"].as<std::string>() == "simple" && (getenv("O2_NO_CATCHALL_EXCEPTIONS") == nullptr || strcmp(getenv("O2_NO_CATCHALL_EXCEPTIONS"), "0") == 0)) {
2633 LOGP(info, "Instrumenting crash signals");
2634 signal(SIGSEGV, handle_crash);
2635 signal(SIGABRT, handle_crash);
2636 signal(SIGBUS, handle_crash);
2637 signal(SIGILL, handle_crash);
2638 signal(SIGFPE, handle_crash);
2639 }
2640 // FIXME: for the time being each child needs to recalculate the workflow,
2641 // so that it can understand what it needs to do. This is obviously
2642 // a bad idea. In the future we should have the client be pushed
2643 // it's own configuration by the driver.
2644 control.forcedTransitions = {
2645 DriverState::DO_CHILD, //
2646 DriverState::BIND_GUI_PORT, //
2647 DriverState::MERGE_CONFIGS, //
2648 DriverState::IMPORT_CURRENT_WORKFLOW, //
2649 DriverState::MATERIALISE_WORKFLOW //
2650 };
2651 } else if ((varmap["dump-workflow"].as<bool>() == true) || (varmap["run"].as<bool>() == false && varmap.count("id") == 0 && isOutputToPipe())) {
2652 control.callbacks = {[filename = varmap["dump-workflow-file"].as<std::string>()](WorkflowSpec const& workflow,
2653 DeviceSpecs const&,
2654 DeviceExecutions const&,
2655 DataProcessorInfos& dataProcessorInfos,
2656 CommandInfo const& commandInfo) {
2657 if (filename == "-") {
2658 WorkflowSerializationHelpers::dump(std::cout, workflow, dataProcessorInfos, commandInfo);
2659 // FIXME: this is to avoid trailing garbage..
2660 exit(0);
2661 } else {
2662 std::ofstream output(filename);
2663 WorkflowSerializationHelpers::dump(output, workflow, dataProcessorInfos, commandInfo);
2664 }
2665 }};
2666 control.forcedTransitions = {
2667 DriverState::EXIT, //
2668 DriverState::PERFORM_CALLBACKS, //
2669 DriverState::MERGE_CONFIGS, //
2670 DriverState::IMPORT_CURRENT_WORKFLOW, //
2671 DriverState::MATERIALISE_WORKFLOW //
2672 };
2673 } else {
2674 // By default we simply start the main loop of the driver.
2675 control.forcedTransitions = {
2676 DriverState::INIT, //
2677 DriverState::BIND_GUI_PORT, //
2678 DriverState::IMPORT_CURRENT_WORKFLOW, //
2679 DriverState::MATERIALISE_WORKFLOW //
2680 };
2681 }
2682}
2683
2685void conflicting_options(const boost::program_options::variables_map& vm,
2686 const std::string& opt1, const std::string& opt2)
2687{
2688 if (vm.count(opt1) && !vm[opt1].defaulted() &&
2689 vm.count(opt2) && !vm[opt2].defaulted()) {
2690 throw std::logic_error(std::string("Conflicting options '") +
2691 opt1 + "' and '" + opt2 + "'.");
2692 }
2693}
2694
2695template <typename T>
2697 std::vector<T>& v,
2698 std::vector<int>& indices)
2699{
2700 using std::swap; // to permit Koenig lookup
2701 for (int i = 0; i < (int)indices.size(); i++) {
2702 auto current = i;
2703 while (i != indices[current]) {
2704 auto next = indices[current];
2705 swap(v[current], v[next]);
2706 indices[current] = current;
2707 current = next;
2708 }
2709 indices[current] = current;
2710 }
2711}
2712
2713// Check if the workflow is resiliant to failures
2714void checkNonResiliency(std::vector<DataProcessorSpec> const& specs,
2715 std::vector<std::pair<int, int>> const& edges)
2716{
2717 auto checkExpendable = [](DataProcessorLabel const& label) {
2718 return label.value == "expendable";
2719 };
2720 auto checkResilient = [](DataProcessorLabel const& label) {
2721 return label.value == "resilient" || label.value == "expendable";
2722 };
2723
2724 for (auto& edge : edges) {
2725 auto& src = specs[edge.first];
2726 auto& dst = specs[edge.second];
2727 if (std::none_of(src.labels.begin(), src.labels.end(), checkExpendable)) {
2728 continue;
2729 }
2730 if (std::any_of(dst.labels.begin(), dst.labels.end(), checkResilient)) {
2731 continue;
2732 }
2733 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.");
2734 }
2735}
2736
2737std::string debugTopoInfo(std::vector<DataProcessorSpec> const& specs,
2738 std::vector<TopoIndexInfo> const& infos,
2739 std::vector<std::pair<int, int>> const& edges)
2740{
2741 std::ostringstream out;
2742
2743 out << "\nTopological info:\n";
2744 for (auto& ti : infos) {
2745 out << specs[ti.index].name << " (index: " << ti.index << ", layer: " << ti.layer << ")\n";
2746 out << " Inputs:\n";
2747 for (auto& ii : specs[ti.index].inputs) {
2748 out << " - " << DataSpecUtils::describe(ii) << "\n";
2749 }
2750 out << "\n Outputs:\n";
2751 for (auto& ii : specs[ti.index].outputs) {
2752 out << " - " << DataSpecUtils::describe(ii) << "\n";
2753 }
2754 }
2755 out << "\nEdges values:\n";
2756 for (auto& e : edges) {
2757 out << specs[e.second].name << " depends on " << specs[e.first].name << "\n";
2758 }
2759 for (auto& d : specs) {
2760 out << "- " << d.name << std::endl;
2761 }
2763 return out.str();
2764}
2765
2766void enableSignposts(std::string const& signpostsToEnable)
2767{
2768 static pid_t pid = getpid();
2769 if (signpostsToEnable.empty() == true) {
2770 auto printAllSignposts = [](char const* name, void* l, void* context) {
2771 auto* log = (_o2_log_t*)l;
2772 LOGP(detail, "Signpost stream {} disabled. Enable it with o2-log -p {} -a {}", name, pid, (void*)&log->stacktrace);
2773 return true;
2774 };
2775 o2_walk_logs(printAllSignposts, nullptr);
2776 return;
2777 }
2778 auto matchingLogEnabler = [](char const* name, void* l, void* context) {
2779 auto* log = (_o2_log_t*)l;
2780 auto* selectedName = (char const*)context;
2781 std::string prefix = "ch.cern.aliceo2.";
2782 auto* last = strchr(selectedName, ':');
2783 int maxDepth = 1;
2784 if (last) {
2785 char* err;
2786 maxDepth = strtol(last + 1, &err, 10);
2787 if (*(last + 1) == '\0' || *err != '\0') {
2788 maxDepth = 1;
2789 }
2790 }
2791
2792 auto fullName = prefix + std::string{selectedName, last ? last - selectedName : strlen(selectedName)};
2793 if (fullName == name) {
2794 LOGP(info, "Enabling signposts for stream \"{}\" with depth {}.", fullName, maxDepth);
2795 _o2_log_set_stacktrace(log, maxDepth);
2796 return false;
2797 } else {
2798 LOGP(info, "Signpost stream \"{}\" disabled. Enable it with o2-log -p {} -a {}", name, pid, (void*)&log->stacktrace);
2799 }
2800 return true;
2801 };
2802 // Split signpostsToEnable by comma using strtok_r
2803 char* saveptr;
2804 char* src = const_cast<char*>(signpostsToEnable.data());
2805 auto* token = strtok_r(src, ",", &saveptr);
2806 while (token) {
2807 o2_walk_logs(matchingLogEnabler, token);
2808 token = strtok_r(nullptr, ",", &saveptr);
2809 }
2810}
2811
2812void overrideAll(o2::framework::ConfigContext& ctx, std::vector<o2::framework::DataProcessorSpec>& workflow)
2813{
2814 overrideCloning(ctx, workflow);
2815 overridePipeline(ctx, workflow);
2816 overrideLabels(ctx, workflow);
2817}
2818
2819o2::framework::ConfigContext createConfigContext(std::unique_ptr<ConfigParamRegistry>& workflowOptionsRegistry,
2820 o2::framework::ServiceRegistry& configRegistry,
2821 std::vector<o2::framework::ConfigParamSpec>& workflowOptions,
2822 std::vector<o2::framework::ConfigParamSpec>& extraOptions, int argc, char** argv)
2823{
2824 std::vector<std::unique_ptr<o2::framework::ParamRetriever>> retrievers;
2825 std::unique_ptr<o2::framework::ParamRetriever> retriever{new o2::framework::BoostOptionsRetriever(true, argc, argv)};
2826 retrievers.emplace_back(std::move(retriever));
2827 auto workflowOptionsStore = std::make_unique<o2::framework::ConfigParamStore>(workflowOptions, std::move(retrievers));
2828 workflowOptionsStore->preload();
2829 workflowOptionsStore->activate();
2830 workflowOptionsRegistry = std::make_unique<ConfigParamRegistry>(std::move(workflowOptionsStore));
2831 extraOptions = o2::framework::ConfigParamDiscovery::discover(*workflowOptionsRegistry, argc, argv);
2832 for (auto& extra : extraOptions) {
2833 workflowOptions.push_back(extra);
2834 }
2835
2836 return o2::framework::ConfigContext(*workflowOptionsRegistry, o2::framework::ServiceRegistryRef{configRegistry}, argc, argv);
2837}
2838
2839std::unique_ptr<o2::framework::ServiceRegistry> createRegistry()
2840{
2841 return std::make_unique<o2::framework::ServiceRegistry>();
2842}
2843
2844// This is a toy executor for the workflow spec
2845// What it needs to do is:
2846//
2847// - Print the properties of each DataProcessorSpec
2848// - Fork one process per DataProcessorSpec
2849// - Parent -> wait for all the children to complete (eventually
2850// killing them all on ctrl-c).
2851// - Child, pick the data-processor ID and start a O2DataProcessorDevice for
2852// each DataProcessorSpec
2853int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& workflow,
2854 std::vector<ChannelConfigurationPolicy> const& channelPolicies,
2855 std::vector<CompletionPolicy> const& completionPolicies,
2856 std::vector<DispatchPolicy> const& dispatchPolicies,
2857 std::vector<ResourcePolicy> const& resourcePolicies,
2858 std::vector<CallbacksPolicy> const& callbacksPolicies,
2859 std::vector<SendingPolicy> const& sendingPolicies,
2860 std::vector<ConfigParamSpec> const& currentWorkflowOptions,
2861 std::vector<ConfigParamSpec> const& detectedParams,
2862 o2::framework::ConfigContext& configContext)
2863{
2864 // Peek very early in the driver options and look for
2865 // signposts, so the we can enable it without going through the whole dance
2866 if (getenv("DPL_DRIVER_SIGNPOSTS")) {
2867 enableSignposts(getenv("DPL_DRIVER_SIGNPOSTS"));
2868 }
2869
2870 std::vector<std::string> currentArgs;
2871 std::vector<PluginInfo> plugins;
2872 std::vector<ForwardingPolicy> forwardingPolicies = ForwardingPolicy::createDefaultPolicies();
2873
2874 for (int ai = 1; ai < argc; ++ai) {
2875 currentArgs.emplace_back(argv[ai]);
2876 }
2877
2878 WorkflowInfo currentWorkflow{
2879 argv[0],
2880 currentArgs,
2881 currentWorkflowOptions};
2882
2883 ProcessingPolicies processingPolicies;
2884 enum LogParsingHelpers::LogLevel minFailureLevel;
2885 bpo::options_description executorOptions("Executor options");
2886 const char* helpDescription = "print help: short, full, executor, or processor name";
2887 enum DriverMode driverMode;
2888 executorOptions.add_options() //
2889 ("help,h", bpo::value<std::string>()->implicit_value("short"), helpDescription) // //
2890 ("quiet,q", bpo::value<bool>()->zero_tokens()->default_value(false), "quiet operation") // //
2891 ("stop,s", bpo::value<bool>()->zero_tokens()->default_value(false), "stop before device start") // //
2892 ("single-step", bpo::value<bool>()->zero_tokens()->default_value(false), "start in single step mode") // //
2893 ("batch,b", bpo::value<std::vector<std::string>>()->zero_tokens()->composing(), "batch processing mode") // //
2894 ("no-batch", bpo::value<bool>()->zero_tokens(), "force gui processing mode") // //
2895 ("no-cleanup", bpo::value<bool>()->zero_tokens()->default_value(false), "do not cleanup the shm segment") // //
2896 ("hostname", bpo::value<std::string>()->default_value("localhost"), "hostname to deploy") // //
2897 ("resources", bpo::value<std::string>()->default_value(""), "resources allocated for the workflow") // //
2898 ("start-port,p", bpo::value<unsigned short>()->default_value(22000), "start port to allocate") // //
2899 ("port-range,pr", bpo::value<unsigned short>()->default_value(1000), "ports in range") // //
2900 ("completion-policy,c", bpo::value<TerminationPolicy>(&processingPolicies.termination)->default_value(TerminationPolicy::QUIT), // //
2901 "what to do when processing is finished: quit, wait") // //
2902 ("error-policy", bpo::value<TerminationPolicy>(&processingPolicies.error)->default_value(TerminationPolicy::QUIT), // //
2903 "what to do when a device has an error: quit, wait") // //
2904 ("min-failure-level", bpo::value<LogParsingHelpers::LogLevel>(&minFailureLevel)->default_value(LogParsingHelpers::LogLevel::Fatal), // //
2905 "minimum message level which will be considered as fatal and exit with 1") // //
2906 ("graphviz,g", bpo::value<bool>()->zero_tokens()->default_value(false), "produce graphviz output") // //
2907 ("mermaid", bpo::value<std::string>()->default_value(""), "produce graph output in mermaid format in file under specified name or on stdout if argument is \"-\"") // //
2908 ("timeout,t", bpo::value<uint64_t>()->default_value(0), "forced exit timeout (in seconds)") // //
2909 ("dds,D", bpo::value<std::string>()->default_value(""), "create DDS configuration") // //
2910 ("dds-workflow-suffix,D", bpo::value<std::string>()->default_value(""), "suffix for DDS names") // //
2911 ("dump-workflow,dump", bpo::value<bool>()->zero_tokens()->default_value(false), "dump workflow as JSON") // //
2912 ("dump-workflow-file", bpo::value<std::string>()->default_value("-"), "file to which do the dump") // //
2913 ("driver-mode", bpo::value<DriverMode>(&driverMode)->default_value(DriverMode::STANDALONE), R"(how to run the driver. default: "standalone". Valid: "embedded")") // //
2914 ("run", bpo::value<bool>()->zero_tokens()->default_value(false), "run workflow merged so far. It implies --batch. Use --no-batch to see the GUI") // //
2915 ("no-IPC", bpo::value<bool>()->zero_tokens()->default_value(false), "disable IPC topology optimization") // //
2916 ("o2-control,o2", bpo::value<std::string>()->default_value(""), "dump O2 Control workflow configuration under the specified name") //
2917 ("resources-monitoring", bpo::value<unsigned short>()->default_value(0), "enable cpu/memory monitoring for provided interval in seconds") //
2918 ("resources-monitoring-dump-interval", bpo::value<unsigned short>()->default_value(0), "dump monitoring information to disk every provided seconds"); //
2919 // some of the options must be forwarded by default to the device
2920 executorOptions.add(DeviceSpecHelpers::getForwardedDeviceOptions());
2921
2922 gHiddenDeviceOptions.add_options() //
2923 ("id,i", bpo::value<std::string>(), "device id for child spawning") //
2924 ("channel-config", bpo::value<std::vector<std::string>>(), "channel configuration") //
2925 ("control", "control plugin") //
2926 ("log-color", "logging color scheme")("color", "logging color scheme");
2927
2928 bpo::options_description visibleOptions;
2929 visibleOptions.add(executorOptions);
2930
2931 auto physicalWorkflow = workflow;
2932 std::map<std::string, size_t> rankIndex;
2933 // We remove the duplicates because for the moment child get themself twice:
2934 // once from the actual definition in the child, a second time from the
2935 // configuration they get passed by their parents.
2936 // Notice that we do not know in which order we will get the workflows, so
2937 // while we keep the order of DataProcessors we reshuffle them based on
2938 // some hopefully unique hash.
2939 size_t workflowHashA = 0;
2940 std::hash<std::string> hash_fn;
2941
2942 for (auto& dp : workflow) {
2943 workflowHashA += hash_fn(dp.name);
2944 }
2945
2946 for (auto& dp : workflow) {
2947 rankIndex.insert(std::make_pair(dp.name, workflowHashA));
2948 }
2949
2950 std::vector<DataProcessorInfo> dataProcessorInfos;
2951 CommandInfo commandInfo{};
2952
2953 if (isatty(STDIN_FILENO) == false && isInputConfig()) {
2954 std::vector<DataProcessorSpec> importedWorkflow;
2955 bool previousWorked = WorkflowSerializationHelpers::import(std::cin, importedWorkflow, dataProcessorInfos, commandInfo);
2956 if (previousWorked == false) {
2957 exit(1);
2958 }
2959
2960 size_t workflowHashB = 0;
2961 for (auto& dp : importedWorkflow) {
2962 workflowHashB += hash_fn(dp.name);
2963 }
2964
2965 // FIXME: Streamline...
2966 // We remove the duplicates because for the moment child get themself twice:
2967 // once from the actual definition in the child, a second time from the
2968 // configuration they get passed by their parents.
2969 for (auto& dp : importedWorkflow) {
2970 auto found = std::find_if(physicalWorkflow.begin(), physicalWorkflow.end(),
2971 [&name = dp.name](DataProcessorSpec const& spec) { return spec.name == name; });
2972 if (found == physicalWorkflow.end()) {
2973 physicalWorkflow.push_back(dp);
2974 rankIndex.insert(std::make_pair(dp.name, workflowHashB));
2975 }
2976 }
2977 }
2978
2983 for (auto& dp : physicalWorkflow) {
2984 auto isExpendable = [](DataProcessorLabel const& label) { return label.value == "expendable" || label.value == "non-critical"; };
2985 if (std::find_if(dp.labels.begin(), dp.labels.end(), isExpendable) != dp.labels.end()) {
2986 for (auto& output : dp.outputs) {
2987 if (output.lifetime == Lifetime::Timeframe) {
2988 output.lifetime = Lifetime::Sporadic;
2989 }
2990 }
2991 }
2992 }
2993
2995 OverrideServiceSpecs driverServicesOverride = ServiceSpecHelpers::parseOverrides(getenv("DPL_DRIVER_OVERRIDE_SERVICES"));
2997 // We insert the hash for the internal devices.
2998 WorkflowHelpers::injectServiceDevices(physicalWorkflow, configContext);
2999 auto reader = std::find_if(physicalWorkflow.begin(), physicalWorkflow.end(), [](DataProcessorSpec& spec) { return spec.name == "internal-dpl-aod-reader"; });
3000 if (reader != physicalWorkflow.end()) {
3001 driverServices.push_back(ArrowSupport::arrowBackendSpec());
3002 }
3003 for (auto& service : driverServices) {
3004 if (service.injectTopology == nullptr) {
3005 continue;
3006 }
3007 WorkflowSpecNode node{physicalWorkflow};
3008 service.injectTopology(node, configContext);
3009 }
3010 for (auto& dp : physicalWorkflow) {
3011 if (dp.name.rfind("internal-", 0) == 0) {
3012 rankIndex.insert(std::make_pair(dp.name, hash_fn("internal")));
3013 }
3014 }
3015
3016 // We sort dataprocessors and Inputs / outputs by name, so that the edges are
3017 // always in the same order.
3018 std::stable_sort(physicalWorkflow.begin(), physicalWorkflow.end(), [](DataProcessorSpec const& a, DataProcessorSpec const& b) {
3019 return a.name < b.name;
3020 });
3021
3022 for (auto& dp : physicalWorkflow) {
3023 std::stable_sort(dp.inputs.begin(), dp.inputs.end(),
3024 [](InputSpec const& a, InputSpec const& b) { return DataSpecUtils::describe(a) < DataSpecUtils::describe(b); });
3025 std::stable_sort(dp.outputs.begin(), dp.outputs.end(),
3026 [](OutputSpec const& a, OutputSpec const& b) { return DataSpecUtils::describe(a) < DataSpecUtils::describe(b); });
3027 }
3028
3029 // Create a list of all the edges, so that we can do a topological sort
3030 // before we create the graph.
3031 std::vector<std::pair<int, int>> edges;
3032
3033 if (physicalWorkflow.size() > 1) {
3034 edges = TopologyPolicyHelpers::buildEdges(physicalWorkflow);
3035
3036 auto topoInfos = WorkflowHelpers::topologicalSort(physicalWorkflow.size(), &edges[0].first, &edges[0].second, sizeof(std::pair<int, int>), edges.size());
3037 if (topoInfos.size() != physicalWorkflow.size()) {
3038 // Check missing resilincy of one of the tasks
3039 checkNonResiliency(physicalWorkflow, edges);
3040 throw std::runtime_error("Unable to do topological sort of the resulting workflow. Do you have loops?\n" + debugTopoInfo(physicalWorkflow, topoInfos, edges));
3041 }
3042 // Sort by layer and then by name, to ensure stability.
3043 std::stable_sort(topoInfos.begin(), topoInfos.end(), [&workflow = physicalWorkflow](TopoIndexInfo const& a, TopoIndexInfo const& b) {
3044 auto aRank = std::make_tuple(a.layer, -workflow.at(a.index).outputs.size(), workflow.at(a.index).name);
3045 auto bRank = std::make_tuple(b.layer, -workflow.at(b.index).outputs.size(), workflow.at(b.index).name);
3046 return aRank < bRank;
3047 });
3048 // Reverse index and apply the result
3049 std::vector<int> dataProcessorOrder;
3050 dataProcessorOrder.resize(topoInfos.size());
3051 for (size_t i = 0; i < topoInfos.size(); ++i) {
3052 dataProcessorOrder[topoInfos[i].index] = i;
3053 }
3054 std::vector<int> newLocations;
3055 newLocations.resize(dataProcessorOrder.size());
3056 for (size_t i = 0; i < dataProcessorOrder.size(); ++i) {
3057 newLocations[dataProcessorOrder[i]] = i;
3058 }
3059 apply_permutation(physicalWorkflow, newLocations);
3060 }
3061
3062 // Use the hidden options as veto, all config specs matching a definition
3063 // in the hidden options are skipped in order to avoid duplicate definitions
3064 // in the main parser. Note: all config specs are forwarded to devices
3065 visibleOptions.add(ConfigParamsHelper::prepareOptionDescriptions(physicalWorkflow, currentWorkflowOptions, gHiddenDeviceOptions));
3066
3067 bpo::options_description od;
3068 od.add(visibleOptions);
3069 od.add(gHiddenDeviceOptions);
3070
3071 // FIXME: decide about the policy for handling unrecognized arguments
3072 // command_line_parser with option allow_unregistered() can be used
3073 using namespace bpo::command_line_style;
3074 auto style = (allow_short | short_allow_adjacent | short_allow_next | allow_long | long_allow_adjacent | long_allow_next | allow_sticky | allow_dash_for_short);
3075 bpo::variables_map varmap;
3076 try {
3077 bpo::store(
3078 bpo::command_line_parser(argc, argv)
3079 .options(od)
3080 .style(style)
3081 .run(),
3082 varmap);
3083 } catch (std::exception const& e) {
3084 LOGP(error, "error parsing options of {}: {}", argv[0], e.what());
3085 exit(1);
3086 }
3087 conflicting_options(varmap, "dds", "o2-control");
3088 conflicting_options(varmap, "dds", "dump-workflow");
3089 conflicting_options(varmap, "dds", "run");
3090 conflicting_options(varmap, "dds", "graphviz");
3091 conflicting_options(varmap, "o2-control", "dump-workflow");
3092 conflicting_options(varmap, "o2-control", "run");
3093 conflicting_options(varmap, "o2-control", "graphviz");
3094 conflicting_options(varmap, "run", "dump-workflow");
3095 conflicting_options(varmap, "run", "graphviz");
3096 conflicting_options(varmap, "run", "mermaid");
3097 conflicting_options(varmap, "dump-workflow", "graphviz");
3098 conflicting_options(varmap, "no-batch", "batch");
3099
3100 if (varmap.count("help")) {
3101 printHelp(varmap, executorOptions, physicalWorkflow, currentWorkflowOptions);
3102 exit(0);
3103 }
3107 if (varmap.count("severity")) {
3108 auto logLevel = varmap["severity"].as<std::string>();
3109 if (logLevel == "debug") {
3110 fair::Logger::SetConsoleSeverity(fair::Severity::debug);
3111 } else if (logLevel == "detail") {
3112 fair::Logger::SetConsoleSeverity(fair::Severity::detail);
3113 } else if (logLevel == "info") {
3114 fair::Logger::SetConsoleSeverity(fair::Severity::info);
3115 } else if (logLevel == "warning") {
3116 fair::Logger::SetConsoleSeverity(fair::Severity::warning);
3117 } else if (logLevel == "error") {
3118 fair::Logger::SetConsoleSeverity(fair::Severity::error);
3119 } else if (logLevel == "important") {
3120 fair::Logger::SetConsoleSeverity(fair::Severity::important);
3121 } else if (logLevel == "alarm") {
3122 fair::Logger::SetConsoleSeverity(fair::Severity::alarm);
3123 } else if (logLevel == "critical") {
3124 fair::Logger::SetConsoleSeverity(fair::Severity::critical);
3125 } else if (logLevel == "fatal") {
3126 fair::Logger::SetConsoleSeverity(fair::Severity::fatal);
3127 } else {
3128 LOGP(error, "Invalid log level '{}'", logLevel);
3129 exit(1);
3130 }
3131 }
3132
3133 enableSignposts(varmap["signposts"].as<std::string>());
3134
3135 auto evaluateBatchOption = [&varmap]() -> bool {
3136 if (varmap.count("no-batch") > 0) {
3137 return false;
3138 }
3139 if (varmap.count("batch") == 0) {
3140 // default value
3141 return isatty(fileno(stdout)) == 0;
3142 }
3143 // FIXME: should actually use the last value, but for some reason the
3144 // values are not filled into the vector, even if specifying `-b true`
3145 // need to find out why the boost program options example is not working
3146 // in our case. Might depend on the parser options
3147 // auto value = varmap["batch"].as<std::vector<std::string>>();
3148 return true;
3149 };
3150 DriverInfo driverInfo{
3151 .sendingPolicies = sendingPolicies,
3152 .forwardingPolicies = forwardingPolicies,
3153 .callbacksPolicies = callbacksPolicies};
3154 driverInfo.states.reserve(10);
3155 driverInfo.sigintRequested = false;
3156 driverInfo.sigchldRequested = false;
3157 driverInfo.channelPolicies = channelPolicies;
3158 driverInfo.completionPolicies = completionPolicies;
3159 driverInfo.dispatchPolicies = dispatchPolicies;
3160 driverInfo.resourcePolicies = resourcePolicies;
3161 driverInfo.argc = argc;
3162 driverInfo.argv = argv;
3163 driverInfo.noSHMCleanup = varmap["no-cleanup"].as<bool>();
3164 driverInfo.processingPolicies.termination = varmap["completion-policy"].as<TerminationPolicy>();
3165 driverInfo.processingPolicies.earlyForward = varmap["early-forward-policy"].as<EarlyForwardPolicy>();
3166 driverInfo.mode = varmap["driver-mode"].as<DriverMode>();
3167
3168 auto batch = evaluateBatchOption();
3169 DriverConfig driverConfig{
3170 .batch = batch,
3171 .driverHasGUI = (batch == false) || getenv("DPL_DRIVER_REMOTE_GUI") != nullptr,
3172 };
3173
3174 if (varmap["error-policy"].defaulted() && driverConfig.batch == false) {
3175 driverInfo.processingPolicies.error = TerminationPolicy::WAIT;
3176 } else {
3177 driverInfo.processingPolicies.error = varmap["error-policy"].as<TerminationPolicy>();
3178 }
3179 driverInfo.minFailureLevel = varmap["min-failure-level"].as<LogParsingHelpers::LogLevel>();
3180 driverInfo.startTime = uv_hrtime();
3181 driverInfo.startTimeMsFromEpoch = std::chrono::duration_cast<std::chrono::milliseconds>(
3182 std::chrono::system_clock::now().time_since_epoch())
3183 .count();
3184 driverInfo.timeout = varmap["timeout"].as<uint64_t>();
3185 driverInfo.deployHostname = varmap["hostname"].as<std::string>();
3186 driverInfo.resources = varmap["resources"].as<std::string>();
3187 driverInfo.resourcesMonitoringInterval = varmap["resources-monitoring"].as<unsigned short>();
3188 driverInfo.resourcesMonitoringDumpInterval = varmap["resources-monitoring-dump-interval"].as<unsigned short>();
3189
3190 // FIXME: should use the whole dataProcessorInfos, actually...
3191 driverInfo.processorInfo = dataProcessorInfos;
3192 driverInfo.configContext = &configContext;
3193
3194 DriverControl driverControl;
3195 initialiseDriverControl(varmap, driverInfo, driverControl);
3196
3197 commandInfo.merge(CommandInfo(argc, argv));
3198
3199 std::string frameworkId;
3200 // If the id is set, this means this is a device,
3201 // otherwise this is the driver.
3202 if (varmap.count("id")) {
3203 // The framework id does not want to know anything about DDS template expansion
3204 // so we simply drop it. Notice that the "id" Property is still the same as the
3205 // original --id option.
3206 frameworkId = std::regex_replace(varmap["id"].as<std::string>(), std::regex{"_dds.*"}, "");
3207 driverInfo.uniqueWorkflowId = fmt::format("{}", getppid());
3208 driverInfo.defaultDriverClient = "stdout://";
3209 } else {
3210 driverInfo.uniqueWorkflowId = fmt::format("{}", getpid());
3211 driverInfo.defaultDriverClient = "ws://";
3212 }
3213 return runStateMachine(physicalWorkflow,
3214 currentWorkflow,
3215 dataProcessorInfos,
3216 commandInfo,
3217 driverControl,
3218 driverInfo,
3219 driverConfig,
3221 detectedParams,
3222 varmap,
3223 driverServices,
3224 frameworkId);
3225}
3226
3227void doBoostException(boost::exception&, char const* processName)
3228{
3229 LOGP(error, "error while setting up workflow in {}: {}",
3230 processName, boost::current_exception_diagnostic_information(true));
3231}
3232#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_EVENT_EMIT_INFO(log, id, name, format,...)
Definition Signpost.h:531
#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