1356 DriverInfo& driverInfo,
1358 std::vector<DeviceMetricsInfo>& metricsInfos,
1359 std::vector<ConfigParamSpec>
const& detectedParams,
1360 boost::program_options::variables_map& varmap,
1361 std::vector<ServiceSpec>& driverServices,
1362 std::string frameworkId)
1366 .shmSegmentId = (int16_t)atoi(varmap[
"shm-segment-id"].as<std::string>().c_str())};
1370 auto* devicesManager =
new DevicesManager{.
controls = controls, .infos = infos, .specs = runningWorkflow.devices, .messages = {}};
1374 std::vector<uv_poll_t*> pollHandles;
1375 std::vector<DeviceStdioContext> childFds;
1377 std::vector<ComputingResource> resources;
1379 if (driverInfo.resources !=
"") {
1385 auto resourceManager = std::make_unique<SimpleResourceManager>(resources);
1388 void* window =
nullptr;
1389 decltype(debugGUI->
getGUIDebugger(infos, runningWorkflow.devices, allStates, dataProcessorInfos, metricsInfos, driverInfo, controls, driverControl)) debugGUICallback;
1392 auto initDebugGUI = []() ->
DebugGUI* {
1393 uv_lib_t supportLib;
1396 result = uv_dlopen(
"libO2FrameworkGUISupport.dylib", &supportLib);
1398 result = uv_dlopen(
"libO2FrameworkGUISupport.so", &supportLib);
1401 LOG(error) << uv_dlerror(&supportLib);
1406 result = uv_dlsym(&supportLib,
"dpl_plugin_callback", (
void**)&dpl_plugin_callback);
1408 LOG(error) << uv_dlerror(&supportLib);
1412 return PluginManager::getByName<DebugGUI>(pluginInstance,
"ImGUIDebugGUI");
1419 if ((driverConfig.
batch ==
false || getenv(
"DPL_DRIVER_REMOTE_GUI") !=
nullptr) && frameworkId.empty()) {
1420 debugGUI = initDebugGUI();
1422 if (driverConfig.
batch ==
false) {
1423 window = debugGUI->
initGUI(
"O2 Framework debug GUI", serviceRegistry);
1425 window = debugGUI->
initGUI(
nullptr, serviceRegistry);
1428 }
else if (getenv(
"DPL_DEVICE_REMOTE_GUI") && !frameworkId.empty()) {
1429 debugGUI = initDebugGUI();
1437 window = debugGUI->
initGUI(
nullptr, serviceRegistry);
1440 if (driverConfig.
batch ==
false && window ==
nullptr && frameworkId.empty()) {
1441 LOG(warn) <<
"Could not create GUI. Switching to batch mode. Do you have GLFW on your system?";
1442 driverConfig.
batch =
true;
1443 if (varmap[
"error-policy"].defaulted()) {
1444 driverInfo.processingPolicies.error = TerminationPolicy::QUIT;
1447 bool guiQuitRequested =
false;
1448 bool hasError =
false;
1452 DriverState current;
1453 DriverState previous;
1459 if (!driverConfig.
batch) {
1461 uv_timer_init(loop, gui_timer);
1464 std::vector<ServiceMetricHandling> metricProcessingCallbacks;
1465 std::vector<ServiceSummaryHandling> summaryCallbacks;
1466 std::vector<ServicePreSchedule> preScheduleCallbacks;
1467 std::vector<ServicePostSchedule> postScheduleCallbacks;
1468 std::vector<ServiceDriverInit> driverInitCallbacks;
1469 for (
auto& service : driverServices) {
1470 if (service.driverStartup ==
nullptr) {
1473 service.driverStartup(serviceRegistry,
DeviceConfig{varmap});
1477 ref.registerService(ServiceRegistryHelpers::handleForService<DevicesManager>(devicesManager));
1479 bool guiTimerExpired =
false;
1481 guiContext.
plugin = debugGUI;
1484 guiContext.
frameCost = &driverInfo.frameCost;
1493 .controls = &controls,
1495 .states = &allStates,
1496 .specs = &runningWorkflow.devices,
1497 .metrics = &metricsInfos,
1498 .metricProcessingCallbacks = &metricProcessingCallbacks,
1499 .summaryCallbacks = &summaryCallbacks,
1500 .driver = &driverInfo,
1502 .isDriver = frameworkId.empty(),
1505 serverContext.serverHandle.data = &serverContext;
1508 uv_timer_init(loop, &force_step_timer);
1510 uv_timer_init(loop, &force_exit_timer);
1512 bool guiDeployedOnce =
false;
1516 metricDumpTimer.data = &serverContext;
1517 bool allChildrenGone =
false;
1524 serverContext.asyncLogProcessing->data = &serverContext;
1525 uv_async_init(loop, serverContext.asyncLogProcessing, [](
uv_async_t* handle) {
1526 auto* context = (DriverServerContext*)handle->data;
1527 processChildrenOutput(context->loop, *context->driver, *context->infos, *context->specs, *context->controls);
1534 driverInfo.states.push_back(transition);
1541 auto currentTime = uv_hrtime();
1542 uint64_t diff = (currentTime - driverInfo.startTime) / 1000000000LL;
1543 if ((
graceful_exit ==
false) && (driverInfo.timeout > 0) && (diff > driverInfo.timeout)) {
1544 LOG(info) <<
"Timout ellapsed. Requesting to quit.";
1549 if (
graceful_exit ==
true && driverInfo.sigintRequested ==
false) {
1550 driverInfo.sigintRequested =
true;
1551 driverInfo.states.resize(0);
1552 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
1557 driverInfo.sigchldRequested =
true;
1558 driverInfo.states.push_back(DriverState::HANDLE_CHILDREN);
1560 if (driverInfo.states.empty() ==
false) {
1562 current = driverInfo.states.back();
1564 current = DriverState::UNKNOWN;
1566 driverInfo.states.pop_back();
1568 case DriverState::BIND_GUI_PORT:
1569 bindGUIPort(driverInfo, serverContext, frameworkId);
1571 case DriverState::INIT:
1572 LOGP(info,
"Initialising O2 Data Processing Layer. Driver PID: {}.", getpid());
1573 LOGP(info,
"Driver listening on port: {}", driverInfo.port);
1576 driverInfo.sa_handle_child.sa_handler = &handle_sigchld;
1577 sigemptyset(&driverInfo.sa_handle_child.sa_mask);
1578 driverInfo.sa_handle_child.sa_flags = SA_RESTART | SA_NOCLDSTOP;
1579 if (sigaction(SIGCHLD, &driverInfo.sa_handle_child,
nullptr) == -1) {
1586 if (driverInfo.noSHMCleanup) {
1587 LOGP(warning,
"Not cleaning up shared memory.");
1595 for (
auto& callback : driverInitCallbacks) {
1596 callback(serviceRegistry, {varmap});
1598 driverInfo.states.push_back(DriverState::RUNNING);
1600 LOG(info) <<
"O2 Data Processing Layer initialised. We brake for nobody.";
1602 LOGF(info,
"Optimised build. O2DEBUG / LOG(debug) / LOGF(debug) / assert statement will not be shown.");
1605 case DriverState::IMPORT_CURRENT_WORKFLOW:
1608 dataProcessorInfos = previousDataProcessorInfos;
1609 for (
auto const& device : runningWorkflow.devices) {
1610 auto exists = std::find_if(dataProcessorInfos.begin(),
1611 dataProcessorInfos.end(),
1612 [
id = device.id](
DataProcessorInfo const& info) ->
bool { return info.name == id; });
1613 if (exists != dataProcessorInfos.end()) {
1617 for (
auto channel : device.inputChannels) {
1620 for (
auto channel : device.outputChannels) {
1623 dataProcessorInfos.push_back(
1632 case DriverState::MATERIALISE_WORKFLOW:
1635 if (driverConfig.
batch ==
true && varmap[
"dds"].as<std::string>().empty() && !varmap[
"dump-workflow"].as<
bool>() && workflowState == WorkflowParsingState::Empty) {
1636 LOGP(error,
"Empty workflow provided while running in batch mode.");
1642 auto altered_workflow = workflow;
1644 auto confNameFromParam = [](std::string
const& paramName) {
1645 std::regex name_regex(R
"(^control:([\w-]+)\/(\w+))");
1646 auto match = std::sregex_token_iterator(paramName.begin(), paramName.end(), name_regex, 0);
1647 if (
match == std::sregex_token_iterator()) {
1648 throw runtime_error_f(
"Malformed process control spec: %s", paramName.c_str());
1650 std::string task = std::sregex_token_iterator(paramName.begin(), paramName.end(), name_regex, 1)->str();
1651 std::string conf = std::sregex_token_iterator(paramName.begin(), paramName.end(), name_regex, 2)->str();
1652 return std::pair{task, conf};
1654 bool altered =
false;
1655 for (
auto& device : altered_workflow) {
1657 if (device.name.find(
"internal") != std::string::npos) {
1661 if (device.inputs.empty() ==
true) {
1665 auto hasMetadata = std::any_of(device.inputs.begin(), device.inputs.end(), [](
InputSpec const& spec) {
1666 return spec.metadata.empty() == false;
1672 auto hasControls = std::any_of(device.inputs.begin(), device.inputs.end(), [](
InputSpec const& spec) {
1673 return std::any_of(spec.metadata.begin(), spec.metadata.end(), [](ConfigParamSpec const& param) {
1674 return param.type == VariantType::Bool && param.name.find(
"control:") != std::string::npos;
1681 LOGP(
debug,
"Adjusting device {}", device.name.c_str());
1684 if (configStore !=
nullptr) {
1685 auto reg = std::make_unique<ConfigParamRegistry>(std::move(configStore));
1686 for (
auto& input : device.inputs) {
1687 for (
auto&
param : input.metadata) {
1688 if (
param.type == VariantType::Bool &&
param.name.find(
"control:") != std::string::npos) {
1689 if (
param.name !=
"control:default" &&
param.name !=
"control:spawn" &&
param.name !=
"control:build" &&
param.name !=
"control:define") {
1690 auto confName = confNameFromParam(
param.name).second;
1691 param.defaultValue = reg->get<
bool>(confName.c_str());
1698 LOGP(
debug,
"Original inputs: ");
1699 for (
auto& input : device.inputs) {
1700 LOGP(
debug,
"-> {}", input.binding);
1702 auto end = device.inputs.end();
1703 auto new_end = std::remove_if(device.inputs.begin(), device.inputs.end(), [](
InputSpec& input) {
1704 auto requested = false;
1705 auto hasControls = false;
1706 for (auto& param : input.metadata) {
1707 if (param.type != VariantType::Bool) {
1710 if (param.name.find(
"control:") != std::string::npos) {
1712 if (param.defaultValue.get<bool>() == true) {
1723 device.inputs.erase(new_end,
end);
1724 LOGP(
debug,
"Adjusted inputs: ");
1725 for (
auto& input : device.inputs) {
1726 LOGP(
debug,
"-> {}", input.binding);
1733 for (
auto& service : driverServices) {
1734 if (service.adjustTopology ==
nullptr) {
1737 service.adjustTopology(node, *driverInfo.configContext);
1745 driverInfo.channelPolicies,
1746 driverInfo.completionPolicies,
1747 driverInfo.dispatchPolicies,
1748 driverInfo.resourcePolicies,
1749 driverInfo.callbacksPolicies,
1750 driverInfo.sendingPolicies,
1751 driverInfo.forwardingPolicies,
1752 runningWorkflow.devices,
1754 driverInfo.uniqueWorkflowId,
1755 *driverInfo.configContext,
1756 !varmap[
"no-IPC"].as<
bool>(),
1757 driverInfo.resourcesMonitoringInterval,
1758 varmap[
"channel-prefix"].as<std::string>(),
1760 metricProcessingCallbacks.clear();
1761 std::vector<std::string> matchingServices;
1764 matchingServices.clear();
1765 for (
auto& device : runningWorkflow.devices) {
1766 for (
auto& service : device.services) {
1768 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1771 if (service.metricHandling) {
1772 metricProcessingCallbacks.push_back(service.metricHandling);
1773 matchingServices.push_back(service.name);
1779 matchingServices.clear();
1780 for (
auto& device : runningWorkflow.devices) {
1781 for (
auto& service : device.services) {
1783 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1786 if (service.summaryHandling) {
1787 summaryCallbacks.push_back(service.summaryHandling);
1788 matchingServices.push_back(service.name);
1793 preScheduleCallbacks.clear();
1794 matchingServices.clear();
1795 for (
auto& device : runningWorkflow.devices) {
1796 for (
auto& service : device.services) {
1798 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1801 if (service.preSchedule) {
1802 preScheduleCallbacks.push_back(service.preSchedule);
1806 postScheduleCallbacks.clear();
1807 matchingServices.clear();
1808 for (
auto& device : runningWorkflow.devices) {
1809 for (
auto& service : device.services) {
1811 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1814 if (service.postSchedule) {
1815 postScheduleCallbacks.push_back(service.postSchedule);
1819 driverInitCallbacks.clear();
1820 matchingServices.clear();
1821 for (
auto& device : runningWorkflow.devices) {
1822 for (
auto& service : device.services) {
1824 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1827 if (service.driverInit) {
1828 driverInitCallbacks.push_back(service.driverInit);
1836 for (
auto& device : runningWorkflow.devices) {
1838 if (device.name.find(
"internal") != std::string::npos) {
1842 if (configStore !=
nullptr) {
1843 auto reg = std::make_unique<ConfigParamRegistry>(std::move(configStore));
1844 for (
auto& option : device.options) {
1845 const char*
name = option.name.c_str();
1846 switch (option.type) {
1847 case VariantType::Int:
1848 option.defaultValue = reg->get<int32_t>(
name);
1850 case VariantType::Int8:
1851 option.defaultValue = reg->get<int8_t>(
name);
1853 case VariantType::Int16:
1854 option.defaultValue = reg->get<int16_t>(
name);
1856 case VariantType::UInt8:
1857 option.defaultValue = reg->get<uint8_t>(
name);
1859 case VariantType::UInt16:
1860 option.defaultValue = reg->get<uint16_t>(
name);
1862 case VariantType::UInt32:
1863 option.defaultValue = reg->get<uint32_t>(
name);
1865 case VariantType::UInt64:
1866 option.defaultValue = reg->get<uint64_t>(
name);
1868 case VariantType::Int64:
1869 option.defaultValue = reg->get<int64_t>(
name);
1871 case VariantType::Float:
1872 option.defaultValue = reg->get<
float>(
name);
1874 case VariantType::Double:
1875 option.defaultValue = reg->get<
double>(
name);
1877 case VariantType::String:
1878 option.defaultValue = reg->get<std::string>(
name);
1880 case VariantType::Bool:
1881 option.defaultValue = reg->get<
bool>(
name);
1883 case VariantType::ArrayInt:
1884 option.defaultValue = reg->get<std::vector<int>>(
name);
1886 case VariantType::ArrayFloat:
1887 option.defaultValue = reg->get<std::vector<float>>(
name);
1889 case VariantType::ArrayDouble:
1890 option.defaultValue = reg->get<std::vector<double>>(
name);
1892 case VariantType::ArrayString:
1893 option.defaultValue = reg->get<std::vector<std::string>>(
name);
1895 case VariantType::Array2DInt:
1898 case VariantType::Array2DFloat:
1901 case VariantType::Array2DDouble:
1904 case VariantType::LabeledArrayInt:
1907 case VariantType::LabeledArrayFloat:
1910 case VariantType::LabeledArrayDouble:
1913 case VariantType::LabeledArrayString:
1922 }
catch (std::runtime_error& e) {
1923 LOGP(error,
"invalid workflow in {}: {}", driverInfo.argv[0], e.what());
1927#ifdef DPL_ENABLE_BACKTRACE
1930 LOGP(error,
"invalid workflow in {}: {}", driverInfo.argv[0], err.what);
1933 LOGP(error,
"invalid workflow in {}: Unknown error while materialising workflow", driverInfo.argv[0]);
1937 case DriverState::DO_CHILD:
1940 kill(getpid(), SIGSTOP);
1942 for (
size_t di = 0;
di < runningWorkflow.devices.size();
di++) {
1944 if (runningWorkflow.devices[
di].id == frameworkId) {
1945 return doChild(driverInfo.argc, driverInfo.argv,
1947 runningWorkflow,
ref,
1949 driverInfo.processingPolicies,
1950 driverInfo.defaultDriverClient,
1955 std::ostringstream ss;
1956 for (
auto& processor : workflow) {
1957 ss <<
" - " << processor.name <<
"\n";
1959 for (
auto& spec : runningWorkflow.devices) {
1960 ss <<
" - " << spec.name <<
"(" << spec.id <<
")"
1963 driverInfo.lastError = fmt::format(
1964 "Unable to find component with id {}."
1965 " Available options:\n{}",
1966 frameworkId, ss.str());
1967 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
1970 case DriverState::REDEPLOY_GUI:
1980 if (!driverConfig.
batch || getenv(
"DPL_DRIVER_REMOTE_GUI")) {
1982 uv_timer_stop(gui_timer);
1985 auto callback = debugGUI->
getGUIDebugger(infos, runningWorkflow.devices, allStates, dataProcessorInfos, metricsInfos, driverInfo, controls, driverControl);
1986 guiContext.
callback = [&serviceRegistry, &driverServices, &debugGUI, &infos, &runningWorkflow, &dataProcessorInfos, &metricsInfos, &driverInfo, &controls, &driverControl, callback]() {
1988 for (
auto& service : driverServices) {
1989 if (service.postRenderGUI) {
1990 service.postRenderGUI(serviceRegistry);
1994 guiContext.
window = window;
1997 gui_timer->data = &guiContext;
2000 guiDeployedOnce =
true;
2003 case DriverState::MERGE_CONFIGS: {
2005 controls.resize(runningWorkflow.devices.size());
2008 if (varmap.count(
"dpl-tracing-flags")) {
2009 for (
auto& control : controls) {
2011 control.tracingFlags = tracingFlags;
2014 deviceExecutions.resize(runningWorkflow.devices.size());
2018 const auto uniformOptions = {
2020 "--aod-memory-rate-limit",
2021 "--aod-writer-json",
2022 "--aod-writer-ntfmerge",
2023 "--aod-writer-resdir",
2024 "--aod-writer-resfile",
2025 "--aod-writer-resmode",
2026 "--aod-writer-maxfilesize",
2027 "--aod-writer-keep",
2028 "--aod-max-io-rate",
2029 "--aod-parent-access-level",
2030 "--aod-parent-base-path-replacement",
2031 "--driver-client-backend",
2032 "--fairmq-ipc-prefix",
2034 "--resources-monitoring",
2035 "--resources-monitoring-dump-interval",
2039 for (
auto& option : uniformOptions) {
2046 driverInfo.processingPolicies.termination == TerminationPolicy::WAIT,
2050 runningWorkflow.devices,
2054 driverInfo.uniqueWorkflowId);
2057 LOGP(error,
"unable to merge configurations in {}: {}", driverInfo.argv[0], err.what);
2058#ifdef DPL_ENABLE_BACKTRACE
2059 std::cerr <<
"\nStacktrace follows:\n\n";
2065 case DriverState::SCHEDULE: {
2070 LOG(info) <<
"Redeployment of configuration asked.";
2071 std::ostringstream forwardedStdin;
2073 infos.reserve(runningWorkflow.devices.size());
2076 unsigned parentCPU = -1;
2077 unsigned parentNode = -1;
2078#if defined(__linux__) && __has_include(<sched.h>)
2079 parentCPU = sched_getcpu();
2080#elif __has_include(<linux/getcpu.h>)
2081 getcpu(&parentCPU, &parentNode,
nullptr);
2082#elif __has_include(<cpuid.h>) && (__x86_64__ || __i386__)
2087 for (
auto& callback : preScheduleCallbacks) {
2088 callback(serviceRegistry, {varmap});
2090 childFds.resize(runningWorkflow.devices.size());
2091 for (
int di = 0;
di < (
int)runningWorkflow.devices.size(); ++
di) {
2092 auto& context = childFds[
di];
2095 if (driverInfo.mode == DriverMode::EMBEDDED || runningWorkflow.devices[
di].resource.hostname != driverInfo.deployHostname) {
2097 runningWorkflow.devices[
di], controls[
di], deviceExecutions[
di], infos, allStates);
2102 runningWorkflow.devices, driverInfo,
2103 controls, deviceExecutions, infos,
2105 serviceRegistry, varmap,
2106 childFds, parentCPU, parentNode);
2111 for (
auto& callback : postScheduleCallbacks) {
2112 callback(serviceRegistry, {varmap});
2114 assert(infos.empty() ==
false);
2119 uv_timer_init(loop, &metricDumpTimer);
2121 driverInfo.resourcesMonitoringDumpInterval * 1000,
2122 driverInfo.resourcesMonitoringDumpInterval * 1000);
2125 for (
const auto& processorInfo : dataProcessorInfos) {
2126 const auto& cmdLineArgs = processorInfo.cmdLineArgs;
2127 if (std::find(cmdLineArgs.begin(), cmdLineArgs.end(),
"--severity") != cmdLineArgs.end()) {
2128 for (
size_t counter = 0;
const auto& spec : runningWorkflow.devices) {
2129 if (spec.name.compare(processorInfo.name) == 0) {
2131 const auto logLevelIt = std::find(cmdLineArgs.begin(), cmdLineArgs.end(),
"--severity") + 1;
2132 if ((*logLevelIt).compare(
"debug") == 0) {
2133 info.logLevel = LogParsingHelpers::LogLevel::Debug;
2134 }
else if ((*logLevelIt).compare(
"detail") == 0) {
2135 info.logLevel = LogParsingHelpers::LogLevel::Debug;
2136 }
else if ((*logLevelIt).compare(
"info") == 0) {
2137 info.logLevel = LogParsingHelpers::LogLevel::Info;
2138 }
else if ((*logLevelIt).compare(
"warning") == 0) {
2139 info.logLevel = LogParsingHelpers::LogLevel::Warning;
2140 }
else if ((*logLevelIt).compare(
"error") == 0) {
2141 info.logLevel = LogParsingHelpers::LogLevel::Error;
2142 }
else if ((*logLevelIt).compare(
"important") == 0) {
2143 info.logLevel = LogParsingHelpers::LogLevel::Info;
2144 }
else if ((*logLevelIt).compare(
"alarm") == 0) {
2145 info.logLevel = LogParsingHelpers::LogLevel::Alarm;
2146 }
else if ((*logLevelIt).compare(
"critical") == 0) {
2147 info.logLevel = LogParsingHelpers::LogLevel::Critical;
2148 }
else if ((*logLevelIt).compare(
"fatal") == 0) {
2149 info.logLevel = LogParsingHelpers::LogLevel::Fatal;
2157 LOG(info) <<
"Redeployment of configuration done.";
2159 case DriverState::RUNNING:
2163 devicesManager->flush();
2167 if (guiTimerExpired ==
false) {
2168 O2_SIGNPOST_EVENT_EMIT(driver, sid,
"mainloop",
"Entering event loop with %{public}s", once ?
"UV_RUN_ONCE" :
"UV_RUN_NOWAIT");
2170 uv_run(loop, once ? UV_RUN_ONCE : UV_RUN_NOWAIT);
2174 if (guiQuitRequested ||
2175 (driverInfo.processingPolicies.termination == TerminationPolicy::QUIT && (
checkIfCanExit(infos) ==
true))) {
2180 LOG(info) <<
"Quitting";
2181 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
2182 }
else if (infos.size() != runningWorkflow.devices.size()) {
2187 driverInfo.states.push_back(DriverState::RUNNING);
2188 driverInfo.states.push_back(DriverState::REDEPLOY_GUI);
2189 driverInfo.states.push_back(DriverState::SCHEDULE);
2190 driverInfo.states.push_back(DriverState::MERGE_CONFIGS);
2191 }
else if (runningWorkflow.devices.empty() && driverConfig.
batch ==
true) {
2192 LOG(info) <<
"No device resulting from the workflow. Quitting.";
2194 driverInfo.states.push_back(DriverState::EXIT);
2195 }
else if (runningWorkflow.devices.empty() && driverConfig.
batch ==
false && !guiDeployedOnce) {
2197 driverInfo.states.push_back(DriverState::RUNNING);
2198 driverInfo.states.push_back(DriverState::REDEPLOY_GUI);
2200 driverInfo.states.push_back(DriverState::RUNNING);
2203 case DriverState::QUIT_REQUESTED:
2204 LOG(info) <<
"QUIT_REQUESTED";
2205 guiQuitRequested =
true;
2212 force_step_timer.data = &infos;
2214 driverInfo.states.push_back(DriverState::HANDLE_CHILDREN);
2216 case DriverState::HANDLE_CHILDREN: {
2220 uv_run(loop, once ? UV_RUN_ONCE : UV_RUN_NOWAIT);
2225 static bool forcefulExitMessage =
true;
2226 if (forcefulExitMessage) {
2227 LOG(info) <<
"Forceful exit requested.";
2228 forcefulExitMessage =
false;
2234 driverInfo.sigchldRequested =
false;
2239 bool supposedToQuit = (guiQuitRequested || canExit ||
graceful_exit);
2241 if (allChildrenGone && (supposedToQuit || driverInfo.processingPolicies.termination == TerminationPolicy::QUIT)) {
2243 driverInfo.states.resize(0);
2244 driverInfo.states.push_back(DriverState::EXIT);
2245 }
else if (hasError && driverInfo.processingPolicies.error == TerminationPolicy::QUIT && !supposedToQuit) {
2247 force_exit_timer.data = &infos;
2248 static bool forceful_timer_started =
false;
2249 if (forceful_timer_started ==
false) {
2250 forceful_timer_started =
true;
2253 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
2254 }
else if (allChildrenGone ==
false && supposedToQuit) {
2255 driverInfo.states.push_back(DriverState::HANDLE_CHILDREN);
2259 case DriverState::EXIT: {
2261 if (driverInfo.resourcesMonitoringDumpInterval) {
2262 uv_timer_stop(&metricDumpTimer);
2264 LOG(info) <<
"Dumping performance metrics to performanceMetrics.json file";
2267 dumpRunSummary(serverContext, driverInfo, infos, runningWorkflow.devices);
2274 if (infos.empty()) {
2277 boost::property_tree::ptree finalConfig;
2278 assert(infos.size() == runningWorkflow.devices.size());
2279 for (
size_t di = 0;
di < infos.size(); ++
di) {
2280 auto& info = infos[
di];
2281 auto& spec = runningWorkflow.devices[
di];
2282 finalConfig.put_child(spec.name, info.currentConfig);
2284 LOG(info) <<
"Dumping used configuration in dpl-config.json";
2286 std::ofstream outDPLConfigFile(
"dpl-config.json", std::ios::out);
2287 if (outDPLConfigFile.is_open()) {
2288 boost::property_tree::write_json(outDPLConfigFile, finalConfig);
2290 LOGP(warning,
"Could not write out final configuration file. Read only run folder?");
2292 if (driverInfo.noSHMCleanup) {
2293 LOGP(warning,
"Not cleaning up shared memory.");
2297 return calculateExitCode(driverInfo, runningWorkflow.devices, infos);
2299 case DriverState::PERFORM_CALLBACKS:
2300 for (
auto& callback : driverControl.
callbacks) {
2301 callback(workflow, runningWorkflow.devices, deviceExecutions, dataProcessorInfos, commandInfo);
2306 LOG(error) <<
"Driver transitioned in an unknown state("
2307 <<
"current: " << (
int)current
2308 <<
", previous: " << (
int)previous
2309 <<
"). Shutting down.";
2310 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
2848 std::vector<ChannelConfigurationPolicy>
const& channelPolicies,
2849 std::vector<CompletionPolicy>
const& completionPolicies,
2850 std::vector<DispatchPolicy>
const& dispatchPolicies,
2851 std::vector<ResourcePolicy>
const& resourcePolicies,
2852 std::vector<CallbacksPolicy>
const& callbacksPolicies,
2853 std::vector<SendingPolicy>
const& sendingPolicies,
2854 std::vector<ConfigParamSpec>
const& currentWorkflowOptions,
2855 std::vector<ConfigParamSpec>
const& detectedParams,
2860 if (getenv(
"DPL_DRIVER_SIGNPOSTS")) {
2864 std::vector<std::string> currentArgs;
2865 std::vector<PluginInfo> plugins;
2868 for (
int ai = 1; ai < argc; ++ai) {
2869 currentArgs.emplace_back(argv[ai]);
2875 currentWorkflowOptions};
2879 bpo::options_description executorOptions(
"Executor options");
2880 const char* helpDescription =
"print help: short, full, executor, or processor name";
2882 executorOptions.add_options()
2883 (
"help,h", bpo::value<std::string>()->implicit_value(
"short"), helpDescription)
2884 (
"quiet,q", bpo::value<bool>()->zero_tokens()->default_value(
false),
"quiet operation")
2885 (
"stop,s", bpo::value<bool>()->zero_tokens()->default_value(
false),
"stop before device start")
2886 (
"single-step", bpo::value<bool>()->zero_tokens()->default_value(
false),
"start in single step mode")
2887 (
"batch,b", bpo::value<std::vector<std::string>>()->zero_tokens()->composing(),
"batch processing mode")
2888 (
"no-batch", bpo::value<bool>()->zero_tokens(),
"force gui processing mode")
2889 (
"no-cleanup", bpo::value<bool>()->zero_tokens()->default_value(
false),
"do not cleanup the shm segment")
2890 (
"hostname", bpo::value<std::string>()->default_value(
"localhost"),
"hostname to deploy")
2891 (
"resources", bpo::value<std::string>()->default_value(
""),
"resources allocated for the workflow")
2892 (
"start-port,p", bpo::value<unsigned short>()->default_value(22000),
"start port to allocate")
2893 (
"port-range,pr", bpo::value<unsigned short>()->default_value(1000),
"ports in range")
2894 (
"completion-policy,c", bpo::value<TerminationPolicy>(&processingPolicies.
termination)->default_value(TerminationPolicy::QUIT),
2895 "what to do when processing is finished: quit, wait")
2896 (
"error-policy", bpo::value<TerminationPolicy>(&processingPolicies.
error)->default_value(TerminationPolicy::QUIT),
2897 "what to do when a device has an error: quit, wait")
2898 (
"min-failure-level", bpo::value<LogParsingHelpers::LogLevel>(&minFailureLevel)->default_value(LogParsingHelpers::LogLevel::Fatal),
2899 "minimum message level which will be considered as fatal and exit with 1")
2900 (
"graphviz,g", bpo::value<bool>()->zero_tokens()->default_value(
false),
"produce graphviz output")
2901 (
"mermaid", bpo::value<std::string>()->default_value(
""),
"produce graph output in mermaid format in file under specified name or on stdout if argument is \"-\"")
2902 (
"timeout,t", bpo::value<uint64_t>()->default_value(0),
"forced exit timeout (in seconds)")
2903 (
"dds,D", bpo::value<std::string>()->default_value(
""),
"create DDS configuration")
2904 (
"dds-workflow-suffix,D", bpo::value<std::string>()->default_value(
""),
"suffix for DDS names")
2905 (
"dump-workflow,dump", bpo::value<bool>()->zero_tokens()->default_value(
false),
"dump workflow as JSON")
2906 (
"dump-workflow-file", bpo::value<std::string>()->default_value(
"-"),
"file to which do the dump")
2907 (
"driver-mode", bpo::value<DriverMode>(&driverMode)->default_value(DriverMode::STANDALONE), R
"(how to run the driver. default: "standalone". Valid: "embedded")")
2908 (
"run", bpo::value<bool>()->zero_tokens()->default_value(
false),
"run workflow merged so far. It implies --batch. Use --no-batch to see the GUI")
2909 (
"no-IPC", bpo::value<bool>()->zero_tokens()->default_value(
false),
"disable IPC topology optimization")
2910 (
"o2-control,o2", bpo::value<std::string>()->default_value(
""),
"dump O2 Control workflow configuration under the specified name")
2911 (
"resources-monitoring", bpo::value<unsigned short>()->default_value(0),
"enable cpu/memory monitoring for provided interval in seconds")
2912 (
"resources-monitoring-dump-interval", bpo::value<unsigned short>()->default_value(0),
"dump monitoring information to disk every provided seconds");
2917 (
"id,i", bpo::value<std::string>(),
"device id for child spawning")
2918 (
"channel-config", bpo::value<std::vector<std::string>>(),
"channel configuration")
2919 (
"control",
"control plugin")
2920 (
"log-color",
"logging color scheme")(
"color",
"logging color scheme");
2922 bpo::options_description visibleOptions;
2923 visibleOptions.add(executorOptions);
2925 auto physicalWorkflow = workflow;
2926 std::map<std::string, size_t> rankIndex;
2933 size_t workflowHashA = 0;
2934 std::hash<std::string> hash_fn;
2936 for (
auto& dp : workflow) {
2937 workflowHashA += hash_fn(dp.name);
2940 for (
auto& dp : workflow) {
2941 rankIndex.insert(std::make_pair(dp.name, workflowHashA));
2944 std::vector<DataProcessorInfo> dataProcessorInfos;
2948 std::vector<DataProcessorSpec> importedWorkflow;
2950 if (previousWorked ==
false) {
2954 size_t workflowHashB = 0;
2955 for (
auto& dp : importedWorkflow) {
2956 workflowHashB += hash_fn(dp.name);
2963 for (
auto& dp : importedWorkflow) {
2964 auto found = std::find_if(physicalWorkflow.begin(), physicalWorkflow.end(),
2966 if (found == physicalWorkflow.end()) {
2967 physicalWorkflow.push_back(dp);
2968 rankIndex.insert(std::make_pair(dp.name, workflowHashB));
2977 for (
auto& dp : physicalWorkflow) {
2979 if (std::find_if(dp.labels.begin(), dp.labels.end(), isExpendable) != dp.labels.end()) {
2980 for (
auto&
output : dp.outputs) {
2981 if (
output.lifetime == Lifetime::Timeframe) {
2982 output.lifetime = Lifetime::Sporadic;
2993 auto reader = std::find_if(physicalWorkflow.begin(), physicalWorkflow.end(), [](
DataProcessorSpec& spec) { return spec.name ==
"internal-dpl-aod-reader"; });
2994 if (reader != physicalWorkflow.end()) {
2997 for (
auto& service : driverServices) {
2998 if (service.injectTopology ==
nullptr) {
3002 service.injectTopology(node, configContext);
3004 for (
auto& dp : physicalWorkflow) {
3005 if (dp.name.rfind(
"internal-", 0) == 0) {
3006 rankIndex.insert(std::make_pair(dp.name, hash_fn(
"internal")));
3013 return a.name < b.name;
3016 for (
auto& dp : physicalWorkflow) {
3017 std::stable_sort(dp.inputs.begin(), dp.inputs.end(),
3018 [](
InputSpec const&
a,
InputSpec const&
b) { return DataSpecUtils::describe(a) < DataSpecUtils::describe(b); });
3019 std::stable_sort(dp.outputs.begin(), dp.outputs.end(),
3020 [](
OutputSpec const&
a,
OutputSpec const&
b) { return DataSpecUtils::describe(a) < DataSpecUtils::describe(b); });
3024 std::vector<TopologyPolicy::DependencyChecker> dependencyCheckers;
3025 dependencyCheckers.reserve(physicalWorkflow.size());
3027 for (
auto& spec : physicalWorkflow) {
3028 for (
auto& policy : topologyPolicies) {
3029 if (policy.matcher(spec)) {
3030 dependencyCheckers.push_back(policy.checkDependency);
3035 assert(dependencyCheckers.size() == physicalWorkflow.size());
3037 auto checkDependencies = [&workflow = physicalWorkflow,
3038 &dependencyCheckers](
int i,
int j) {
3040 return checker(workflow[
i], workflow[
j]);
3045 std::vector<std::pair<int, int>> edges;
3047 if (physicalWorkflow.size() > 1) {
3048 for (
size_t i = 0;
i < physicalWorkflow.size() - 1; ++
i) {
3049 for (
size_t j =
i;
j < physicalWorkflow.size(); ++
j) {
3050 if (
i ==
j && checkDependencies(
i,
j)) {
3051 throw std::runtime_error(physicalWorkflow[
i].
name +
" depends on itself");
3054 if (checkDependencies(
i,
j)) {
3055 edges.emplace_back(
j,
i);
3058 if (checkDependencies(
j,
i)) {
3059 edges.emplace_back(
i,
j);
3061 std::ostringstream
str;
3062 for (
auto x : {
i,
j}) {
3063 str << physicalWorkflow[
x].name <<
":\n";
3065 for (
auto& input : physicalWorkflow[
x].inputs) {
3066 str <<
"- " << input <<
"\n";
3068 str <<
"outputs:\n";
3069 for (
auto&
output : physicalWorkflow[
x].outputs) {
3073 throw std::runtime_error(physicalWorkflow[
i].
name +
" has circular dependency with " + physicalWorkflow[
j].
name +
":\n" +
str.str());
3080 if (topoInfos.size() != physicalWorkflow.size()) {
3083 throw std::runtime_error(
"Unable to do topological sort of the resulting workflow. Do you have loops?\n" +
debugTopoInfo(physicalWorkflow, topoInfos, edges));
3087 auto aRank = std::make_tuple(a.layer, -workflow.at(a.index).outputs.size(), workflow.at(a.index).name);
3088 auto bRank = std::make_tuple(b.layer, -workflow.at(b.index).outputs.size(), workflow.at(b.index).name);
3089 return aRank < bRank;
3092 std::vector<int> dataProcessorOrder;
3093 dataProcessorOrder.resize(topoInfos.size());
3094 for (
size_t i = 0;
i < topoInfos.size(); ++
i) {
3095 dataProcessorOrder[topoInfos[
i].index] =
i;
3097 std::vector<int> newLocations;
3098 newLocations.resize(dataProcessorOrder.size());
3099 for (
size_t i = 0;
i < dataProcessorOrder.size(); ++
i) {
3100 newLocations[dataProcessorOrder[
i]] =
i;
3110 bpo::options_description od;
3111 od.add(visibleOptions);
3116 using namespace bpo::command_line_style;
3117 auto style = (allow_short | short_allow_adjacent | short_allow_next | allow_long | long_allow_adjacent | long_allow_next | allow_sticky | allow_dash_for_short);
3118 bpo::variables_map varmap;
3121 bpo::command_line_parser(argc, argv)
3126 }
catch (std::exception
const& e) {
3127 LOGP(error,
"error parsing options of {}: {}", argv[0], e.what());
3143 if (varmap.count(
"help")) {
3144 printHelp(varmap, executorOptions, physicalWorkflow, currentWorkflowOptions);
3150 if (varmap.count(
"severity")) {
3151 auto logLevel = varmap[
"severity"].as<std::string>();
3152 if (logLevel ==
"debug") {
3153 fair::Logger::SetConsoleSeverity(fair::Severity::debug);
3154 }
else if (logLevel ==
"detail") {
3155 fair::Logger::SetConsoleSeverity(fair::Severity::detail);
3156 }
else if (logLevel ==
"info") {
3157 fair::Logger::SetConsoleSeverity(fair::Severity::info);
3158 }
else if (logLevel ==
"warning") {
3159 fair::Logger::SetConsoleSeverity(fair::Severity::warning);
3160 }
else if (logLevel ==
"error") {
3161 fair::Logger::SetConsoleSeverity(fair::Severity::error);
3162 }
else if (logLevel ==
"important") {
3163 fair::Logger::SetConsoleSeverity(fair::Severity::important);
3164 }
else if (logLevel ==
"alarm") {
3165 fair::Logger::SetConsoleSeverity(fair::Severity::alarm);
3166 }
else if (logLevel ==
"critical") {
3167 fair::Logger::SetConsoleSeverity(fair::Severity::critical);
3168 }
else if (logLevel ==
"fatal") {
3169 fair::Logger::SetConsoleSeverity(fair::Severity::fatal);
3171 LOGP(error,
"Invalid log level '{}'", logLevel);
3178 auto evaluateBatchOption = [&varmap]() ->
bool {
3179 if (varmap.count(
"no-batch") > 0) {
3182 if (varmap.count(
"batch") == 0) {
3184 return isatty(fileno(stdout)) == 0;
3193 DriverInfo driverInfo{
3194 .sendingPolicies = sendingPolicies,
3195 .forwardingPolicies = forwardingPolicies,
3196 .callbacksPolicies = callbacksPolicies};
3197 driverInfo.states.reserve(10);
3198 driverInfo.sigintRequested =
false;
3199 driverInfo.sigchldRequested =
false;
3200 driverInfo.channelPolicies = channelPolicies;
3201 driverInfo.completionPolicies = completionPolicies;
3202 driverInfo.dispatchPolicies = dispatchPolicies;
3203 driverInfo.resourcePolicies = resourcePolicies;
3204 driverInfo.argc = argc;
3205 driverInfo.argv = argv;
3206 driverInfo.noSHMCleanup = varmap[
"no-cleanup"].as<
bool>();
3207 driverInfo.processingPolicies.termination = varmap[
"completion-policy"].as<
TerminationPolicy>();
3208 driverInfo.processingPolicies.earlyForward = varmap[
"early-forward-policy"].as<
EarlyForwardPolicy>();
3209 driverInfo.mode = varmap[
"driver-mode"].as<
DriverMode>();
3211 auto batch = evaluateBatchOption();
3214 .driverHasGUI = (batch ==
false) || getenv(
"DPL_DRIVER_REMOTE_GUI") !=
nullptr,
3217 if (varmap[
"error-policy"].defaulted() && driverConfig.batch ==
false) {
3218 driverInfo.processingPolicies.error = TerminationPolicy::WAIT;
3220 driverInfo.processingPolicies.error = varmap[
"error-policy"].as<
TerminationPolicy>();
3223 driverInfo.startTime = uv_hrtime();
3224 driverInfo.startTimeMsFromEpoch = std::chrono::duration_cast<std::chrono::milliseconds>(
3225 std::chrono::system_clock::now().time_since_epoch())
3227 driverInfo.timeout = varmap[
"timeout"].as<uint64_t>();
3228 driverInfo.deployHostname = varmap[
"hostname"].as<std::string>();
3229 driverInfo.resources = varmap[
"resources"].as<std::string>();
3230 driverInfo.resourcesMonitoringInterval = varmap[
"resources-monitoring"].as<
unsigned short>();
3231 driverInfo.resourcesMonitoringDumpInterval = varmap[
"resources-monitoring-dump-interval"].as<
unsigned short>();
3234 driverInfo.processorInfo = dataProcessorInfos;
3235 driverInfo.configContext = &configContext;
3242 std::string frameworkId;
3245 if (varmap.count(
"id")) {
3249 frameworkId = std::regex_replace(varmap[
"id"].as<std::string>(), std::regex{
"_dds.*"},
"");
3250 driverInfo.uniqueWorkflowId = fmt::format(
"{}", getppid());
3251 driverInfo.defaultDriverClient =
"stdout://";
3253 driverInfo.uniqueWorkflowId = fmt::format(
"{}", getpid());
3254 driverInfo.defaultDriverClient =
"ws://";