1357 DriverInfo& driverInfo,
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)
1367 .shmSegmentId = (int16_t)atoi(varmap[
"shm-segment-id"].as<std::string>().c_str())};
1371 auto* devicesManager =
new DevicesManager{.
controls = controls, .infos = infos, .specs = runningWorkflow.devices, .messages = {}};
1375 std::vector<uv_poll_t*> pollHandles;
1376 std::vector<DeviceStdioContext> childFds;
1378 std::vector<ComputingResource> resources;
1380 if (driverInfo.resources !=
"") {
1386 auto resourceManager = std::make_unique<SimpleResourceManager>(resources);
1389 void* window =
nullptr;
1390 decltype(debugGUI->
getGUIDebugger(infos, runningWorkflow.devices, allStates, dataProcessorInfos, metricsInfos, driverInfo, controls, driverControl)) debugGUICallback;
1393 auto initDebugGUI = []() ->
DebugGUI* {
1394 uv_lib_t supportLib;
1397 result = uv_dlopen(
"libO2FrameworkGUISupport.dylib", &supportLib);
1399 result = uv_dlopen(
"libO2FrameworkGUISupport.so", &supportLib);
1402 LOG(error) << uv_dlerror(&supportLib);
1407 result = uv_dlsym(&supportLib,
"dpl_plugin_callback", (
void**)&dpl_plugin_callback);
1409 LOG(error) << uv_dlerror(&supportLib);
1413 return PluginManager::getByName<DebugGUI>(pluginInstance,
"ImGUIDebugGUI");
1420 if ((driverConfig.
batch ==
false || getenv(
"DPL_DRIVER_REMOTE_GUI") !=
nullptr) && frameworkId.empty()) {
1421 debugGUI = initDebugGUI();
1423 if (driverConfig.
batch ==
false) {
1424 window = debugGUI->
initGUI(
"O2 Framework debug GUI", serviceRegistry);
1426 window = debugGUI->
initGUI(
nullptr, serviceRegistry);
1429 }
else if (getenv(
"DPL_DEVICE_REMOTE_GUI") && !frameworkId.empty()) {
1430 debugGUI = initDebugGUI();
1438 window = debugGUI->
initGUI(
nullptr, serviceRegistry);
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;
1448 bool guiQuitRequested =
false;
1449 bool hasError =
false;
1453 DriverState current;
1454 DriverState previous;
1460 if (!driverConfig.
batch) {
1462 uv_timer_init(loop, gui_timer);
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) {
1474 service.driverStartup(serviceRegistry,
DeviceConfig{varmap});
1478 ref.registerService(ServiceRegistryHelpers::handleForService<DevicesManager>(devicesManager));
1480 bool guiTimerExpired =
false;
1482 guiContext.
plugin = debugGUI;
1485 guiContext.
frameCost = &driverInfo.frameCost;
1494 .controls = &controls,
1496 .states = &allStates,
1497 .specs = &runningWorkflow.devices,
1498 .metrics = &metricsInfos,
1499 .metricProcessingCallbacks = &metricProcessingCallbacks,
1500 .summaryCallbacks = &summaryCallbacks,
1501 .driver = &driverInfo,
1503 .isDriver = frameworkId.empty(),
1506 serverContext.serverHandle.data = &serverContext;
1509 uv_timer_init(loop, &force_step_timer);
1511 uv_timer_init(loop, &force_exit_timer);
1513 bool guiDeployedOnce =
false;
1517 metricDumpTimer.data = &serverContext;
1518 bool allChildrenGone =
false;
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);
1535 driverInfo.states.push_back(transition);
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.";
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);
1558 driverInfo.sigchldRequested =
true;
1559 driverInfo.states.push_back(DriverState::HANDLE_CHILDREN);
1561 if (driverInfo.states.empty() ==
false) {
1563 current = driverInfo.states.back();
1565 current = DriverState::UNKNOWN;
1567 driverInfo.states.pop_back();
1569 case DriverState::BIND_GUI_PORT:
1570 bindGUIPort(driverInfo, serverContext, frameworkId);
1572 case DriverState::INIT:
1573 LOGP(info,
"Initialising O2 Data Processing Layer. Driver PID: {}.", getpid());
1574 LOGP(info,
"Driver listening on port: {}", driverInfo.port);
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) {
1587 if (driverInfo.noSHMCleanup) {
1588 LOGP(warning,
"Not cleaning up shared memory.");
1596 for (
auto& callback : driverInitCallbacks) {
1597 callback(serviceRegistry, {varmap});
1599 driverInfo.states.push_back(DriverState::RUNNING);
1601 LOG(info) <<
"O2 Data Processing Layer initialised. We brake for nobody.";
1603 LOGF(info,
"Optimised build. O2DEBUG / LOG(debug) / LOGF(debug) / assert statement will not be shown.");
1606 case DriverState::IMPORT_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()) {
1618 for (
auto channel : device.inputChannels) {
1621 for (
auto channel : device.outputChannels) {
1624 dataProcessorInfos.push_back(
1633 case DriverState::MATERIALISE_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.");
1643 auto altered_workflow = workflow;
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());
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};
1655 bool altered =
false;
1656 for (
auto& device : altered_workflow) {
1658 if (device.name.find(
"internal") != std::string::npos) {
1662 if (device.inputs.empty() ==
true) {
1666 auto hasMetadata = std::any_of(device.inputs.begin(), device.inputs.end(), [](
InputSpec const& spec) {
1667 return spec.metadata.empty() == false;
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;
1682 LOGP(
debug,
"Adjusting device {}", device.name.c_str());
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());
1699 LOGP(
debug,
"Original inputs: ");
1700 for (
auto& input : device.inputs) {
1701 LOGP(
debug,
"-> {}", input.binding);
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) {
1711 if (param.name.find(
"control:") != std::string::npos) {
1713 if (param.defaultValue.get<bool>() == true) {
1724 device.inputs.erase(new_end,
end);
1725 LOGP(
debug,
"Adjusted inputs: ");
1726 for (
auto& input : device.inputs) {
1727 LOGP(
debug,
"-> {}", input.binding);
1734 for (
auto& service : driverServices) {
1735 if (service.adjustTopology ==
nullptr) {
1738 service.adjustTopology(node, *driverInfo.configContext);
1746 driverInfo.channelPolicies,
1747 driverInfo.completionPolicies,
1748 driverInfo.dispatchPolicies,
1749 driverInfo.resourcePolicies,
1750 driverInfo.callbacksPolicies,
1751 driverInfo.sendingPolicies,
1752 driverInfo.forwardingPolicies,
1753 runningWorkflow.devices,
1755 driverInfo.uniqueWorkflowId,
1756 *driverInfo.configContext,
1757 !varmap[
"no-IPC"].as<
bool>(),
1758 driverInfo.resourcesMonitoringInterval,
1759 varmap[
"channel-prefix"].as<std::string>(),
1761 metricProcessingCallbacks.clear();
1762 std::vector<std::string> matchingServices;
1765 matchingServices.clear();
1766 for (
auto& device : runningWorkflow.devices) {
1767 for (
auto& service : device.services) {
1769 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1772 if (service.metricHandling) {
1773 metricProcessingCallbacks.push_back(service.metricHandling);
1774 matchingServices.push_back(service.name);
1780 matchingServices.clear();
1781 for (
auto& device : runningWorkflow.devices) {
1782 for (
auto& service : device.services) {
1784 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1787 if (service.summaryHandling) {
1788 summaryCallbacks.push_back(service.summaryHandling);
1789 matchingServices.push_back(service.name);
1794 preScheduleCallbacks.clear();
1795 matchingServices.clear();
1796 for (
auto& device : runningWorkflow.devices) {
1797 for (
auto& service : device.services) {
1799 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1802 if (service.preSchedule) {
1803 preScheduleCallbacks.push_back(service.preSchedule);
1807 postScheduleCallbacks.clear();
1808 matchingServices.clear();
1809 for (
auto& device : runningWorkflow.devices) {
1810 for (
auto& service : device.services) {
1812 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1815 if (service.postSchedule) {
1816 postScheduleCallbacks.push_back(service.postSchedule);
1820 driverInitCallbacks.clear();
1821 matchingServices.clear();
1822 for (
auto& device : runningWorkflow.devices) {
1823 for (
auto& service : device.services) {
1825 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1828 if (service.driverInit) {
1829 driverInitCallbacks.push_back(service.driverInit);
1837 for (
auto& device : runningWorkflow.devices) {
1839 if (device.name.find(
"internal") != std::string::npos) {
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);
1851 case VariantType::Int8:
1852 option.defaultValue = reg->get<int8_t>(
name);
1854 case VariantType::Int16:
1855 option.defaultValue = reg->get<int16_t>(
name);
1857 case VariantType::UInt8:
1858 option.defaultValue = reg->get<uint8_t>(
name);
1860 case VariantType::UInt16:
1861 option.defaultValue = reg->get<uint16_t>(
name);
1863 case VariantType::UInt32:
1864 option.defaultValue = reg->get<uint32_t>(
name);
1866 case VariantType::UInt64:
1867 option.defaultValue = reg->get<uint64_t>(
name);
1869 case VariantType::Int64:
1870 option.defaultValue = reg->get<int64_t>(
name);
1872 case VariantType::Float:
1873 option.defaultValue = reg->get<
float>(
name);
1875 case VariantType::Double:
1876 option.defaultValue = reg->get<
double>(
name);
1878 case VariantType::String:
1879 option.defaultValue = reg->get<std::string>(
name);
1881 case VariantType::Bool:
1882 option.defaultValue = reg->get<
bool>(
name);
1884 case VariantType::ArrayInt:
1885 option.defaultValue = reg->get<std::vector<int>>(
name);
1887 case VariantType::ArrayFloat:
1888 option.defaultValue = reg->get<std::vector<float>>(
name);
1890 case VariantType::ArrayDouble:
1891 option.defaultValue = reg->get<std::vector<double>>(
name);
1893 case VariantType::ArrayString:
1894 option.defaultValue = reg->get<std::vector<std::string>>(
name);
1896 case VariantType::Array2DInt:
1899 case VariantType::Array2DFloat:
1902 case VariantType::Array2DDouble:
1905 case VariantType::LabeledArrayInt:
1908 case VariantType::LabeledArrayFloat:
1911 case VariantType::LabeledArrayDouble:
1914 case VariantType::LabeledArrayString:
1923 }
catch (std::runtime_error& e) {
1924 LOGP(error,
"invalid workflow in {}: {}", driverInfo.argv[0], e.what());
1928#ifdef DPL_ENABLE_BACKTRACE
1931 LOGP(error,
"invalid workflow in {}: {}", driverInfo.argv[0], err.what);
1934 LOGP(error,
"invalid workflow in {}: Unknown error while materialising workflow", driverInfo.argv[0]);
1938 case DriverState::DO_CHILD:
1941 kill(getpid(), SIGSTOP);
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,
1948 runningWorkflow,
ref,
1950 driverInfo.processingPolicies,
1951 driverInfo.defaultDriverClient,
1956 std::ostringstream ss;
1957 for (
auto& processor : workflow) {
1958 ss <<
" - " << processor.name <<
"\n";
1960 for (
auto& spec : runningWorkflow.devices) {
1961 ss <<
" - " << spec.name <<
"(" << spec.id <<
")"
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);
1971 case DriverState::REDEPLOY_GUI:
1981 if (!driverConfig.
batch || getenv(
"DPL_DRIVER_REMOTE_GUI")) {
1983 uv_timer_stop(gui_timer);
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]() {
1989 for (
auto& service : driverServices) {
1990 if (service.postRenderGUI) {
1991 service.postRenderGUI(serviceRegistry);
1995 guiContext.
window = window;
1998 gui_timer->data = &guiContext;
2001 guiDeployedOnce =
true;
2004 case DriverState::MERGE_CONFIGS: {
2006 controls.resize(runningWorkflow.devices.size());
2009 if (varmap.count(
"dpl-tracing-flags")) {
2010 for (
auto& control : controls) {
2012 control.tracingFlags = tracingFlags;
2015 deviceExecutions.resize(runningWorkflow.devices.size());
2019 const auto uniformOptions = {
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",
2035 "--resources-monitoring",
2036 "--resources-monitoring-dump-interval",
2040 for (
auto& option : uniformOptions) {
2047 driverInfo.processingPolicies.termination == TerminationPolicy::WAIT,
2051 runningWorkflow.devices,
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";
2066 case DriverState::SCHEDULE: {
2071 LOG(info) <<
"Redeployment of configuration asked.";
2072 std::ostringstream forwardedStdin;
2074 infos.reserve(runningWorkflow.devices.size());
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__)
2088 for (
auto& callback : preScheduleCallbacks) {
2089 callback(serviceRegistry, {varmap});
2091 childFds.resize(runningWorkflow.devices.size());
2092 for (
int di = 0;
di < (
int)runningWorkflow.devices.size(); ++
di) {
2093 auto& context = childFds[
di];
2096 if (driverInfo.mode == DriverMode::EMBEDDED || runningWorkflow.devices[
di].resource.hostname != driverInfo.deployHostname) {
2098 runningWorkflow.devices[
di], controls[
di], deviceExecutions[
di], infos, allStates);
2103 runningWorkflow.devices, driverInfo,
2104 controls, deviceExecutions, infos,
2106 serviceRegistry, varmap,
2107 childFds, parentCPU, parentNode);
2112 for (
auto& callback : postScheduleCallbacks) {
2113 callback(serviceRegistry, {varmap});
2115 assert(infos.empty() ==
false);
2120 uv_timer_init(loop, &metricDumpTimer);
2122 driverInfo.resourcesMonitoringDumpInterval * 1000,
2123 driverInfo.resourcesMonitoringDumpInterval * 1000);
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) {
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;
2158 LOG(info) <<
"Redeployment of configuration done.";
2160 case DriverState::RUNNING:
2164 devicesManager->flush();
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");
2171 uv_run(loop, once ? UV_RUN_ONCE : UV_RUN_NOWAIT);
2175 if (guiQuitRequested ||
2176 (driverInfo.processingPolicies.termination == TerminationPolicy::QUIT && (
checkIfCanExit(infos) ==
true))) {
2181 LOG(info) <<
"Quitting";
2182 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
2183 }
else if (infos.size() != runningWorkflow.devices.size()) {
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.";
2195 driverInfo.states.push_back(DriverState::EXIT);
2196 }
else if (runningWorkflow.devices.empty() && driverConfig.
batch ==
false && !guiDeployedOnce) {
2198 driverInfo.states.push_back(DriverState::RUNNING);
2199 driverInfo.states.push_back(DriverState::REDEPLOY_GUI);
2201 driverInfo.states.push_back(DriverState::RUNNING);
2204 case DriverState::QUIT_REQUESTED:
2205 LOG(info) <<
"QUIT_REQUESTED";
2206 guiQuitRequested =
true;
2213 force_step_timer.data = &infos;
2215 driverInfo.states.push_back(DriverState::HANDLE_CHILDREN);
2217 case DriverState::HANDLE_CHILDREN: {
2221 uv_run(loop, once ? UV_RUN_ONCE : UV_RUN_NOWAIT);
2226 static bool forcefulExitMessage =
true;
2227 if (forcefulExitMessage) {
2228 LOG(info) <<
"Forceful exit requested.";
2229 forcefulExitMessage =
false;
2235 driverInfo.sigchldRequested =
false;
2240 bool supposedToQuit = (guiQuitRequested || canExit ||
graceful_exit);
2242 if (allChildrenGone && (supposedToQuit || driverInfo.processingPolicies.termination == TerminationPolicy::QUIT)) {
2244 driverInfo.states.resize(0);
2245 driverInfo.states.push_back(DriverState::EXIT);
2246 }
else if (hasError && driverInfo.processingPolicies.error == TerminationPolicy::QUIT && !supposedToQuit) {
2248 force_exit_timer.data = &infos;
2249 static bool forceful_timer_started =
false;
2250 if (forceful_timer_started ==
false) {
2251 forceful_timer_started =
true;
2254 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
2255 }
else if (allChildrenGone ==
false && supposedToQuit) {
2256 driverInfo.states.push_back(DriverState::HANDLE_CHILDREN);
2260 case DriverState::EXIT: {
2262 if (driverInfo.resourcesMonitoringDumpInterval) {
2263 uv_timer_stop(&metricDumpTimer);
2265 LOG(info) <<
"Dumping performance metrics to performanceMetrics.json file";
2268 dumpRunSummary(serverContext, driverInfo, infos, runningWorkflow.devices);
2275 if (infos.empty()) {
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);
2285 LOG(info) <<
"Dumping used configuration in dpl-config.json";
2287 std::ofstream outDPLConfigFile(
"dpl-config.json", std::ios::out);
2288 if (outDPLConfigFile.is_open()) {
2289 boost::property_tree::write_json(outDPLConfigFile, finalConfig);
2291 LOGP(warning,
"Could not write out final configuration file. Read only run folder?");
2293 if (driverInfo.noSHMCleanup) {
2294 LOGP(warning,
"Not cleaning up shared memory.");
2298 return calculateExitCode(driverInfo, runningWorkflow.devices, infos);
2300 case DriverState::PERFORM_CALLBACKS:
2301 for (
auto& callback : driverControl.
callbacks) {
2302 callback(workflow, runningWorkflow.devices, deviceExecutions, dataProcessorInfos, commandInfo);
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);
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,
2861 if (getenv(
"DPL_DRIVER_SIGNPOSTS")) {
2865 std::vector<std::string> currentArgs;
2866 std::vector<PluginInfo> plugins;
2869 for (
int ai = 1; ai < argc; ++ai) {
2870 currentArgs.emplace_back(argv[ai]);
2876 currentWorkflowOptions};
2880 bpo::options_description executorOptions(
"Executor options");
2881 const char* helpDescription =
"print help: short, full, executor, or processor name";
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");
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");
2923 bpo::options_description visibleOptions;
2924 visibleOptions.add(executorOptions);
2926 auto physicalWorkflow = workflow;
2927 std::map<std::string, size_t> rankIndex;
2934 size_t workflowHashA = 0;
2935 std::hash<std::string> hash_fn;
2937 for (
auto& dp : workflow) {
2938 workflowHashA += hash_fn(dp.name);
2941 for (
auto& dp : workflow) {
2942 rankIndex.insert(std::make_pair(dp.name, workflowHashA));
2945 std::vector<DataProcessorInfo> dataProcessorInfos;
2949 std::vector<DataProcessorSpec> importedWorkflow;
2951 if (previousWorked ==
false) {
2955 size_t workflowHashB = 0;
2956 for (
auto& dp : importedWorkflow) {
2957 workflowHashB += hash_fn(dp.name);
2964 for (
auto& dp : importedWorkflow) {
2965 auto found = std::find_if(physicalWorkflow.begin(), physicalWorkflow.end(),
2967 if (found == physicalWorkflow.end()) {
2968 physicalWorkflow.push_back(dp);
2969 rankIndex.insert(std::make_pair(dp.name, workflowHashB));
2978 for (
auto& dp : physicalWorkflow) {
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;
2994 auto reader = std::find_if(physicalWorkflow.begin(), physicalWorkflow.end(), [](
DataProcessorSpec& spec) { return spec.name ==
"internal-dpl-aod-reader"; });
2995 if (reader != physicalWorkflow.end()) {
2998 for (
auto& service : driverServices) {
2999 if (service.injectTopology ==
nullptr) {
3003 service.injectTopology(node, configContext);
3005 for (
auto& dp : physicalWorkflow) {
3006 if (dp.name.rfind(
"internal-", 0) == 0) {
3007 rankIndex.insert(std::make_pair(dp.name, hash_fn(
"internal")));
3014 return a.name < b.name;
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); });
3026 std::vector<std::pair<int, int>> edges;
3028 if (physicalWorkflow.size() > 1) {
3032 if (topoInfos.size() != physicalWorkflow.size()) {
3035 throw std::runtime_error(
"Unable to do topological sort of the resulting workflow. Do you have loops?\n" +
debugTopoInfo(physicalWorkflow, topoInfos, edges));
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;
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;
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;
3062 bpo::options_description od;
3063 od.add(visibleOptions);
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;
3073 bpo::command_line_parser(argc, argv)
3078 }
catch (std::exception
const& e) {
3079 LOGP(error,
"error parsing options of {}: {}", argv[0], e.what());
3095 if (varmap.count(
"help")) {
3096 printHelp(varmap, executorOptions, physicalWorkflow, currentWorkflowOptions);
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);
3123 LOGP(error,
"Invalid log level '{}'", logLevel);
3130 auto evaluateBatchOption = [&varmap]() ->
bool {
3131 if (varmap.count(
"no-batch") > 0) {
3134 if (varmap.count(
"batch") == 0) {
3136 return isatty(fileno(stdout)) == 0;
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>();
3163 auto batch = evaluateBatchOption();
3166 .driverHasGUI = (batch ==
false) || getenv(
"DPL_DRIVER_REMOTE_GUI") !=
nullptr,
3169 if (varmap[
"error-policy"].defaulted() && driverConfig.batch ==
false) {
3170 driverInfo.processingPolicies.error = TerminationPolicy::WAIT;
3172 driverInfo.processingPolicies.error = varmap[
"error-policy"].as<
TerminationPolicy>();
3175 driverInfo.startTime = uv_hrtime();
3176 driverInfo.startTimeMsFromEpoch = std::chrono::duration_cast<std::chrono::milliseconds>(
3177 std::chrono::system_clock::now().time_since_epoch())
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>();
3186 driverInfo.processorInfo = dataProcessorInfos;
3187 driverInfo.configContext = &configContext;
3194 std::string frameworkId;
3197 if (varmap.count(
"id")) {
3201 frameworkId = std::regex_replace(varmap[
"id"].as<std::string>(), std::regex{
"_dds.*"},
"");
3202 driverInfo.uniqueWorkflowId = fmt::format(
"{}", getppid());
3203 driverInfo.defaultDriverClient =
"stdout://";
3205 driverInfo.uniqueWorkflowId = fmt::format(
"{}", getpid());
3206 driverInfo.defaultDriverClient =
"ws://";