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 = "25";
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("^table-bytes-.*");
1237 dumpableMetrics.emplace_back("^total-timeframes.*");
1238 dumpableMetrics.emplace_back("^device_state.*");
1239 dumpableMetrics.emplace_back("^total_wall_time_ms$");
1240 return dumpableMetrics;
1241}
1242
1244{
1245 auto* context = (DriverServerContext*)handle->data;
1246
1247 static auto performanceMetrics = getDumpableMetrics();
1249 context->driver->metrics, *(context->specs), performanceMetrics);
1250}
1251
1252void dumpRunSummary(DriverServerContext& context, DriverInfo const& driverInfo, DeviceInfos const& infos, DeviceSpecs const& specs)
1253{
1254 if (infos.empty()) {
1255 return;
1256 }
1257 LOGP(info, "## Processes completed. Run summary:");
1258 LOGP(info, "### Devices started: {}", infos.size());
1259 for (size_t di = 0; di < infos.size(); ++di) {
1260 auto& info = infos[di];
1261 auto& spec = specs[di];
1262 if (info.exitStatus) {
1263 LOGP(error, " - Device {}: pid {} (exit {})", spec.name, info.pid, info.exitStatus);
1264 } else {
1265 LOGP(info, " - Device {}: pid {} (exit {})", spec.name, info.pid, info.exitStatus);
1266 }
1267 if (info.exitStatus != 0 && info.firstSevereError.empty() == false) {
1268 LOGP(info, " - First error: {}", info.firstSevereError);
1269 }
1270 if (info.exitStatus != 0 && info.lastError != info.firstSevereError) {
1271 LOGP(info, " - Last error: {}", info.lastError);
1272 }
1273 }
1274 for (auto& summary : *context.summaryCallbacks) {
1275 summary(ServiceMetricsInfo{*context.metrics, *context.specs, *context.infos, context.driver->metrics, driverInfo});
1276 }
1277}
1278
1279auto bindGUIPort = [](DriverInfo& driverInfo, DriverServerContext& serverContext, std::string frameworkId) {
1280 uv_tcp_init(serverContext.loop, &serverContext.serverHandle);
1281
1282 driverInfo.port = 8080 + (getpid() % 30000);
1283
1284 if (getenv("DPL_REMOTE_GUI_PORT")) {
1285 try {
1286 driverInfo.port = stoi(std::string(getenv("DPL_REMOTE_GUI_PORT")));
1287 } catch (std::invalid_argument) {
1288 LOG(error) << "DPL_REMOTE_GUI_PORT not a valid integer";
1289 } catch (std::out_of_range) {
1290 LOG(error) << "DPL_REMOTE_GUI_PORT out of range (integer)";
1291 }
1292 if (driverInfo.port < 1024 || driverInfo.port > 65535) {
1293 LOG(error) << "DPL_REMOTE_GUI_PORT out of range (1024-65535)";
1294 }
1295 }
1296
1297 int result = 0;
1298 struct sockaddr_in* serverAddr = nullptr;
1299
1300 // Do not offer websocket endpoint for devices
1301 // FIXME: this was blocking david's workflows. For now
1302 // there is no point in any case to have devices
1303 // offering a web based API, but it might make sense in
1304 // the future to inspect them via some web based interface.
1305 if (serverContext.isDriver) {
1306 do {
1307 free(serverAddr);
1308 if (driverInfo.port > 64000) {
1309 throw runtime_error_f("Unable to find a free port for the driver. Last attempt returned %d", result);
1310 }
1311 serverAddr = (sockaddr_in*)malloc(sizeof(sockaddr_in));
1312 uv_ip4_addr("0.0.0.0", driverInfo.port, serverAddr);
1313 auto bindResult = uv_tcp_bind(&serverContext.serverHandle, (const struct sockaddr*)serverAddr, 0);
1314 if (bindResult != 0) {
1315 driverInfo.port++;
1316 usleep(1000);
1317 continue;
1318 }
1319 result = uv_listen((uv_stream_t*)&serverContext.serverHandle, 100, ws_connect_callback);
1320 if (result != 0) {
1321 driverInfo.port++;
1322 usleep(1000);
1323 continue;
1324 }
1325 } while (result != 0);
1326 } else if (getenv("DPL_DEVICE_REMOTE_GUI") && !serverContext.isDriver) {
1327 do {
1328 free(serverAddr);
1329 if (driverInfo.port > 64000) {
1330 throw runtime_error_f("Unable to find a free port for the driver. Last attempt returned %d", result);
1331 }
1332 serverAddr = (sockaddr_in*)malloc(sizeof(sockaddr_in));
1333 uv_ip4_addr("0.0.0.0", driverInfo.port, serverAddr);
1334 auto bindResult = uv_tcp_bind(&serverContext.serverHandle, (const struct sockaddr*)serverAddr, 0);
1335 if (bindResult != 0) {
1336 driverInfo.port++;
1337 usleep(1000);
1338 continue;
1339 }
1340 result = uv_listen((uv_stream_t*)&serverContext.serverHandle, 100, ws_connect_callback);
1341 if (result != 0) {
1342 driverInfo.port++;
1343 usleep(1000);
1344 continue;
1345 }
1346 LOG(info) << "Device GUI port: " << driverInfo.port << " " << frameworkId;
1347 } while (result != 0);
1348 }
1349};
1350
1351// This is the handler for the parent inner loop.
1353 WorkflowInfo const& workflowInfo,
1354 DataProcessorInfos const& previousDataProcessorInfos,
1355 CommandInfo const& commandInfo,
1356 DriverControl& driverControl,
1357 DriverInfo& driverInfo,
1358 DriverConfig& driverConfig,
1359 std::vector<DeviceMetricsInfo>& metricsInfos,
1360 std::vector<ConfigParamSpec> const& detectedParams,
1361 boost::program_options::variables_map& varmap,
1362 std::vector<ServiceSpec>& driverServices,
1363 std::string frameworkId)
1364{
1365 RunningWorkflowInfo runningWorkflow{
1366 .uniqueWorkflowId = driverInfo.uniqueWorkflowId,
1367 .shmSegmentId = (int16_t)atoi(varmap["shm-segment-id"].as<std::string>().c_str())};
1368 DeviceInfos infos;
1369 DeviceControls controls;
1370 DataProcessingStatesInfos allStates;
1371 auto* devicesManager = new DevicesManager{.controls = controls, .infos = infos, .specs = runningWorkflow.devices, .messages = {}};
1372 DeviceExecutions deviceExecutions;
1373 DataProcessorInfos dataProcessorInfos = previousDataProcessorInfos;
1374
1375 std::vector<uv_poll_t*> pollHandles;
1376 std::vector<DeviceStdioContext> childFds;
1377
1378 std::vector<ComputingResource> resources;
1379
1380 if (driverInfo.resources != "") {
1381 resources = ComputingResourceHelpers::parseResources(driverInfo.resources);
1382 } else {
1384 }
1385
1386 auto resourceManager = std::make_unique<SimpleResourceManager>(resources);
1387
1388 DebugGUI* debugGUI = nullptr;
1389 void* window = nullptr;
1390 decltype(debugGUI->getGUIDebugger(infos, runningWorkflow.devices, allStates, dataProcessorInfos, metricsInfos, driverInfo, controls, driverControl)) debugGUICallback;
1391
1392 // An empty frameworkId means this is the driver, so we initialise the GUI
1393 auto initDebugGUI = []() -> DebugGUI* {
1394 uv_lib_t supportLib;
1395 int result = 0;
1396#ifdef __APPLE__
1397 result = uv_dlopen("libO2FrameworkGUISupport.dylib", &supportLib);
1398#else
1399 result = uv_dlopen("libO2FrameworkGUISupport.so", &supportLib);
1400#endif
1401 if (result == -1) {
1402 LOG(error) << uv_dlerror(&supportLib);
1403 return nullptr;
1404 }
1405 DPLPluginHandle* (*dpl_plugin_callback)(DPLPluginHandle*);
1406
1407 result = uv_dlsym(&supportLib, "dpl_plugin_callback", (void**)&dpl_plugin_callback);
1408 if (result == -1) {
1409 LOG(error) << uv_dlerror(&supportLib);
1410 return nullptr;
1411 }
1412 DPLPluginHandle* pluginInstance = dpl_plugin_callback(nullptr);
1413 return PluginManager::getByName<DebugGUI>(pluginInstance, "ImGUIDebugGUI");
1414 };
1415
1416 // We initialise this in the driver, because different drivers might have
1417 // different versions of the service
1418 ServiceRegistry serviceRegistry;
1419
1420 if ((driverConfig.batch == false || getenv("DPL_DRIVER_REMOTE_GUI") != nullptr) && frameworkId.empty()) {
1421 debugGUI = initDebugGUI();
1422 if (debugGUI) {
1423 if (driverConfig.batch == false) {
1424 window = debugGUI->initGUI("O2 Framework debug GUI", serviceRegistry);
1425 } else {
1426 window = debugGUI->initGUI(nullptr, serviceRegistry);
1427 }
1428 }
1429 } else if (getenv("DPL_DEVICE_REMOTE_GUI") && !frameworkId.empty()) {
1430 debugGUI = initDebugGUI();
1431 // We never run the GUI on desktop for devices. All
1432 // you can do is to connect to the remote version.
1433 // this is done to avoid having a proliferation of
1434 // GUIs popping up when the variable is set globally.
1435 // FIXME: maybe this is not what we want, but it should
1436 // be ok for now.
1437 if (debugGUI) {
1438 window = debugGUI->initGUI(nullptr, serviceRegistry);
1439 }
1440 }
1441 if (driverConfig.batch == false && window == nullptr && frameworkId.empty()) {
1442 LOG(warn) << "Could not create GUI. Switching to batch mode. Do you have GLFW on your system?";
1443 driverConfig.batch = true;
1444 if (varmap["error-policy"].defaulted()) {
1445 driverInfo.processingPolicies.error = TerminationPolicy::QUIT;
1446 }
1447 }
1448 bool guiQuitRequested = false;
1449 bool hasError = false;
1450
1451 // FIXME: I should really have some way of exiting the
1452 // parent..
1453 DriverState current;
1454 DriverState previous;
1455
1456 uv_loop_t* loop = uv_loop_new();
1457
1458 uv_timer_t* gui_timer = nullptr;
1459
1460 if (!driverConfig.batch) {
1461 gui_timer = (uv_timer_t*)malloc(sizeof(uv_timer_t));
1462 uv_timer_init(loop, gui_timer);
1463 }
1464
1465 std::vector<ServiceMetricHandling> metricProcessingCallbacks;
1466 std::vector<ServiceSummaryHandling> summaryCallbacks;
1467 std::vector<ServicePreSchedule> preScheduleCallbacks;
1468 std::vector<ServicePostSchedule> postScheduleCallbacks;
1469 std::vector<ServiceDriverInit> driverInitCallbacks;
1470 for (auto& service : driverServices) {
1471 if (service.driverStartup == nullptr) {
1472 continue;
1473 }
1474 service.driverStartup(serviceRegistry, DeviceConfig{varmap});
1475 }
1476
1477 ServiceRegistryRef ref{serviceRegistry};
1478 ref.registerService(ServiceRegistryHelpers::handleForService<DevicesManager>(devicesManager));
1479
1480 bool guiTimerExpired = false;
1481 GuiCallbackContext guiContext;
1482 guiContext.plugin = debugGUI;
1483 guiContext.frameLast = uv_hrtime();
1484 guiContext.frameLatency = &driverInfo.frameLatency;
1485 guiContext.frameCost = &driverInfo.frameCost;
1486 guiContext.guiQuitRequested = &guiQuitRequested;
1487 guiContext.guiTimerExpired = &guiTimerExpired;
1488
1489 // This is to make sure we can process metrics, commands, configuration
1490 // changes coming from websocket (or even via any standard uv_stream_t, I guess).
1491 DriverServerContext serverContext{
1492 .registry = {serviceRegistry},
1493 .loop = loop,
1494 .controls = &controls,
1495 .infos = &infos,
1496 .states = &allStates,
1497 .specs = &runningWorkflow.devices,
1498 .metrics = &metricsInfos,
1499 .metricProcessingCallbacks = &metricProcessingCallbacks,
1500 .summaryCallbacks = &summaryCallbacks,
1501 .driver = &driverInfo,
1502 .gui = &guiContext,
1503 .isDriver = frameworkId.empty(),
1504 };
1505
1506 serverContext.serverHandle.data = &serverContext;
1507
1508 uv_timer_t force_step_timer;
1509 uv_timer_init(loop, &force_step_timer);
1510 uv_timer_t force_exit_timer;
1511 uv_timer_init(loop, &force_exit_timer);
1512
1513 bool guiDeployedOnce = false;
1514 bool once = false;
1515
1516 uv_timer_t metricDumpTimer;
1517 metricDumpTimer.data = &serverContext;
1518 bool allChildrenGone = false;
1519 guiContext.allChildrenGone = &allChildrenGone;
1520 O2_SIGNPOST_ID_FROM_POINTER(sid, driver, loop);
1521 O2_SIGNPOST_START(driver, sid, "driver", "Starting driver loop");
1522
1523 // Async callback to process the output of the children, if needed.
1524 serverContext.asyncLogProcessing = (uv_async_t*)malloc(sizeof(uv_async_t));
1525 serverContext.asyncLogProcessing->data = &serverContext;
1526 uv_async_init(loop, serverContext.asyncLogProcessing, [](uv_async_t* handle) {
1527 auto* context = (DriverServerContext*)handle->data;
1528 processChildrenOutput(context->loop, *context->driver, *context->infos, *context->specs, *context->controls);
1529 });
1530
1531 while (true) {
1532 // If control forced some transition on us, we push it to the queue.
1533 if (driverControl.forcedTransitions.empty() == false) {
1534 for (auto transition : driverControl.forcedTransitions) {
1535 driverInfo.states.push_back(transition);
1536 }
1537 driverControl.forcedTransitions.resize(0);
1538 }
1539 // In case a timeout was requested, we check if we are running
1540 // for more than the timeout duration and exit in case that's the case.
1541 {
1542 auto currentTime = uv_hrtime();
1543 uint64_t diff = (currentTime - driverInfo.startTime) / 1000000000LL;
1544 if ((graceful_exit == false) && (driverInfo.timeout > 0) && (diff > driverInfo.timeout)) {
1545 LOG(info) << "Timout ellapsed. Requesting to quit.";
1546 graceful_exit = true;
1547 }
1548 }
1549 // Move to exit loop if sigint was sent we execute this only once.
1550 if (graceful_exit == true && driverInfo.sigintRequested == false) {
1551 driverInfo.sigintRequested = true;
1552 driverInfo.states.resize(0);
1553 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
1554 }
1555 // If one of the children dies and sigint was not requested
1556 // we should decide what to do.
1557 if (sigchld_requested == true && driverInfo.sigchldRequested == false) {
1558 driverInfo.sigchldRequested = true;
1559 driverInfo.states.push_back(DriverState::HANDLE_CHILDREN);
1560 }
1561 if (driverInfo.states.empty() == false) {
1562 previous = current;
1563 current = driverInfo.states.back();
1564 } else {
1565 current = DriverState::UNKNOWN;
1566 }
1567 driverInfo.states.pop_back();
1568 switch (current) {
1569 case DriverState::BIND_GUI_PORT:
1570 bindGUIPort(driverInfo, serverContext, frameworkId);
1571 break;
1572 case DriverState::INIT:
1573 LOGP(info, "Initialising O2 Data Processing Layer. Driver PID: {}.", getpid());
1574 LOGP(info, "Driver listening on port: {}", driverInfo.port);
1575
1576 // Install signal handler for quitting children.
1577 driverInfo.sa_handle_child.sa_handler = &handle_sigchld;
1578 sigemptyset(&driverInfo.sa_handle_child.sa_mask);
1579 driverInfo.sa_handle_child.sa_flags = SA_RESTART | SA_NOCLDSTOP;
1580 if (sigaction(SIGCHLD, &driverInfo.sa_handle_child, nullptr) == -1) {
1581 perror(nullptr);
1582 exit(1);
1583 }
1584
1587 if (driverInfo.noSHMCleanup) {
1588 LOGP(warning, "Not cleaning up shared memory.");
1589 } else {
1590 cleanupSHM(driverInfo.uniqueWorkflowId);
1591 }
1596 for (auto& callback : driverInitCallbacks) {
1597 callback(serviceRegistry, {varmap});
1598 }
1599 driverInfo.states.push_back(DriverState::RUNNING);
1600 // driverInfo.states.push_back(DriverState::REDEPLOY_GUI);
1601 LOG(info) << "O2 Data Processing Layer initialised. We brake for nobody.";
1602#ifdef NDEBUG
1603 LOGF(info, "Optimised build. O2DEBUG / LOG(debug) / LOGF(debug) / assert statement will not be shown.");
1604#endif
1605 break;
1606 case DriverState::IMPORT_CURRENT_WORKFLOW:
1607 // This state is needed to fill the metadata structure
1608 // which contains how to run the current workflow
1609 dataProcessorInfos = previousDataProcessorInfos;
1610 for (auto const& device : runningWorkflow.devices) {
1611 auto exists = std::find_if(dataProcessorInfos.begin(),
1612 dataProcessorInfos.end(),
1613 [id = device.id](DataProcessorInfo const& info) -> bool { return info.name == id; });
1614 if (exists != dataProcessorInfos.end()) {
1615 continue;
1616 }
1617 std::vector<std::string> channels;
1618 for (auto channel : device.inputChannels) {
1619 channels.push_back(channel.name);
1620 }
1621 for (auto channel : device.outputChannels) {
1622 channels.push_back(channel.name);
1623 }
1624 dataProcessorInfos.push_back(
1626 device.id,
1627 workflowInfo.executable,
1628 workflowInfo.args,
1629 workflowInfo.options,
1630 channels});
1631 }
1632 break;
1633 case DriverState::MATERIALISE_WORKFLOW:
1634 try {
1635 auto workflowState = WorkflowHelpers::verifyWorkflow(workflow);
1636 if (driverConfig.batch == true && varmap["dds"].as<std::string>().empty() && !varmap["dump-workflow"].as<bool>() && workflowState == WorkflowParsingState::Empty) {
1637 LOGP(error, "Empty workflow provided while running in batch mode.");
1638 return 1;
1639 }
1640
1643 auto altered_workflow = workflow;
1644
1645 auto confNameFromParam = [](std::string const& paramName) {
1646 std::regex name_regex(R"(^control:([\w-]+)\/(\w+))");
1647 auto match = std::sregex_token_iterator(paramName.begin(), paramName.end(), name_regex, 0);
1648 if (match == std::sregex_token_iterator()) {
1649 throw runtime_error_f("Malformed process control spec: %s", paramName.c_str());
1650 }
1651 std::string task = std::sregex_token_iterator(paramName.begin(), paramName.end(), name_regex, 1)->str();
1652 std::string conf = std::sregex_token_iterator(paramName.begin(), paramName.end(), name_regex, 2)->str();
1653 return std::pair{task, conf};
1654 };
1655 bool altered = false;
1656 for (auto& device : altered_workflow) {
1657 // ignore internal devices
1658 if (device.name.find("internal") != std::string::npos) {
1659 continue;
1660 }
1661 // ignore devices with no inputs
1662 if (device.inputs.empty() == true) {
1663 continue;
1664 }
1665 // ignore devices with no metadata in inputs
1666 auto hasMetadata = std::any_of(device.inputs.begin(), device.inputs.end(), [](InputSpec const& spec) {
1667 return spec.metadata.empty() == false;
1668 });
1669 if (!hasMetadata) {
1670 continue;
1671 }
1672 // ignore devices with no control options
1673 auto hasControls = std::any_of(device.inputs.begin(), device.inputs.end(), [](InputSpec const& spec) {
1674 return std::any_of(spec.metadata.begin(), spec.metadata.end(), [](ConfigParamSpec const& param) {
1675 return param.type == VariantType::Bool && param.name.find("control:") != std::string::npos;
1676 });
1677 });
1678 if (!hasControls) {
1679 continue;
1680 }
1681
1682 LOGP(debug, "Adjusting device {}", device.name.c_str());
1683
1684 auto configStore = DeviceConfigurationHelpers::getConfiguration(serviceRegistry, device.name.c_str(), device.options);
1685 if (configStore != nullptr) {
1686 auto reg = std::make_unique<ConfigParamRegistry>(std::move(configStore));
1687 for (auto& input : device.inputs) {
1688 for (auto& param : input.metadata) {
1689 if (param.type == VariantType::Bool && param.name.find("control:") != std::string::npos) {
1690 if (param.name != "control:default" && param.name != "control:spawn" && param.name != "control:build" && param.name != "control:define") {
1691 auto confName = confNameFromParam(param.name).second;
1692 param.defaultValue = reg->get<bool>(confName.c_str());
1693 }
1694 }
1695 }
1696 }
1697 }
1699 LOGP(debug, "Original inputs: ");
1700 for (auto& input : device.inputs) {
1701 LOGP(debug, "-> {}", input.binding);
1702 }
1703 auto end = device.inputs.end();
1704 auto new_end = std::remove_if(device.inputs.begin(), device.inputs.end(), [](InputSpec& input) {
1705 auto requested = false;
1706 auto hasControls = false;
1707 for (auto& param : input.metadata) {
1708 if (param.type != VariantType::Bool) {
1709 continue;
1710 }
1711 if (param.name.find("control:") != std::string::npos) {
1712 hasControls = true;
1713 if (param.defaultValue.get<bool>() == true) {
1714 requested = true;
1715 break;
1716 }
1717 }
1718 }
1719 if (hasControls) {
1720 return !requested;
1721 }
1722 return false;
1723 });
1724 device.inputs.erase(new_end, end);
1725 LOGP(debug, "Adjusted inputs: ");
1726 for (auto& input : device.inputs) {
1727 LOGP(debug, "-> {}", input.binding);
1728 }
1729 altered = true;
1730 }
1731 WorkflowHelpers::adjustTopology(altered_workflow, *driverInfo.configContext);
1732 if (altered) {
1733 WorkflowSpecNode node{altered_workflow};
1734 for (auto& service : driverServices) {
1735 if (service.adjustTopology == nullptr) {
1736 continue;
1737 }
1738 service.adjustTopology(node, *driverInfo.configContext);
1739 }
1740 }
1741
1742 // These allow services customization via an environment variable
1743 OverrideServiceSpecs overrides = ServiceSpecHelpers::parseOverrides(getenv("DPL_OVERRIDE_SERVICES"));
1744 DeviceSpecHelpers::validate(altered_workflow);
1746 driverInfo.channelPolicies,
1747 driverInfo.completionPolicies,
1748 driverInfo.dispatchPolicies,
1749 driverInfo.resourcePolicies,
1750 driverInfo.callbacksPolicies,
1751 driverInfo.sendingPolicies,
1752 driverInfo.forwardingPolicies,
1753 runningWorkflow.devices,
1754 *resourceManager,
1755 driverInfo.uniqueWorkflowId,
1756 *driverInfo.configContext,
1757 !varmap["no-IPC"].as<bool>(),
1758 driverInfo.resourcesMonitoringInterval,
1759 varmap["channel-prefix"].as<std::string>(),
1760 overrides);
1761 metricProcessingCallbacks.clear();
1762 std::vector<std::string> matchingServices;
1763
1764 // FIXME: once moving to C++20, we can use templated lambdas.
1765 matchingServices.clear();
1766 for (auto& device : runningWorkflow.devices) {
1767 for (auto& service : device.services) {
1768 // If a service with the same name is already registered, skip it
1769 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1770 continue;
1771 }
1772 if (service.metricHandling) {
1773 metricProcessingCallbacks.push_back(service.metricHandling);
1774 matchingServices.push_back(service.name);
1775 }
1776 }
1777 }
1778
1779 // FIXME: once moving to C++20, we can use templated lambdas.
1780 matchingServices.clear();
1781 for (auto& device : runningWorkflow.devices) {
1782 for (auto& service : device.services) {
1783 // If a service with the same name is already registered, skip it
1784 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1785 continue;
1786 }
1787 if (service.summaryHandling) {
1788 summaryCallbacks.push_back(service.summaryHandling);
1789 matchingServices.push_back(service.name);
1790 }
1791 }
1792 }
1793
1794 preScheduleCallbacks.clear();
1795 matchingServices.clear();
1796 for (auto& device : runningWorkflow.devices) {
1797 for (auto& service : device.services) {
1798 // If a service with the same name is already registered, skip it
1799 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1800 continue;
1801 }
1802 if (service.preSchedule) {
1803 preScheduleCallbacks.push_back(service.preSchedule);
1804 }
1805 }
1806 }
1807 postScheduleCallbacks.clear();
1808 matchingServices.clear();
1809 for (auto& device : runningWorkflow.devices) {
1810 for (auto& service : device.services) {
1811 // If a service with the same name is already registered, skip it
1812 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1813 continue;
1814 }
1815 if (service.postSchedule) {
1816 postScheduleCallbacks.push_back(service.postSchedule);
1817 }
1818 }
1819 }
1820 driverInitCallbacks.clear();
1821 matchingServices.clear();
1822 for (auto& device : runningWorkflow.devices) {
1823 for (auto& service : device.services) {
1824 // If a service with the same name is already registered, skip it
1825 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1826 continue;
1827 }
1828 if (service.driverInit) {
1829 driverInitCallbacks.push_back(service.driverInit);
1830 }
1831 }
1832 }
1833
1834 // This should expand nodes so that we can build a consistent DAG.
1835
1836 // This updates the options in the runningWorkflow.devices
1837 for (auto& device : runningWorkflow.devices) {
1838 // ignore internal devices
1839 if (device.name.find("internal") != std::string::npos) {
1840 continue;
1841 }
1842 auto configStore = DeviceConfigurationHelpers::getConfiguration(serviceRegistry, device.name.c_str(), device.options);
1843 if (configStore != nullptr) {
1844 auto reg = std::make_unique<ConfigParamRegistry>(std::move(configStore));
1845 for (auto& option : device.options) {
1846 const char* name = option.name.c_str();
1847 switch (option.type) {
1848 case VariantType::Int:
1849 option.defaultValue = reg->get<int32_t>(name);
1850 break;
1851 case VariantType::Int8:
1852 option.defaultValue = reg->get<int8_t>(name);
1853 break;
1854 case VariantType::Int16:
1855 option.defaultValue = reg->get<int16_t>(name);
1856 break;
1857 case VariantType::UInt8:
1858 option.defaultValue = reg->get<uint8_t>(name);
1859 break;
1860 case VariantType::UInt16:
1861 option.defaultValue = reg->get<uint16_t>(name);
1862 break;
1863 case VariantType::UInt32:
1864 option.defaultValue = reg->get<uint32_t>(name);
1865 break;
1866 case VariantType::UInt64:
1867 option.defaultValue = reg->get<uint64_t>(name);
1868 break;
1869 case VariantType::Int64:
1870 option.defaultValue = reg->get<int64_t>(name);
1871 break;
1872 case VariantType::Float:
1873 option.defaultValue = reg->get<float>(name);
1874 break;
1875 case VariantType::Double:
1876 option.defaultValue = reg->get<double>(name);
1877 break;
1878 case VariantType::String:
1879 option.defaultValue = reg->get<std::string>(name);
1880 break;
1881 case VariantType::Bool:
1882 option.defaultValue = reg->get<bool>(name);
1883 break;
1884 case VariantType::ArrayInt:
1885 option.defaultValue = reg->get<std::vector<int>>(name);
1886 break;
1887 case VariantType::ArrayFloat:
1888 option.defaultValue = reg->get<std::vector<float>>(name);
1889 break;
1890 case VariantType::ArrayDouble:
1891 option.defaultValue = reg->get<std::vector<double>>(name);
1892 break;
1893 case VariantType::ArrayString:
1894 option.defaultValue = reg->get<std::vector<std::string>>(name);
1895 break;
1896 case VariantType::Array2DInt:
1897 option.defaultValue = reg->get<Array2D<int>>(name);
1898 break;
1899 case VariantType::Array2DFloat:
1900 option.defaultValue = reg->get<Array2D<float>>(name);
1901 break;
1902 case VariantType::Array2DDouble:
1903 option.defaultValue = reg->get<Array2D<double>>(name);
1904 break;
1905 case VariantType::LabeledArrayInt:
1906 option.defaultValue = reg->get<LabeledArray<int>>(name);
1907 break;
1908 case VariantType::LabeledArrayFloat:
1909 option.defaultValue = reg->get<LabeledArray<float>>(name);
1910 break;
1911 case VariantType::LabeledArrayDouble:
1912 option.defaultValue = reg->get<LabeledArray<double>>(name);
1913 break;
1914 case VariantType::LabeledArrayString:
1915 option.defaultValue = reg->get<LabeledArray<std::string>>(name);
1916 break;
1917 default:
1918 break;
1919 }
1920 }
1921 }
1922 }
1923 } catch (std::runtime_error& e) {
1924 LOGP(error, "invalid workflow in {}: {}", driverInfo.argv[0], e.what());
1925 return 1;
1928#ifdef DPL_ENABLE_BACKTRACE
1929 BacktraceHelpers::demangled_backtrace_symbols(err.backtrace, err.maxBacktrace, STDERR_FILENO);
1930#endif
1931 LOGP(error, "invalid workflow in {}: {}", driverInfo.argv[0], err.what);
1932 return 1;
1933 } catch (...) {
1934 LOGP(error, "invalid workflow in {}: Unknown error while materialising workflow", driverInfo.argv[0]);
1935 return 1;
1936 }
1937 break;
1938 case DriverState::DO_CHILD:
1939 // We do not start the process if by default we are stopped.
1940 if (driverControl.defaultStopped) {
1941 kill(getpid(), SIGSTOP);
1942 }
1943 for (size_t di = 0; di < runningWorkflow.devices.size(); di++) {
1945 if (runningWorkflow.devices[di].id == frameworkId) {
1946 return doChild(driverInfo.argc, driverInfo.argv,
1947 serviceRegistry,
1948 runningWorkflow, ref,
1949 driverConfig,
1950 driverInfo.processingPolicies,
1951 driverInfo.defaultDriverClient,
1952 loop);
1953 }
1954 }
1955 {
1956 std::ostringstream ss;
1957 for (auto& processor : workflow) {
1958 ss << " - " << processor.name << "\n";
1959 }
1960 for (auto& spec : runningWorkflow.devices) {
1961 ss << " - " << spec.name << "(" << spec.id << ")"
1962 << "\n";
1963 }
1964 driverInfo.lastError = fmt::format(
1965 "Unable to find component with id {}."
1966 " Available options:\n{}",
1967 frameworkId, ss.str());
1968 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
1969 }
1970 break;
1971 case DriverState::REDEPLOY_GUI:
1972 // The callback for the GUI needs to be recalculated every time
1973 // the deployed configuration changes, e.g. a new device
1974 // has been added to the topology.
1975 // We need to recreate the GUI callback every time we reschedule
1976 // because getGUIDebugger actually recreates the GUI state.
1977 // Notice also that we need the actual gui_timer only for the
1978 // case the GUI runs in interactive mode, however we deploy the
1979 // GUI in both interactive and non-interactive mode, if the
1980 // DPL_DRIVER_REMOTE_GUI environment variable is set.
1981 if (!driverConfig.batch || getenv("DPL_DRIVER_REMOTE_GUI")) {
1982 if (gui_timer) {
1983 uv_timer_stop(gui_timer);
1984 }
1985
1986 auto callback = debugGUI->getGUIDebugger(infos, runningWorkflow.devices, allStates, dataProcessorInfos, metricsInfos, driverInfo, controls, driverControl);
1987 guiContext.callback = [&serviceRegistry, &driverServices, &debugGUI, &infos, &runningWorkflow, &dataProcessorInfos, &metricsInfos, &driverInfo, &controls, &driverControl, callback]() {
1988 callback();
1989 for (auto& service : driverServices) {
1990 if (service.postRenderGUI) {
1991 service.postRenderGUI(serviceRegistry);
1992 }
1993 }
1994 };
1995 guiContext.window = window;
1996
1997 if (gui_timer) {
1998 gui_timer->data = &guiContext;
1999 uv_timer_start(gui_timer, gui_callback, 0, 20);
2000 }
2001 guiDeployedOnce = true;
2002 }
2003 break;
2004 case DriverState::MERGE_CONFIGS: {
2005 try {
2006 controls.resize(runningWorkflow.devices.size());
2009 if (varmap.count("dpl-tracing-flags")) {
2010 for (auto& control : controls) {
2011 auto tracingFlags = DeviceStateHelpers::parseTracingFlags(varmap["dpl-tracing-flags"].as<std::string>());
2012 control.tracingFlags = tracingFlags;
2013 }
2014 }
2015 deviceExecutions.resize(runningWorkflow.devices.size());
2016
2017 // Options which should be uniform across all
2018 // the subworkflow invokations.
2019 const auto uniformOptions = {
2020 "--aod-file",
2021 "--aod-memory-rate-limit",
2022 "--aod-writer-json",
2023 "--aod-writer-ntfmerge",
2024 "--aod-writer-resdir",
2025 "--aod-writer-resfile",
2026 "--aod-writer-resmode",
2027 "--aod-writer-maxfilesize",
2028 "--aod-writer-keep",
2029 "--aod-max-io-rate",
2030 "--aod-parent-access-level",
2031 "--aod-parent-base-path-replacement",
2032 "--driver-client-backend",
2033 "--fairmq-ipc-prefix",
2034 "--readers",
2035 "--resources-monitoring",
2036 "--resources-monitoring-dump-interval",
2037 "--time-limit",
2038 };
2039
2040 for (auto& option : uniformOptions) {
2041 DeviceSpecHelpers::reworkHomogeneousOption(dataProcessorInfos, option, nullptr);
2042 }
2043
2044 DeviceSpecHelpers::reworkShmSegmentSize(dataProcessorInfos);
2046 driverControl.defaultStopped,
2047 driverInfo.processingPolicies.termination == TerminationPolicy::WAIT,
2048 driverInfo.port,
2049 driverConfig,
2050 dataProcessorInfos,
2051 runningWorkflow.devices,
2052 deviceExecutions,
2053 controls,
2054 detectedParams,
2055 driverInfo.uniqueWorkflowId);
2058 LOGP(error, "unable to merge configurations in {}: {}", driverInfo.argv[0], err.what);
2059#ifdef DPL_ENABLE_BACKTRACE
2060 std::cerr << "\nStacktrace follows:\n\n";
2061 BacktraceHelpers::demangled_backtrace_symbols(err.backtrace, err.maxBacktrace, STDERR_FILENO);
2062#endif
2063 return 1;
2064 }
2065 } break;
2066 case DriverState::SCHEDULE: {
2067 // FIXME: for the moment modifying the topology means we rebuild completely
2068 // all the devices and we restart them. This is also what DDS does at
2069 // a larger scale. In principle one could try to do a delta and only
2070 // restart the data processors which need to be restarted.
2071 LOG(info) << "Redeployment of configuration asked.";
2072 std::ostringstream forwardedStdin;
2073 WorkflowSerializationHelpers::dump(forwardedStdin, workflow, dataProcessorInfos, commandInfo);
2074 infos.reserve(runningWorkflow.devices.size());
2075
2076 // This is guaranteed to be a single CPU.
2077 unsigned parentCPU = -1;
2078 unsigned parentNode = -1;
2079#if defined(__linux__) && __has_include(<sched.h>)
2080 parentCPU = sched_getcpu();
2081#elif __has_include(<linux/getcpu.h>)
2082 getcpu(&parentCPU, &parentNode, nullptr);
2083#elif __has_include(<cpuid.h>) && (__x86_64__ || __i386__)
2084 // FIXME: this is a last resort as it is apparently buggy
2085 // on some Intel CPUs.
2086 GETCPU(parentCPU);
2087#endif
2088 for (auto& callback : preScheduleCallbacks) {
2089 callback(serviceRegistry, {varmap});
2090 }
2091 childFds.resize(runningWorkflow.devices.size());
2092 for (int di = 0; di < (int)runningWorkflow.devices.size(); ++di) {
2093 auto& context = childFds[di];
2094 createPipes(context.childstdin);
2095 createPipes(context.childstdout);
2096 if (driverInfo.mode == DriverMode::EMBEDDED || runningWorkflow.devices[di].resource.hostname != driverInfo.deployHostname) {
2097 spawnRemoteDevice(loop, forwardedStdin.str(),
2098 runningWorkflow.devices[di], controls[di], deviceExecutions[di], infos, allStates);
2099 } else {
2100 DeviceRef ref{di};
2101 spawnDevice(loop,
2102 ref,
2103 runningWorkflow.devices, driverInfo,
2104 controls, deviceExecutions, infos,
2105 allStates,
2106 serviceRegistry, varmap,
2107 childFds, parentCPU, parentNode);
2108 }
2109 }
2110 handleSignals();
2111 handleChildrenStdio(&serverContext, forwardedStdin.str(), childFds, pollHandles);
2112 for (auto& callback : postScheduleCallbacks) {
2113 callback(serviceRegistry, {varmap});
2114 }
2115 assert(infos.empty() == false);
2116
2117 // In case resource monitoring is requested, we dump metrics to disk
2118 // every 3 minutes.
2119 if (driverInfo.resourcesMonitoringDumpInterval && ResourcesMonitoringHelper::isResourcesMonitoringEnabled(driverInfo.resourcesMonitoringInterval)) {
2120 uv_timer_init(loop, &metricDumpTimer);
2121 uv_timer_start(&metricDumpTimer, dumpMetricsCallback,
2122 driverInfo.resourcesMonitoringDumpInterval * 1000,
2123 driverInfo.resourcesMonitoringDumpInterval * 1000);
2124 }
2126 for (const auto& processorInfo : dataProcessorInfos) {
2127 const auto& cmdLineArgs = processorInfo.cmdLineArgs;
2128 if (std::find(cmdLineArgs.begin(), cmdLineArgs.end(), "--severity") != cmdLineArgs.end()) {
2129 for (size_t counter = 0; const auto& spec : runningWorkflow.devices) {
2130 if (spec.name.compare(processorInfo.name) == 0) {
2131 auto& info = infos[counter];
2132 const auto logLevelIt = std::find(cmdLineArgs.begin(), cmdLineArgs.end(), "--severity") + 1;
2133 if ((*logLevelIt).compare("debug") == 0) {
2134 info.logLevel = LogParsingHelpers::LogLevel::Debug;
2135 } else if ((*logLevelIt).compare("detail") == 0) {
2136 info.logLevel = LogParsingHelpers::LogLevel::Debug;
2137 } else if ((*logLevelIt).compare("info") == 0) {
2138 info.logLevel = LogParsingHelpers::LogLevel::Info;
2139 } else if ((*logLevelIt).compare("warning") == 0) {
2140 info.logLevel = LogParsingHelpers::LogLevel::Warning;
2141 } else if ((*logLevelIt).compare("error") == 0) {
2142 info.logLevel = LogParsingHelpers::LogLevel::Error;
2143 } else if ((*logLevelIt).compare("important") == 0) {
2144 info.logLevel = LogParsingHelpers::LogLevel::Info;
2145 } else if ((*logLevelIt).compare("alarm") == 0) {
2146 info.logLevel = LogParsingHelpers::LogLevel::Alarm;
2147 } else if ((*logLevelIt).compare("critical") == 0) {
2148 info.logLevel = LogParsingHelpers::LogLevel::Critical;
2149 } else if ((*logLevelIt).compare("fatal") == 0) {
2150 info.logLevel = LogParsingHelpers::LogLevel::Fatal;
2151 }
2152 break;
2153 }
2154 ++counter;
2155 }
2156 }
2157 }
2158 LOG(info) << "Redeployment of configuration done.";
2159 } break;
2160 case DriverState::RUNNING:
2161 // Run any pending libUV event loop, block if
2162 // any, so that we do not consume CPU time when the driver is
2163 // idle.
2164 devicesManager->flush();
2165 // We print the event loop for the gui only once every
2166 // 6000 iterations (i.e. ~2 minutes). To avoid spamming, while still
2167 // being able to see the event loop in case of a deadlock / systematic failure.
2168 if (guiTimerExpired == false) {
2169 O2_SIGNPOST_EVENT_EMIT(driver, sid, "mainloop", "Entering event loop with %{public}s", once ? "UV_RUN_ONCE" : "UV_RUN_NOWAIT");
2170 }
2171 uv_run(loop, once ? UV_RUN_ONCE : UV_RUN_NOWAIT);
2172 once = true;
2173 // Calculate what we should do next and eventually
2174 // show the GUI
2175 if (guiQuitRequested ||
2176 (driverInfo.processingPolicies.termination == TerminationPolicy::QUIT && (checkIfCanExit(infos) == true))) {
2177 // Something requested to quit. This can be a user
2178 // interaction with the GUI or (if --completion-policy=quit)
2179 // it could mean that the workflow does not have anything else to do.
2180 // Let's update the GUI one more time and then EXIT.
2181 LOG(info) << "Quitting";
2182 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
2183 } else if (infos.size() != runningWorkflow.devices.size()) {
2184 // If the number of devices is different from
2185 // the DeviceInfos it means the speicification
2186 // does not match what is running, so we need to do
2187 // further scheduling.
2188 driverInfo.states.push_back(DriverState::RUNNING);
2189 driverInfo.states.push_back(DriverState::REDEPLOY_GUI);
2190 driverInfo.states.push_back(DriverState::SCHEDULE);
2191 driverInfo.states.push_back(DriverState::MERGE_CONFIGS);
2192 } else if (runningWorkflow.devices.empty() && driverConfig.batch == true) {
2193 LOG(info) << "No device resulting from the workflow. Quitting.";
2194 // If there are no deviceSpecs, we exit.
2195 driverInfo.states.push_back(DriverState::EXIT);
2196 } else if (runningWorkflow.devices.empty() && driverConfig.batch == false && !guiDeployedOnce) {
2197 // In case of an empty workflow, we need to deploy the GUI at least once.
2198 driverInfo.states.push_back(DriverState::RUNNING);
2199 driverInfo.states.push_back(DriverState::REDEPLOY_GUI);
2200 } else {
2201 driverInfo.states.push_back(DriverState::RUNNING);
2202 }
2203 break;
2204 case DriverState::QUIT_REQUESTED:
2205 LOG(info) << "QUIT_REQUESTED";
2206 guiQuitRequested = true;
2207 // We send SIGCONT to make sure stopped children are resumed
2208 killChildren(infos, SIGCONT);
2209 // We send SIGTERM to make sure we do the STOP transition in FairMQ
2210 killChildren(infos, SIGTERM);
2211 // We have a timer to send SIGUSR1 to make sure we advance all devices
2212 // in a timely manner.
2213 force_step_timer.data = &infos;
2214 uv_timer_start(&force_step_timer, single_step_callback, 0, 300);
2215 driverInfo.states.push_back(DriverState::HANDLE_CHILDREN);
2216 break;
2217 case DriverState::HANDLE_CHILDREN: {
2218 // Run any pending libUV event loop, block if
2219 // any, so that we do not consume CPU time when the driver is
2220 // idle.
2221 uv_run(loop, once ? UV_RUN_ONCE : UV_RUN_NOWAIT);
2222 once = true;
2223 // I allow queueing of more sigchld only when
2224 // I process the previous call
2225 if (forceful_exit == true) {
2226 static bool forcefulExitMessage = true;
2227 if (forcefulExitMessage) {
2228 LOG(info) << "Forceful exit requested.";
2229 forcefulExitMessage = false;
2230 }
2231 killChildren(infos, SIGCONT);
2232 killChildren(infos, SIGKILL);
2233 }
2234 sigchld_requested = false;
2235 driverInfo.sigchldRequested = false;
2236 processChildrenOutput(loop, driverInfo, infos, runningWorkflow.devices, controls);
2237 hasError = processSigChild(infos, runningWorkflow.devices);
2238 allChildrenGone = areAllChildrenGone(infos);
2239 bool canExit = checkIfCanExit(infos);
2240 bool supposedToQuit = (guiQuitRequested || canExit || graceful_exit);
2241
2242 if (allChildrenGone && (supposedToQuit || driverInfo.processingPolicies.termination == TerminationPolicy::QUIT)) {
2243 // We move to the exit, regardless of where we were
2244 driverInfo.states.resize(0);
2245 driverInfo.states.push_back(DriverState::EXIT);
2246 } else if (hasError && driverInfo.processingPolicies.error == TerminationPolicy::QUIT && !supposedToQuit) {
2247 graceful_exit = 1;
2248 force_exit_timer.data = &infos;
2249 static bool forceful_timer_started = false;
2250 if (forceful_timer_started == false) {
2251 forceful_timer_started = true;
2252 uv_timer_start(&force_exit_timer, force_exit_callback, 15000, 3000);
2253 }
2254 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
2255 } else if (allChildrenGone == false && supposedToQuit) {
2256 driverInfo.states.push_back(DriverState::HANDLE_CHILDREN);
2257 } else {
2258 }
2259 } break;
2260 case DriverState::EXIT: {
2261 if (ResourcesMonitoringHelper::isResourcesMonitoringEnabled(driverInfo.resourcesMonitoringInterval)) {
2262 if (driverInfo.resourcesMonitoringDumpInterval) {
2263 uv_timer_stop(&metricDumpTimer);
2264 }
2265 LOG(info) << "Dumping performance metrics to performanceMetrics.json file";
2266 dumpMetricsCallback(&metricDumpTimer);
2267 }
2268 dumpRunSummary(serverContext, driverInfo, infos, runningWorkflow.devices);
2269 // This is a clean exit. Before we do so, if required,
2270 // we dump the configuration of all the devices so that
2271 // we can reuse it. Notice we do not dump anything if
2272 // the workflow was not really run.
2273 // NOTE: is this really what we want? should we run
2274 // SCHEDULE and dump the full configuration as well?
2275 if (infos.empty()) {
2276 return 0;
2277 }
2278 boost::property_tree::ptree finalConfig;
2279 assert(infos.size() == runningWorkflow.devices.size());
2280 for (size_t di = 0; di < infos.size(); ++di) {
2281 auto& info = infos[di];
2282 auto& spec = runningWorkflow.devices[di];
2283 finalConfig.put_child(spec.name, info.currentConfig);
2284 }
2285 LOG(info) << "Dumping used configuration in dpl-config.json";
2286
2287 std::ofstream outDPLConfigFile("dpl-config.json", std::ios::out);
2288 if (outDPLConfigFile.is_open()) {
2289 boost::property_tree::write_json(outDPLConfigFile, finalConfig);
2290 } else {
2291 LOGP(warning, "Could not write out final configuration file. Read only run folder?");
2292 }
2293 if (driverInfo.noSHMCleanup) {
2294 LOGP(warning, "Not cleaning up shared memory.");
2295 } else {
2296 cleanupSHM(driverInfo.uniqueWorkflowId);
2297 }
2298 return calculateExitCode(driverInfo, runningWorkflow.devices, infos);
2299 }
2300 case DriverState::PERFORM_CALLBACKS:
2301 for (auto& callback : driverControl.callbacks) {
2302 callback(workflow, runningWorkflow.devices, deviceExecutions, dataProcessorInfos, commandInfo);
2303 }
2304 driverControl.callbacks.clear();
2305 break;
2306 default:
2307 LOG(error) << "Driver transitioned in an unknown state("
2308 << "current: " << (int)current
2309 << ", previous: " << (int)previous
2310 << "). Shutting down.";
2311 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
2312 }
2313 }
2314 O2_SIGNPOST_END(driver, sid, "driver", "End driver loop");
2315}
2316
2317// Print help
2318void printHelp(bpo::variables_map const& varmap,
2319 bpo::options_description const& executorOptions,
2320 std::vector<DataProcessorSpec> const& physicalWorkflow,
2321 std::vector<ConfigParamSpec> const& currentWorkflowOptions)
2322{
2323 auto mode = varmap["help"].as<std::string>();
2324 bpo::options_description helpOptions;
2325 if (mode == "full" || mode == "short" || mode == "executor") {
2326 helpOptions.add(executorOptions);
2327 }
2328 // this time no veto is applied, so all the options are added for printout
2329 if (mode == "executor") {
2330 // nothing more
2331 } else if (mode == "workflow") {
2332 // executor options and workflow options, skip the actual workflow
2333 o2::framework::WorkflowSpec emptyWorkflow;
2334 helpOptions.add(ConfigParamsHelper::prepareOptionDescriptions(emptyWorkflow, currentWorkflowOptions));
2335 } else if (mode == "full" || mode == "short") {
2336 helpOptions.add(ConfigParamsHelper::prepareOptionDescriptions(physicalWorkflow, currentWorkflowOptions,
2337 bpo::options_description(),
2338 mode));
2339 } else {
2340 helpOptions.add(ConfigParamsHelper::prepareOptionDescriptions(physicalWorkflow, {},
2341 bpo::options_description(),
2342 mode));
2343 }
2344 if (helpOptions.options().size() == 0) {
2345 // the specified argument is invalid, add at leat the executor options
2346 mode += " is an invalid argument, please use correct argument for";
2347 helpOptions.add(executorOptions);
2348 }
2349 std::cout << "ALICE O2 DPL workflow driver" //
2350 << " (" << mode << " help)" << std::endl //
2351 << helpOptions << std::endl; //
2352}
2353
2354// Helper to find out if stdout is actually attached to a pipe.
2356{
2357 struct stat s;
2358 fstat(STDOUT_FILENO, &s);
2359 return ((s.st_mode & S_IFIFO) != 0);
2360}
2361
2363{
2364 struct stat s;
2365 int r = fstat(STDIN_FILENO, &s);
2366 // If stdin cannot be statted, we assume the shell is some sort of
2367 // non-interactive container thing
2368 if (r < 0) {
2369 return false;
2370 }
2371 // If stdin is a pipe or a file, we try to fetch configuration from there
2372 return ((s.st_mode & S_IFIFO) != 0 || (s.st_mode & S_IFREG) != 0);
2373}
2374
2376{
2377 struct CloningSpec {
2378 std::string templateMatcher;
2379 std::string cloneName;
2380 };
2381 auto s = ctx.options().get<std::string>("clone");
2382 std::vector<CloningSpec> specs;
2383 std::string delimiter = ",";
2384
2385 while (s.empty() == false) {
2386 auto newPos = s.find(delimiter);
2387 auto token = s.substr(0, newPos);
2388 auto split = token.find(":");
2389 if (split == std::string::npos) {
2390 throw std::runtime_error("bad clone definition. Syntax <template-processor>:<clone-name>");
2391 }
2392 auto key = token.substr(0, split);
2393 token.erase(0, split + 1);
2394 size_t error;
2395 std::string value = "";
2396 try {
2397 auto numValue = std::stoll(token, &error, 10);
2398 if (token[error] != '\0') {
2399 throw std::runtime_error("bad name for clone:" + token);
2400 }
2401 value = key + "_c" + std::to_string(numValue);
2402 } catch (std::invalid_argument& e) {
2403 value = token;
2404 }
2405 specs.push_back({key, value});
2406 s.erase(0, newPos + (newPos == std::string::npos ? 0 : 1));
2407 }
2408 if (s.empty() == false && specs.empty() == true) {
2409 throw std::runtime_error("bad pipeline definition. Syntax <processor>:<pipeline>");
2410 }
2411
2412 std::vector<DataProcessorSpec> extraSpecs;
2413 for (auto& spec : specs) {
2414 for (auto& processor : workflow) {
2415 if (processor.name == spec.templateMatcher) {
2416 auto clone = processor;
2417 clone.name = spec.cloneName;
2418 extraSpecs.push_back(clone);
2419 }
2420 }
2421 }
2422 workflow.insert(workflow.end(), extraSpecs.begin(), extraSpecs.end());
2423}
2424
2426{
2427 struct PipelineSpec {
2428 std::string matcher;
2429 int64_t pipeline;
2430 };
2431 auto s = ctx.options().get<std::string>("pipeline");
2432 std::vector<PipelineSpec> specs;
2433 std::string delimiter = ",";
2434
2435 while (s.empty() == false) {
2436 auto newPos = s.find(delimiter);
2437 auto token = s.substr(0, newPos);
2438 auto split = token.find(":");
2439 if (split == std::string::npos) {
2440 throw std::runtime_error("bad pipeline definition. Syntax <processor>:<pipeline>");
2441 }
2442 auto key = token.substr(0, split);
2443 token.erase(0, split + 1);
2444 size_t error;
2445 auto value = std::stoll(token, &error, 10);
2446 if (token[error] != '\0') {
2447 throw std::runtime_error("Bad pipeline definition. Expecting integer");
2448 }
2449 specs.push_back({key, value});
2450 s.erase(0, newPos + (newPos == std::string::npos ? 0 : 1));
2451 }
2452 if (s.empty() == false && specs.empty() == true) {
2453 throw std::runtime_error("bad pipeline definition. Syntax <processor>:<pipeline>");
2454 }
2455
2456 for (auto& spec : specs) {
2457 for (auto& processor : workflow) {
2458 if (processor.name == spec.matcher) {
2459 processor.maxInputTimeslices = spec.pipeline;
2460 }
2461 }
2462 }
2463}
2464
2466{
2467 struct LabelsSpec {
2468 std::string_view matcher;
2469 std::vector<std::string> labels;
2470 };
2471 std::vector<LabelsSpec> specs;
2472
2473 auto labelsString = ctx.options().get<std::string>("labels");
2474 if (labelsString.empty()) {
2475 return;
2476 }
2477 std::string_view sv{labelsString};
2478
2479 size_t specStart = 0;
2480 size_t specEnd = 0;
2481 constexpr char specDelim = ',';
2482 constexpr char labelDelim = ':';
2483 do {
2484 specEnd = sv.find(specDelim, specStart);
2485 auto token = sv.substr(specStart, specEnd == std::string_view::npos ? std::string_view::npos : specEnd - specStart);
2486 if (token.empty()) {
2487 throw std::runtime_error("bad labels definition. Syntax <processor>:<label>[:<label>][,<processor>:<label>[:<label>]");
2488 }
2489
2490 size_t labelDelimPos = token.find(labelDelim);
2491 if (labelDelimPos == 0 || labelDelimPos == std::string_view::npos) {
2492 throw std::runtime_error("bad labels definition. Syntax <processor>:<label>[:<label>][,<processor>:<label>[:<label>]");
2493 }
2494 LabelsSpec spec{.matcher = token.substr(0, labelDelimPos), .labels = {}};
2495
2496 size_t labelEnd = labelDelimPos + 1;
2497 do {
2498 size_t labelStart = labelDelimPos + 1;
2499 labelEnd = token.find(labelDelim, labelStart);
2500 auto label = labelEnd == std::string_view::npos ? token.substr(labelStart) : token.substr(labelStart, labelEnd - labelStart);
2501 if (label.empty()) {
2502 throw std::runtime_error("bad labels definition. Syntax <processor>:<label>[:<label>][,<processor>:<label>[:<label>]");
2503 }
2504 spec.labels.emplace_back(label);
2505 labelDelimPos = labelEnd;
2506 } while (labelEnd != std::string_view::npos);
2507
2508 specs.push_back(spec);
2509 specStart = specEnd + 1;
2510 } while (specEnd != std::string_view::npos);
2511
2512 if (labelsString.empty() == false && specs.empty() == true) {
2513 throw std::runtime_error("bad labels definition. Syntax <processor>:<label>[:<label>][,<processor>:<label>[:<label>]");
2514 }
2515
2516 for (auto& spec : specs) {
2517 for (auto& processor : workflow) {
2518 if (processor.name == spec.matcher) {
2519 for (const auto& label : spec.labels) {
2520 if (std::find_if(processor.labels.begin(), processor.labels.end(),
2521 [label](const auto& procLabel) { return procLabel.value == label; }) == processor.labels.end()) {
2522 processor.labels.push_back({label});
2523 }
2524 }
2525 }
2526 }
2527 }
2528}
2529
2531void initialiseDriverControl(bpo::variables_map const& varmap,
2532 DriverInfo& driverInfo,
2533 DriverControl& control)
2534{
2535 // Control is initialised outside the main loop because
2536 // command line options are really affecting control.
2537 control.defaultQuiet = varmap["quiet"].as<bool>();
2538 control.defaultStopped = varmap["stop"].as<bool>();
2539
2540 if (varmap["single-step"].as<bool>()) {
2541 control.state = DriverControlState::STEP;
2542 } else {
2543 control.state = DriverControlState::PLAY;
2544 }
2545
2546 if (varmap["graphviz"].as<bool>()) {
2547 // Dump a graphviz representation of what I will do.
2548 control.callbacks = {[](WorkflowSpec const&,
2549 DeviceSpecs const& specs,
2550 DeviceExecutions const&,
2552 CommandInfo const&) {
2554 }};
2555 control.forcedTransitions = {
2556 DriverState::EXIT, //
2557 DriverState::PERFORM_CALLBACKS, //
2558 DriverState::MERGE_CONFIGS, //
2559 DriverState::IMPORT_CURRENT_WORKFLOW, //
2560 DriverState::MATERIALISE_WORKFLOW //
2561 };
2562 } else if (!varmap["dds"].as<std::string>().empty()) {
2563 // Dump a DDS representation of what I will do.
2564 // Notice that compared to DDS we need to schedule things,
2565 // because DDS needs to be able to have actual Executions in
2566 // order to provide a correct configuration.
2567 control.callbacks = {[filename = varmap["dds"].as<std::string>(),
2568 workflowSuffix = varmap["dds-workflow-suffix"],
2569 driverMode = driverInfo.mode](WorkflowSpec const& workflow,
2570 DeviceSpecs const& specs,
2571 DeviceExecutions const& executions,
2572 DataProcessorInfos& dataProcessorInfos,
2573 CommandInfo const& commandInfo) {
2574 if (filename == "-") {
2575 DDSConfigHelpers::dumpDeviceSpec2DDS(std::cout, driverMode, workflowSuffix.as<std::string>(), workflow, dataProcessorInfos, specs, executions, commandInfo);
2576 } else {
2577 std::ofstream out(filename);
2578 DDSConfigHelpers::dumpDeviceSpec2DDS(out, driverMode, workflowSuffix.as<std::string>(), workflow, dataProcessorInfos, specs, executions, commandInfo);
2579 }
2580 }};
2581 control.forcedTransitions = {
2582 DriverState::EXIT, //
2583 DriverState::PERFORM_CALLBACKS, //
2584 DriverState::MERGE_CONFIGS, //
2585 DriverState::IMPORT_CURRENT_WORKFLOW, //
2586 DriverState::MATERIALISE_WORKFLOW //
2587 };
2588 } else if (!varmap["o2-control"].as<std::string>().empty() or !varmap["mermaid"].as<std::string>().empty()) {
2589 // Dump the workflow in o2-control and/or mermaid format
2590 control.callbacks = {[filename = varmap["mermaid"].as<std::string>(),
2591 workflowName = varmap["o2-control"].as<std::string>()](WorkflowSpec const&,
2592 DeviceSpecs const& specs,
2593 DeviceExecutions const& executions,
2595 CommandInfo const& commandInfo) {
2596 if (!workflowName.empty()) {
2597 dumpDeviceSpec2O2Control(workflowName, specs, executions, commandInfo);
2598 }
2599 if (!filename.empty()) {
2600 if (filename == "-") {
2602 } else {
2603 std::ofstream output(filename);
2605 }
2606 }
2607 }};
2608 control.forcedTransitions = {
2609 DriverState::EXIT, //
2610 DriverState::PERFORM_CALLBACKS, //
2611 DriverState::MERGE_CONFIGS, //
2612 DriverState::IMPORT_CURRENT_WORKFLOW, //
2613 DriverState::MATERIALISE_WORKFLOW //
2614 };
2615
2616 } else if (varmap.count("id")) {
2617 // Add our own stacktrace dumping
2618 if (getenv("O2_NO_CATCHALL_EXCEPTIONS") != nullptr && strcmp(getenv("O2_NO_CATCHALL_EXCEPTIONS"), "0") != 0) {
2619 LOGP(info, "Not instrumenting crash signals because O2_NO_CATCHALL_EXCEPTIONS is set");
2620 gEnv->SetValue("Root.Stacktrace", "no");
2621 gSystem->ResetSignal(kSigSegmentationViolation, kTRUE);
2622 rlimit limit;
2623 if (getrlimit(RLIMIT_CORE, &limit) == 0) {
2624 LOGP(info, "Core limit: {} {}", limit.rlim_cur, limit.rlim_max);
2625 }
2626 }
2627 if (varmap["stacktrace-on-signal"].as<std::string>() == "simple" && (getenv("O2_NO_CATCHALL_EXCEPTIONS") == nullptr || strcmp(getenv("O2_NO_CATCHALL_EXCEPTIONS"), "0") == 0)) {
2628 LOGP(info, "Instrumenting crash signals");
2629 signal(SIGSEGV, handle_crash);
2630 signal(SIGABRT, handle_crash);
2631 signal(SIGBUS, handle_crash);
2632 signal(SIGILL, handle_crash);
2633 signal(SIGFPE, handle_crash);
2634 }
2635 // FIXME: for the time being each child needs to recalculate the workflow,
2636 // so that it can understand what it needs to do. This is obviously
2637 // a bad idea. In the future we should have the client be pushed
2638 // it's own configuration by the driver.
2639 control.forcedTransitions = {
2640 DriverState::DO_CHILD, //
2641 DriverState::BIND_GUI_PORT, //
2642 DriverState::MERGE_CONFIGS, //
2643 DriverState::IMPORT_CURRENT_WORKFLOW, //
2644 DriverState::MATERIALISE_WORKFLOW //
2645 };
2646 } else if ((varmap["dump-workflow"].as<bool>() == true) || (varmap["run"].as<bool>() == false && varmap.count("id") == 0 && isOutputToPipe())) {
2647 control.callbacks = {[filename = varmap["dump-workflow-file"].as<std::string>()](WorkflowSpec const& workflow,
2648 DeviceSpecs const&,
2649 DeviceExecutions const&,
2650 DataProcessorInfos& dataProcessorInfos,
2651 CommandInfo const& commandInfo) {
2652 if (filename == "-") {
2653 WorkflowSerializationHelpers::dump(std::cout, workflow, dataProcessorInfos, commandInfo);
2654 // FIXME: this is to avoid trailing garbage..
2655 exit(0);
2656 } else {
2657 std::ofstream output(filename);
2658 WorkflowSerializationHelpers::dump(output, workflow, dataProcessorInfos, commandInfo);
2659 }
2660 }};
2661 control.forcedTransitions = {
2662 DriverState::EXIT, //
2663 DriverState::PERFORM_CALLBACKS, //
2664 DriverState::MERGE_CONFIGS, //
2665 DriverState::IMPORT_CURRENT_WORKFLOW, //
2666 DriverState::MATERIALISE_WORKFLOW //
2667 };
2668 } else {
2669 // By default we simply start the main loop of the driver.
2670 control.forcedTransitions = {
2671 DriverState::INIT, //
2672 DriverState::BIND_GUI_PORT, //
2673 DriverState::IMPORT_CURRENT_WORKFLOW, //
2674 DriverState::MATERIALISE_WORKFLOW //
2675 };
2676 }
2677}
2678
2680void conflicting_options(const boost::program_options::variables_map& vm,
2681 const std::string& opt1, const std::string& opt2)
2682{
2683 if (vm.count(opt1) && !vm[opt1].defaulted() &&
2684 vm.count(opt2) && !vm[opt2].defaulted()) {
2685 throw std::logic_error(std::string("Conflicting options '") +
2686 opt1 + "' and '" + opt2 + "'.");
2687 }
2688}
2689
2690template <typename T>
2692 std::vector<T>& v,
2693 std::vector<int>& indices)
2694{
2695 using std::swap; // to permit Koenig lookup
2696 for (int i = 0; i < (int)indices.size(); i++) {
2697 auto current = i;
2698 while (i != indices[current]) {
2699 auto next = indices[current];
2700 swap(v[current], v[next]);
2701 indices[current] = current;
2702 current = next;
2703 }
2704 indices[current] = current;
2705 }
2706}
2707
2708// Check if the workflow is resiliant to failures
2709void checkNonResiliency(std::vector<DataProcessorSpec> const& specs,
2710 std::vector<std::pair<int, int>> const& edges)
2711{
2712 auto checkExpendable = [](DataProcessorLabel const& label) {
2713 return label.value == "expendable";
2714 };
2715 auto checkResilient = [](DataProcessorLabel const& label) {
2716 return label.value == "resilient" || label.value == "expendable";
2717 };
2718
2719 for (auto& edge : edges) {
2720 auto& src = specs[edge.first];
2721 auto& dst = specs[edge.second];
2722 if (std::none_of(src.labels.begin(), src.labels.end(), checkExpendable)) {
2723 continue;
2724 }
2725 if (std::any_of(dst.labels.begin(), dst.labels.end(), checkResilient)) {
2726 continue;
2727 }
2728 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.");
2729 }
2730}
2731
2732std::string debugTopoInfo(std::vector<DataProcessorSpec> const& specs,
2733 std::vector<TopoIndexInfo> const& infos,
2734 std::vector<std::pair<int, int>> const& edges)
2735{
2736 std::ostringstream out;
2737
2738 out << "\nTopological info:\n";
2739 for (auto& ti : infos) {
2740 out << specs[ti.index].name << " (index: " << ti.index << ", layer: " << ti.layer << ")\n";
2741 out << " Inputs:\n";
2742 for (auto& ii : specs[ti.index].inputs) {
2743 out << " - " << DataSpecUtils::describe(ii) << "\n";
2744 }
2745 out << "\n Outputs:\n";
2746 for (auto& ii : specs[ti.index].outputs) {
2747 out << " - " << DataSpecUtils::describe(ii) << "\n";
2748 }
2749 }
2750 out << "\nEdges values:\n";
2751 for (auto& e : edges) {
2752 out << specs[e.second].name << " depends on " << specs[e.first].name << "\n";
2753 }
2754 for (auto& d : specs) {
2755 out << "- " << d.name << std::endl;
2756 }
2758 return out.str();
2759}
2760
2761void enableSignposts(std::string const& signpostsToEnable)
2762{
2763 static pid_t pid = getpid();
2764 if (signpostsToEnable.empty() == true) {
2765 auto printAllSignposts = [](char const* name, void* l, void* context) {
2766 auto* log = (_o2_log_t*)l;
2767 LOGP(detail, "Signpost stream {} disabled. Enable it with o2-log -p {} -a {}", name, pid, (void*)&log->stacktrace);
2768 return true;
2769 };
2770 o2_walk_logs(printAllSignposts, nullptr);
2771 return;
2772 }
2773 auto matchingLogEnabler = [](char const* name, void* l, void* context) {
2774 auto* log = (_o2_log_t*)l;
2775 auto* selectedName = (char const*)context;
2776 std::string prefix = "ch.cern.aliceo2.";
2777 auto* last = strchr(selectedName, ':');
2778 int maxDepth = 1;
2779 if (last) {
2780 char* err;
2781 maxDepth = strtol(last + 1, &err, 10);
2782 if (*(last + 1) == '\0' || *err != '\0') {
2783 maxDepth = 1;
2784 }
2785 }
2786
2787 auto fullName = prefix + std::string{selectedName, last ? last - selectedName : strlen(selectedName)};
2788 if (fullName == name) {
2789 LOGP(info, "Enabling signposts for stream \"{}\" with depth {}.", fullName, maxDepth);
2790 _o2_log_set_stacktrace(log, maxDepth);
2791 return false;
2792 } else {
2793 LOGP(info, "Signpost stream \"{}\" disabled. Enable it with o2-log -p {} -a {}", name, pid, (void*)&log->stacktrace);
2794 }
2795 return true;
2796 };
2797 // Split signpostsToEnable by comma using strtok_r
2798 char* saveptr;
2799 char* src = const_cast<char*>(signpostsToEnable.data());
2800 auto* token = strtok_r(src, ",", &saveptr);
2801 while (token) {
2802 o2_walk_logs(matchingLogEnabler, token);
2803 token = strtok_r(nullptr, ",", &saveptr);
2804 }
2805}
2806
2807void overrideAll(o2::framework::ConfigContext& ctx, std::vector<o2::framework::DataProcessorSpec>& workflow)
2808{
2809 overrideCloning(ctx, workflow);
2810 overridePipeline(ctx, workflow);
2811 overrideLabels(ctx, workflow);
2812}
2813
2814o2::framework::ConfigContext createConfigContext(std::unique_ptr<ConfigParamRegistry>& workflowOptionsRegistry,
2815 o2::framework::ServiceRegistry& configRegistry,
2816 std::vector<o2::framework::ConfigParamSpec>& workflowOptions,
2817 std::vector<o2::framework::ConfigParamSpec>& extraOptions, int argc, char** argv)
2818{
2819 std::vector<std::unique_ptr<o2::framework::ParamRetriever>> retrievers;
2820 std::unique_ptr<o2::framework::ParamRetriever> retriever{new o2::framework::BoostOptionsRetriever(true, argc, argv)};
2821 retrievers.emplace_back(std::move(retriever));
2822 auto workflowOptionsStore = std::make_unique<o2::framework::ConfigParamStore>(workflowOptions, std::move(retrievers));
2823 workflowOptionsStore->preload();
2824 workflowOptionsStore->activate();
2825 workflowOptionsRegistry = std::make_unique<ConfigParamRegistry>(std::move(workflowOptionsStore));
2826 extraOptions = o2::framework::ConfigParamDiscovery::discover(*workflowOptionsRegistry, argc, argv);
2827 for (auto& extra : extraOptions) {
2828 workflowOptions.push_back(extra);
2829 }
2830
2831 return o2::framework::ConfigContext(*workflowOptionsRegistry, o2::framework::ServiceRegistryRef{configRegistry}, argc, argv);
2832}
2833
2834std::unique_ptr<o2::framework::ServiceRegistry> createRegistry()
2835{
2836 return std::make_unique<o2::framework::ServiceRegistry>();
2837}
2838
2839// This is a toy executor for the workflow spec
2840// What it needs to do is:
2841//
2842// - Print the properties of each DataProcessorSpec
2843// - Fork one process per DataProcessorSpec
2844// - Parent -> wait for all the children to complete (eventually
2845// killing them all on ctrl-c).
2846// - Child, pick the data-processor ID and start a O2DataProcessorDevice for
2847// each DataProcessorSpec
2848int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& workflow,
2849 std::vector<ChannelConfigurationPolicy> const& channelPolicies,
2850 std::vector<CompletionPolicy> const& completionPolicies,
2851 std::vector<DispatchPolicy> const& dispatchPolicies,
2852 std::vector<ResourcePolicy> const& resourcePolicies,
2853 std::vector<CallbacksPolicy> const& callbacksPolicies,
2854 std::vector<SendingPolicy> const& sendingPolicies,
2855 std::vector<ConfigParamSpec> const& currentWorkflowOptions,
2856 std::vector<ConfigParamSpec> const& detectedParams,
2857 o2::framework::ConfigContext& configContext)
2858{
2859 // Peek very early in the driver options and look for
2860 // signposts, so the we can enable it without going through the whole dance
2861 if (getenv("DPL_DRIVER_SIGNPOSTS")) {
2862 enableSignposts(getenv("DPL_DRIVER_SIGNPOSTS"));
2863 }
2864
2865 std::vector<std::string> currentArgs;
2866 std::vector<PluginInfo> plugins;
2867 std::vector<ForwardingPolicy> forwardingPolicies = ForwardingPolicy::createDefaultPolicies();
2868
2869 for (int ai = 1; ai < argc; ++ai) {
2870 currentArgs.emplace_back(argv[ai]);
2871 }
2872
2873 WorkflowInfo currentWorkflow{
2874 argv[0],
2875 currentArgs,
2876 currentWorkflowOptions};
2877
2878 ProcessingPolicies processingPolicies;
2879 enum LogParsingHelpers::LogLevel minFailureLevel;
2880 bpo::options_description executorOptions("Executor options");
2881 const char* helpDescription = "print help: short, full, executor, or processor name";
2882 enum DriverMode driverMode;
2883 executorOptions.add_options() //
2884 ("help,h", bpo::value<std::string>()->implicit_value("short"), helpDescription) // //
2885 ("quiet,q", bpo::value<bool>()->zero_tokens()->default_value(false), "quiet operation") // //
2886 ("stop,s", bpo::value<bool>()->zero_tokens()->default_value(false), "stop before device start") // //
2887 ("single-step", bpo::value<bool>()->zero_tokens()->default_value(false), "start in single step mode") // //
2888 ("batch,b", bpo::value<std::vector<std::string>>()->zero_tokens()->composing(), "batch processing mode") // //
2889 ("no-batch", bpo::value<bool>()->zero_tokens(), "force gui processing mode") // //
2890 ("no-cleanup", bpo::value<bool>()->zero_tokens()->default_value(false), "do not cleanup the shm segment") // //
2891 ("hostname", bpo::value<std::string>()->default_value("localhost"), "hostname to deploy") // //
2892 ("resources", bpo::value<std::string>()->default_value(""), "resources allocated for the workflow") // //
2893 ("start-port,p", bpo::value<unsigned short>()->default_value(22000), "start port to allocate") // //
2894 ("port-range,pr", bpo::value<unsigned short>()->default_value(1000), "ports in range") // //
2895 ("completion-policy,c", bpo::value<TerminationPolicy>(&processingPolicies.termination)->default_value(TerminationPolicy::QUIT), // //
2896 "what to do when processing is finished: quit, wait") // //
2897 ("error-policy", bpo::value<TerminationPolicy>(&processingPolicies.error)->default_value(TerminationPolicy::QUIT), // //
2898 "what to do when a device has an error: quit, wait") // //
2899 ("min-failure-level", bpo::value<LogParsingHelpers::LogLevel>(&minFailureLevel)->default_value(LogParsingHelpers::LogLevel::Fatal), // //
2900 "minimum message level which will be considered as fatal and exit with 1") // //
2901 ("graphviz,g", bpo::value<bool>()->zero_tokens()->default_value(false), "produce graphviz output") // //
2902 ("mermaid", bpo::value<std::string>()->default_value(""), "produce graph output in mermaid format in file under specified name or on stdout if argument is \"-\"") // //
2903 ("timeout,t", bpo::value<uint64_t>()->default_value(0), "forced exit timeout (in seconds)") // //
2904 ("dds,D", bpo::value<std::string>()->default_value(""), "create DDS configuration") // //
2905 ("dds-workflow-suffix,D", bpo::value<std::string>()->default_value(""), "suffix for DDS names") // //
2906 ("dump-workflow,dump", bpo::value<bool>()->zero_tokens()->default_value(false), "dump workflow as JSON") // //
2907 ("dump-workflow-file", bpo::value<std::string>()->default_value("-"), "file to which do the dump") // //
2908 ("driver-mode", bpo::value<DriverMode>(&driverMode)->default_value(DriverMode::STANDALONE), R"(how to run the driver. default: "standalone". Valid: "embedded")") // //
2909 ("run", bpo::value<bool>()->zero_tokens()->default_value(false), "run workflow merged so far. It implies --batch. Use --no-batch to see the GUI") // //
2910 ("no-IPC", bpo::value<bool>()->zero_tokens()->default_value(false), "disable IPC topology optimization") // //
2911 ("o2-control,o2", bpo::value<std::string>()->default_value(""), "dump O2 Control workflow configuration under the specified name") //
2912 ("resources-monitoring", bpo::value<unsigned short>()->default_value(0), "enable cpu/memory monitoring for provided interval in seconds") //
2913 ("resources-monitoring-dump-interval", bpo::value<unsigned short>()->default_value(0), "dump monitoring information to disk every provided seconds"); //
2914 // some of the options must be forwarded by default to the device
2915 executorOptions.add(DeviceSpecHelpers::getForwardedDeviceOptions());
2916
2917 gHiddenDeviceOptions.add_options() //
2918 ("id,i", bpo::value<std::string>(), "device id for child spawning") //
2919 ("channel-config", bpo::value<std::vector<std::string>>(), "channel configuration") //
2920 ("control", "control plugin") //
2921 ("log-color", "logging color scheme")("color", "logging color scheme");
2922
2923 bpo::options_description visibleOptions;
2924 visibleOptions.add(executorOptions);
2925
2926 auto physicalWorkflow = workflow;
2927 std::map<std::string, size_t> rankIndex;
2928 // We remove the duplicates because for the moment child get themself twice:
2929 // once from the actual definition in the child, a second time from the
2930 // configuration they get passed by their parents.
2931 // Notice that we do not know in which order we will get the workflows, so
2932 // while we keep the order of DataProcessors we reshuffle them based on
2933 // some hopefully unique hash.
2934 size_t workflowHashA = 0;
2935 std::hash<std::string> hash_fn;
2936
2937 for (auto& dp : workflow) {
2938 workflowHashA += hash_fn(dp.name);
2939 }
2940
2941 for (auto& dp : workflow) {
2942 rankIndex.insert(std::make_pair(dp.name, workflowHashA));
2943 }
2944
2945 std::vector<DataProcessorInfo> dataProcessorInfos;
2946 CommandInfo commandInfo{};
2947
2948 if (isatty(STDIN_FILENO) == false && isInputConfig()) {
2949 std::vector<DataProcessorSpec> importedWorkflow;
2950 bool previousWorked = WorkflowSerializationHelpers::import(std::cin, importedWorkflow, dataProcessorInfos, commandInfo);
2951 if (previousWorked == false) {
2952 exit(1);
2953 }
2954
2955 size_t workflowHashB = 0;
2956 for (auto& dp : importedWorkflow) {
2957 workflowHashB += hash_fn(dp.name);
2958 }
2959
2960 // FIXME: Streamline...
2961 // We remove the duplicates because for the moment child get themself twice:
2962 // once from the actual definition in the child, a second time from the
2963 // configuration they get passed by their parents.
2964 for (auto& dp : importedWorkflow) {
2965 auto found = std::find_if(physicalWorkflow.begin(), physicalWorkflow.end(),
2966 [&name = dp.name](DataProcessorSpec const& spec) { return spec.name == name; });
2967 if (found == physicalWorkflow.end()) {
2968 physicalWorkflow.push_back(dp);
2969 rankIndex.insert(std::make_pair(dp.name, workflowHashB));
2970 }
2971 }
2972 }
2973
2978 for (auto& dp : physicalWorkflow) {
2979 auto isExpendable = [](DataProcessorLabel const& label) { return label.value == "expendable" || label.value == "non-critical"; };
2980 if (std::find_if(dp.labels.begin(), dp.labels.end(), isExpendable) != dp.labels.end()) {
2981 for (auto& output : dp.outputs) {
2982 if (output.lifetime == Lifetime::Timeframe) {
2983 output.lifetime = Lifetime::Sporadic;
2984 }
2985 }
2986 }
2987 }
2988
2990 OverrideServiceSpecs driverServicesOverride = ServiceSpecHelpers::parseOverrides(getenv("DPL_DRIVER_OVERRIDE_SERVICES"));
2992 // We insert the hash for the internal devices.
2993 WorkflowHelpers::injectServiceDevices(physicalWorkflow, configContext);
2994 auto reader = std::find_if(physicalWorkflow.begin(), physicalWorkflow.end(), [](DataProcessorSpec& spec) { return spec.name == "internal-dpl-aod-reader"; });
2995 if (reader != physicalWorkflow.end()) {
2996 driverServices.push_back(ArrowSupport::arrowBackendSpec());
2997 }
2998 for (auto& service : driverServices) {
2999 if (service.injectTopology == nullptr) {
3000 continue;
3001 }
3002 WorkflowSpecNode node{physicalWorkflow};
3003 service.injectTopology(node, configContext);
3004 }
3005 for (auto& dp : physicalWorkflow) {
3006 if (dp.name.rfind("internal-", 0) == 0) {
3007 rankIndex.insert(std::make_pair(dp.name, hash_fn("internal")));
3008 }
3009 }
3010
3011 // We sort dataprocessors and Inputs / outputs by name, so that the edges are
3012 // always in the same order.
3013 std::stable_sort(physicalWorkflow.begin(), physicalWorkflow.end(), [](DataProcessorSpec const& a, DataProcessorSpec const& b) {
3014 return a.name < b.name;
3015 });
3016
3017 for (auto& dp : physicalWorkflow) {
3018 std::stable_sort(dp.inputs.begin(), dp.inputs.end(),
3019 [](InputSpec const& a, InputSpec const& b) { return DataSpecUtils::describe(a) < DataSpecUtils::describe(b); });
3020 std::stable_sort(dp.outputs.begin(), dp.outputs.end(),
3021 [](OutputSpec const& a, OutputSpec const& b) { return DataSpecUtils::describe(a) < DataSpecUtils::describe(b); });
3022 }
3023
3024 // Create a list of all the edges, so that we can do a topological sort
3025 // before we create the graph.
3026 std::vector<std::pair<int, int>> edges;
3027
3028 if (physicalWorkflow.size() > 1) {
3029 edges = TopologyPolicyHelpers::buildEdges(physicalWorkflow);
3030
3031 auto topoInfos = WorkflowHelpers::topologicalSort(physicalWorkflow.size(), &edges[0].first, &edges[0].second, sizeof(std::pair<int, int>), edges.size());
3032 if (topoInfos.size() != physicalWorkflow.size()) {
3033 // Check missing resilincy of one of the tasks
3034 checkNonResiliency(physicalWorkflow, edges);
3035 throw std::runtime_error("Unable to do topological sort of the resulting workflow. Do you have loops?\n" + debugTopoInfo(physicalWorkflow, topoInfos, edges));
3036 }
3037 // Sort by layer and then by name, to ensure stability.
3038 std::stable_sort(topoInfos.begin(), topoInfos.end(), [&workflow = physicalWorkflow](TopoIndexInfo const& a, TopoIndexInfo const& b) {
3039 auto aRank = std::make_tuple(a.layer, -workflow.at(a.index).outputs.size(), workflow.at(a.index).name);
3040 auto bRank = std::make_tuple(b.layer, -workflow.at(b.index).outputs.size(), workflow.at(b.index).name);
3041 return aRank < bRank;
3042 });
3043 // Reverse index and apply the result
3044 std::vector<int> dataProcessorOrder;
3045 dataProcessorOrder.resize(topoInfos.size());
3046 for (size_t i = 0; i < topoInfos.size(); ++i) {
3047 dataProcessorOrder[topoInfos[i].index] = i;
3048 }
3049 std::vector<int> newLocations;
3050 newLocations.resize(dataProcessorOrder.size());
3051 for (size_t i = 0; i < dataProcessorOrder.size(); ++i) {
3052 newLocations[dataProcessorOrder[i]] = i;
3053 }
3054 apply_permutation(physicalWorkflow, newLocations);
3055 }
3056
3057 // Use the hidden options as veto, all config specs matching a definition
3058 // in the hidden options are skipped in order to avoid duplicate definitions
3059 // in the main parser. Note: all config specs are forwarded to devices
3060 visibleOptions.add(ConfigParamsHelper::prepareOptionDescriptions(physicalWorkflow, currentWorkflowOptions, gHiddenDeviceOptions));
3061
3062 bpo::options_description od;
3063 od.add(visibleOptions);
3064 od.add(gHiddenDeviceOptions);
3065
3066 // FIXME: decide about the policy for handling unrecognized arguments
3067 // command_line_parser with option allow_unregistered() can be used
3068 using namespace bpo::command_line_style;
3069 auto style = (allow_short | short_allow_adjacent | short_allow_next | allow_long | long_allow_adjacent | long_allow_next | allow_sticky | allow_dash_for_short);
3070 bpo::variables_map varmap;
3071 try {
3072 bpo::store(
3073 bpo::command_line_parser(argc, argv)
3074 .options(od)
3075 .style(style)
3076 .run(),
3077 varmap);
3078 } catch (std::exception const& e) {
3079 LOGP(error, "error parsing options of {}: {}", argv[0], e.what());
3080 exit(1);
3081 }
3082 conflicting_options(varmap, "dds", "o2-control");
3083 conflicting_options(varmap, "dds", "dump-workflow");
3084 conflicting_options(varmap, "dds", "run");
3085 conflicting_options(varmap, "dds", "graphviz");
3086 conflicting_options(varmap, "o2-control", "dump-workflow");
3087 conflicting_options(varmap, "o2-control", "run");
3088 conflicting_options(varmap, "o2-control", "graphviz");
3089 conflicting_options(varmap, "run", "dump-workflow");
3090 conflicting_options(varmap, "run", "graphviz");
3091 conflicting_options(varmap, "run", "mermaid");
3092 conflicting_options(varmap, "dump-workflow", "graphviz");
3093 conflicting_options(varmap, "no-batch", "batch");
3094
3095 if (varmap.count("help")) {
3096 printHelp(varmap, executorOptions, physicalWorkflow, currentWorkflowOptions);
3097 exit(0);
3098 }
3102 if (varmap.count("severity")) {
3103 auto logLevel = varmap["severity"].as<std::string>();
3104 if (logLevel == "debug") {
3105 fair::Logger::SetConsoleSeverity(fair::Severity::debug);
3106 } else if (logLevel == "detail") {
3107 fair::Logger::SetConsoleSeverity(fair::Severity::detail);
3108 } else if (logLevel == "info") {
3109 fair::Logger::SetConsoleSeverity(fair::Severity::info);
3110 } else if (logLevel == "warning") {
3111 fair::Logger::SetConsoleSeverity(fair::Severity::warning);
3112 } else if (logLevel == "error") {
3113 fair::Logger::SetConsoleSeverity(fair::Severity::error);
3114 } else if (logLevel == "important") {
3115 fair::Logger::SetConsoleSeverity(fair::Severity::important);
3116 } else if (logLevel == "alarm") {
3117 fair::Logger::SetConsoleSeverity(fair::Severity::alarm);
3118 } else if (logLevel == "critical") {
3119 fair::Logger::SetConsoleSeverity(fair::Severity::critical);
3120 } else if (logLevel == "fatal") {
3121 fair::Logger::SetConsoleSeverity(fair::Severity::fatal);
3122 } else {
3123 LOGP(error, "Invalid log level '{}'", logLevel);
3124 exit(1);
3125 }
3126 }
3127
3128 enableSignposts(varmap["signposts"].as<std::string>());
3129
3130 auto evaluateBatchOption = [&varmap]() -> bool {
3131 if (varmap.count("no-batch") > 0) {
3132 return false;
3133 }
3134 if (varmap.count("batch") == 0) {
3135 // default value
3136 return isatty(fileno(stdout)) == 0;
3137 }
3138 // FIXME: should actually use the last value, but for some reason the
3139 // values are not filled into the vector, even if specifying `-b true`
3140 // need to find out why the boost program options example is not working
3141 // in our case. Might depend on the parser options
3142 // auto value = varmap["batch"].as<std::vector<std::string>>();
3143 return true;
3144 };
3145 DriverInfo driverInfo{
3146 .sendingPolicies = sendingPolicies,
3147 .forwardingPolicies = forwardingPolicies,
3148 .callbacksPolicies = callbacksPolicies};
3149 driverInfo.states.reserve(10);
3150 driverInfo.sigintRequested = false;
3151 driverInfo.sigchldRequested = false;
3152 driverInfo.channelPolicies = channelPolicies;
3153 driverInfo.completionPolicies = completionPolicies;
3154 driverInfo.dispatchPolicies = dispatchPolicies;
3155 driverInfo.resourcePolicies = resourcePolicies;
3156 driverInfo.argc = argc;
3157 driverInfo.argv = argv;
3158 driverInfo.noSHMCleanup = varmap["no-cleanup"].as<bool>();
3159 driverInfo.processingPolicies.termination = varmap["completion-policy"].as<TerminationPolicy>();
3160 driverInfo.processingPolicies.earlyForward = varmap["early-forward-policy"].as<EarlyForwardPolicy>();
3161 driverInfo.mode = varmap["driver-mode"].as<DriverMode>();
3162
3163 auto batch = evaluateBatchOption();
3164 DriverConfig driverConfig{
3165 .batch = batch,
3166 .driverHasGUI = (batch == false) || getenv("DPL_DRIVER_REMOTE_GUI") != nullptr,
3167 };
3168
3169 if (varmap["error-policy"].defaulted() && driverConfig.batch == false) {
3170 driverInfo.processingPolicies.error = TerminationPolicy::WAIT;
3171 } else {
3172 driverInfo.processingPolicies.error = varmap["error-policy"].as<TerminationPolicy>();
3173 }
3174 driverInfo.minFailureLevel = varmap["min-failure-level"].as<LogParsingHelpers::LogLevel>();
3175 driverInfo.startTime = uv_hrtime();
3176 driverInfo.startTimeMsFromEpoch = std::chrono::duration_cast<std::chrono::milliseconds>(
3177 std::chrono::system_clock::now().time_since_epoch())
3178 .count();
3179 driverInfo.timeout = varmap["timeout"].as<uint64_t>();
3180 driverInfo.deployHostname = varmap["hostname"].as<std::string>();
3181 driverInfo.resources = varmap["resources"].as<std::string>();
3182 driverInfo.resourcesMonitoringInterval = varmap["resources-monitoring"].as<unsigned short>();
3183 driverInfo.resourcesMonitoringDumpInterval = varmap["resources-monitoring-dump-interval"].as<unsigned short>();
3184
3185 // FIXME: should use the whole dataProcessorInfos, actually...
3186 driverInfo.processorInfo = dataProcessorInfos;
3187 driverInfo.configContext = &configContext;
3188
3189 DriverControl driverControl;
3190 initialiseDriverControl(varmap, driverInfo, driverControl);
3191
3192 commandInfo.merge(CommandInfo(argc, argv));
3193
3194 std::string frameworkId;
3195 // If the id is set, this means this is a device,
3196 // otherwise this is the driver.
3197 if (varmap.count("id")) {
3198 // The framework id does not want to know anything about DDS template expansion
3199 // so we simply drop it. Notice that the "id" Property is still the same as the
3200 // original --id option.
3201 frameworkId = std::regex_replace(varmap["id"].as<std::string>(), std::regex{"_dds.*"}, "");
3202 driverInfo.uniqueWorkflowId = fmt::format("{}", getppid());
3203 driverInfo.defaultDriverClient = "stdout://";
3204 } else {
3205 driverInfo.uniqueWorkflowId = fmt::format("{}", getpid());
3206 driverInfo.defaultDriverClient = "ws://";
3207 }
3208 return runStateMachine(physicalWorkflow,
3209 currentWorkflow,
3210 dataProcessorInfos,
3211 commandInfo,
3212 driverControl,
3213 driverInfo,
3214 driverConfig,
3216 detectedParams,
3217 varmap,
3218 driverServices,
3219 frameworkId);
3220}
3221
3222void doBoostException(boost::exception&, char const* processName)
3223{
3224 LOGP(error, "error while setting up workflow in {}: {}",
3225 processName, boost::current_exception_diagnostic_information(true));
3226}
3227#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
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:547
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:483
#define O2_SIGNPOST_ID_FROM_POINTER(name, log, pointer)
Definition Signpost.h:499
#define O2_SIGNPOST_END(log, id, name, format,...)
Definition Signpost.h:602
void _o2_log_set_stacktrace(_o2_log_t *log, int stacktrace)
#define O2_SIGNPOST_ID_GENERATE(name, log)
Definition Signpost.h:500
#define O2_SIGNPOST_EVENT_EMIT(log, id, name, format,...)
Definition Signpost.h:516
#define O2_SIGNPOST_START(log, id, name, format,...)
Definition Signpost.h:596
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