1358 DriverInfo& driverInfo,
1360 std::vector<DeviceMetricsInfo>& metricsInfos,
1361 std::vector<ConfigParamSpec>
const& detectedParams,
1362 boost::program_options::variables_map& varmap,
1363 std::vector<ServiceSpec>& driverServices,
1364 std::string frameworkId)
1368 .shmSegmentId = (int16_t)atoi(varmap[
"shm-segment-id"].as<std::string>().c_str())};
1372 auto* devicesManager =
new DevicesManager{.
controls = controls, .infos = infos, .specs = runningWorkflow.devices, .messages = {}};
1376 std::vector<uv_poll_t*> pollHandles;
1377 std::vector<DeviceStdioContext> childFds;
1379 std::vector<ComputingResource> resources;
1381 if (driverInfo.resources !=
"") {
1387 auto resourceManager = std::make_unique<SimpleResourceManager>(resources);
1390 void* window =
nullptr;
1391 decltype(debugGUI->
getGUIDebugger(infos, runningWorkflow.devices, allStates, dataProcessorInfos, metricsInfos, driverInfo, controls, driverControl)) debugGUICallback;
1394 auto initDebugGUI = []() ->
DebugGUI* {
1395 uv_lib_t supportLib;
1398 result = uv_dlopen(
"libO2FrameworkGUISupport.dylib", &supportLib);
1400 result = uv_dlopen(
"libO2FrameworkGUISupport.so", &supportLib);
1403 LOG(error) << uv_dlerror(&supportLib);
1408 result = uv_dlsym(&supportLib,
"dpl_plugin_callback", (
void**)&dpl_plugin_callback);
1410 LOG(error) << uv_dlerror(&supportLib);
1414 return PluginManager::getByName<DebugGUI>(pluginInstance,
"ImGUIDebugGUI");
1421 if ((driverConfig.
batch ==
false || getenv(
"DPL_DRIVER_REMOTE_GUI") !=
nullptr) && frameworkId.empty()) {
1422 debugGUI = initDebugGUI();
1424 if (driverConfig.
batch ==
false) {
1425 window = debugGUI->
initGUI(
"O2 Framework debug GUI", serviceRegistry);
1427 window = debugGUI->
initGUI(
nullptr, serviceRegistry);
1430 }
else if (getenv(
"DPL_DEVICE_REMOTE_GUI") && !frameworkId.empty()) {
1431 debugGUI = initDebugGUI();
1439 window = debugGUI->
initGUI(
nullptr, serviceRegistry);
1442 if (driverConfig.
batch ==
false && window ==
nullptr && frameworkId.empty()) {
1443 LOG(warn) <<
"Could not create GUI. Switching to batch mode. Do you have GLFW on your system?";
1444 driverConfig.
batch =
true;
1445 if (varmap[
"error-policy"].defaulted()) {
1446 driverInfo.processingPolicies.error = TerminationPolicy::QUIT;
1449 bool guiQuitRequested =
false;
1450 bool hasError =
false;
1454 DriverState current;
1455 DriverState previous;
1461 if (!driverConfig.
batch) {
1463 uv_timer_init(loop, gui_timer);
1466 std::vector<ServiceMetricHandling> metricProcessingCallbacks;
1467 std::vector<ServiceSummaryHandling> summaryCallbacks;
1468 std::vector<ServicePreSchedule> preScheduleCallbacks;
1469 std::vector<ServicePostSchedule> postScheduleCallbacks;
1470 std::vector<ServiceDriverInit> driverInitCallbacks;
1471 for (
auto& service : driverServices) {
1472 if (service.driverStartup ==
nullptr) {
1475 service.driverStartup(serviceRegistry,
DeviceConfig{varmap});
1479 ref.registerService(ServiceRegistryHelpers::handleForService<DevicesManager>(devicesManager));
1481 bool guiTimerExpired =
false;
1483 guiContext.
plugin = debugGUI;
1486 guiContext.
frameCost = &driverInfo.frameCost;
1495 .controls = &controls,
1497 .states = &allStates,
1498 .specs = &runningWorkflow.devices,
1499 .metrics = &metricsInfos,
1500 .metricProcessingCallbacks = &metricProcessingCallbacks,
1501 .summaryCallbacks = &summaryCallbacks,
1502 .driver = &driverInfo,
1504 .isDriver = frameworkId.empty(),
1507 serverContext.serverHandle.data = &serverContext;
1510 uv_timer_init(loop, &force_step_timer);
1512 uv_timer_init(loop, &force_exit_timer);
1514 bool guiDeployedOnce =
false;
1518 metricDumpTimer.data = &serverContext;
1519 bool allChildrenGone =
false;
1526 serverContext.asyncLogProcessing->data = &serverContext;
1527 uv_async_init(loop, serverContext.asyncLogProcessing, [](
uv_async_t* handle) {
1528 auto* context = (DriverServerContext*)handle->data;
1529 processChildrenOutput(context->loop, *context->driver, *context->infos, *context->specs, *context->controls);
1536 driverInfo.states.push_back(transition);
1543 auto currentTime = uv_hrtime();
1544 uint64_t diff = (currentTime - driverInfo.startTime) / 1000000000LL;
1545 if ((
graceful_exit ==
false) && (driverInfo.timeout > 0) && (diff > driverInfo.timeout)) {
1546 LOG(info) <<
"Timout ellapsed. Requesting to quit.";
1551 if (
graceful_exit ==
true && driverInfo.sigintRequested ==
false) {
1552 driverInfo.sigintRequested =
true;
1553 driverInfo.states.resize(0);
1554 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
1559 driverInfo.sigchldRequested =
true;
1560 driverInfo.states.push_back(DriverState::HANDLE_CHILDREN);
1562 if (driverInfo.states.empty() ==
false) {
1564 current = driverInfo.states.back();
1566 current = DriverState::UNKNOWN;
1568 driverInfo.states.pop_back();
1570 case DriverState::BIND_GUI_PORT:
1571 bindGUIPort(driverInfo, serverContext, frameworkId);
1573 case DriverState::INIT:
1574 LOGP(info,
"Initialising O2 Data Processing Layer. Driver PID: {}.", getpid());
1575 LOGP(info,
"Driver listening on port: {}", driverInfo.port);
1578 driverInfo.sa_handle_child.sa_handler = &handle_sigchld;
1579 sigemptyset(&driverInfo.sa_handle_child.sa_mask);
1580 driverInfo.sa_handle_child.sa_flags = SA_RESTART | SA_NOCLDSTOP;
1581 if (sigaction(SIGCHLD, &driverInfo.sa_handle_child,
nullptr) == -1) {
1588 if (driverInfo.noSHMCleanup) {
1589 LOGP(warning,
"Not cleaning up shared memory.");
1597 for (
auto& callback : driverInitCallbacks) {
1598 callback(serviceRegistry, {varmap});
1600 driverInfo.states.push_back(DriverState::RUNNING);
1602 LOG(info) <<
"O2 Data Processing Layer initialised. We brake for nobody.";
1604 LOGF(info,
"Optimised build. O2DEBUG / LOG(debug) / LOGF(debug) / assert statement will not be shown.");
1607 case DriverState::IMPORT_CURRENT_WORKFLOW:
1610 dataProcessorInfos = previousDataProcessorInfos;
1611 for (
auto const& device : runningWorkflow.devices) {
1612 auto exists = std::find_if(dataProcessorInfos.begin(),
1613 dataProcessorInfos.end(),
1614 [
id = device.id](
DataProcessorInfo const& info) ->
bool { return info.name == id; });
1615 if (exists != dataProcessorInfos.end()) {
1619 for (
auto channel : device.inputChannels) {
1622 for (
auto channel : device.outputChannels) {
1625 dataProcessorInfos.push_back(
1634 case DriverState::MATERIALISE_WORKFLOW:
1637 if (driverConfig.
batch ==
true && varmap[
"dds"].as<std::string>().empty() && !varmap[
"dump-workflow"].as<
bool>() && workflowState == WorkflowParsingState::Empty) {
1638 LOGP(error,
"Empty workflow provided while running in batch mode.");
1644 auto altered_workflow = workflow;
1646 auto confNameFromParam = [](std::string
const& paramName) {
1647 std::regex name_regex(R
"(^control:([\w-]+)\/(\w+))");
1648 auto match = std::sregex_token_iterator(paramName.begin(), paramName.end(), name_regex, 0);
1649 if (
match == std::sregex_token_iterator()) {
1650 throw runtime_error_f(
"Malformed process control spec: %s", paramName.c_str());
1652 std::string task = std::sregex_token_iterator(paramName.begin(), paramName.end(), name_regex, 1)->str();
1653 std::string conf = std::sregex_token_iterator(paramName.begin(), paramName.end(), name_regex, 2)->str();
1654 return std::pair{task, conf};
1656 bool altered =
false;
1657 for (
auto& device : altered_workflow) {
1659 if (device.name.find(
"internal") != std::string::npos) {
1663 if (device.inputs.empty() ==
true) {
1667 auto hasMetadata = std::any_of(device.inputs.begin(), device.inputs.end(), [](
InputSpec const& spec) {
1668 return spec.metadata.empty() == false;
1674 auto hasControls = std::any_of(device.inputs.begin(), device.inputs.end(), [](
InputSpec const& spec) {
1675 return std::any_of(spec.metadata.begin(), spec.metadata.end(), [](ConfigParamSpec const& param) {
1676 return param.type == VariantType::Bool && param.name.find(
"control:") != std::string::npos;
1683 LOGP(
debug,
"Adjusting device {}", device.name.c_str());
1686 if (configStore !=
nullptr) {
1687 auto reg = std::make_unique<ConfigParamRegistry>(std::move(configStore));
1688 for (
auto& input : device.inputs) {
1689 for (
auto&
param : input.metadata) {
1690 if (
param.type == VariantType::Bool &&
param.name.find(
"control:") != std::string::npos) {
1691 if (
param.name !=
"control:default" &&
param.name !=
"control:spawn" &&
param.name !=
"control:build" &&
param.name !=
"control:define") {
1692 auto confName = confNameFromParam(
param.name).second;
1693 param.defaultValue = reg->get<
bool>(confName.c_str());
1700 LOGP(
debug,
"Original inputs: ");
1701 for (
auto& input : device.inputs) {
1702 LOGP(
debug,
"-> {}", input.binding);
1704 auto end = device.inputs.end();
1705 auto new_end = std::remove_if(device.inputs.begin(), device.inputs.end(), [](
InputSpec& input) {
1706 auto requested = false;
1707 auto hasControls = false;
1708 for (auto& param : input.metadata) {
1709 if (param.type != VariantType::Bool) {
1712 if (param.name.find(
"control:") != std::string::npos) {
1714 if (param.defaultValue.get<bool>() == true) {
1725 device.inputs.erase(new_end,
end);
1726 LOGP(
debug,
"Adjusted inputs: ");
1727 for (
auto& input : device.inputs) {
1728 LOGP(
debug,
"-> {}", input.binding);
1735 for (
auto& service : driverServices) {
1736 if (service.adjustTopology ==
nullptr) {
1739 service.adjustTopology(node, *driverInfo.configContext);
1747 driverInfo.channelPolicies,
1748 driverInfo.completionPolicies,
1749 driverInfo.dispatchPolicies,
1750 driverInfo.resourcePolicies,
1751 driverInfo.callbacksPolicies,
1752 driverInfo.sendingPolicies,
1753 driverInfo.forwardingPolicies,
1754 runningWorkflow.devices,
1756 driverInfo.uniqueWorkflowId,
1757 *driverInfo.configContext,
1758 !varmap[
"no-IPC"].as<
bool>(),
1759 driverInfo.resourcesMonitoringInterval,
1760 varmap[
"channel-prefix"].as<std::string>(),
1762 metricProcessingCallbacks.clear();
1763 std::vector<std::string> matchingServices;
1766 matchingServices.clear();
1767 for (
auto& device : runningWorkflow.devices) {
1768 for (
auto& service : device.services) {
1770 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1773 if (service.metricHandling) {
1774 metricProcessingCallbacks.push_back(service.metricHandling);
1775 matchingServices.push_back(service.name);
1781 matchingServices.clear();
1782 for (
auto& device : runningWorkflow.devices) {
1783 for (
auto& service : device.services) {
1785 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1788 if (service.summaryHandling) {
1789 summaryCallbacks.push_back(service.summaryHandling);
1790 matchingServices.push_back(service.name);
1795 preScheduleCallbacks.clear();
1796 matchingServices.clear();
1797 for (
auto& device : runningWorkflow.devices) {
1798 for (
auto& service : device.services) {
1800 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1803 if (service.preSchedule) {
1804 preScheduleCallbacks.push_back(service.preSchedule);
1808 postScheduleCallbacks.clear();
1809 matchingServices.clear();
1810 for (
auto& device : runningWorkflow.devices) {
1811 for (
auto& service : device.services) {
1813 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1816 if (service.postSchedule) {
1817 postScheduleCallbacks.push_back(service.postSchedule);
1821 driverInitCallbacks.clear();
1822 matchingServices.clear();
1823 for (
auto& device : runningWorkflow.devices) {
1824 for (
auto& service : device.services) {
1826 if (std::find(matchingServices.begin(), matchingServices.end(), service.name) != matchingServices.end()) {
1829 if (service.driverInit) {
1830 driverInitCallbacks.push_back(service.driverInit);
1838 for (
auto& device : runningWorkflow.devices) {
1840 if (device.name.find(
"internal") != std::string::npos) {
1844 if (configStore !=
nullptr) {
1845 auto reg = std::make_unique<ConfigParamRegistry>(std::move(configStore));
1846 for (
auto& option : device.options) {
1847 const char*
name = option.name.c_str();
1848 switch (option.type) {
1849 case VariantType::Int:
1850 option.defaultValue = reg->get<int32_t>(
name);
1852 case VariantType::Int8:
1853 option.defaultValue = reg->get<int8_t>(
name);
1855 case VariantType::Int16:
1856 option.defaultValue = reg->get<int16_t>(
name);
1858 case VariantType::UInt8:
1859 option.defaultValue = reg->get<uint8_t>(
name);
1861 case VariantType::UInt16:
1862 option.defaultValue = reg->get<uint16_t>(
name);
1864 case VariantType::UInt32:
1865 option.defaultValue = reg->get<uint32_t>(
name);
1867 case VariantType::UInt64:
1868 option.defaultValue = reg->get<uint64_t>(
name);
1870 case VariantType::Int64:
1871 option.defaultValue = reg->get<int64_t>(
name);
1873 case VariantType::Float:
1874 option.defaultValue = reg->get<
float>(
name);
1876 case VariantType::Double:
1877 option.defaultValue = reg->get<
double>(
name);
1879 case VariantType::String:
1880 option.defaultValue = reg->get<std::string>(
name);
1882 case VariantType::Bool:
1883 option.defaultValue = reg->get<
bool>(
name);
1885 case VariantType::ArrayInt:
1886 option.defaultValue = reg->get<std::vector<int>>(
name);
1888 case VariantType::ArrayFloat:
1889 option.defaultValue = reg->get<std::vector<float>>(
name);
1891 case VariantType::ArrayDouble:
1892 option.defaultValue = reg->get<std::vector<double>>(
name);
1894 case VariantType::ArrayString:
1895 option.defaultValue = reg->get<std::vector<std::string>>(
name);
1897 case VariantType::Array2DInt:
1900 case VariantType::Array2DFloat:
1903 case VariantType::Array2DDouble:
1906 case VariantType::LabeledArrayInt:
1909 case VariantType::LabeledArrayFloat:
1912 case VariantType::LabeledArrayDouble:
1915 case VariantType::LabeledArrayString:
1924 }
catch (std::runtime_error& e) {
1925 LOGP(error,
"invalid workflow in {}: {}", driverInfo.argv[0], e.what());
1929#ifdef DPL_ENABLE_BACKTRACE
1932 LOGP(error,
"invalid workflow in {}: {}", driverInfo.argv[0], err.what);
1935 LOGP(error,
"invalid workflow in {}: Unknown error while materialising workflow", driverInfo.argv[0]);
1939 case DriverState::DO_CHILD:
1942 kill(getpid(), SIGSTOP);
1944 for (
size_t di = 0;
di < runningWorkflow.devices.size();
di++) {
1946 if (runningWorkflow.devices[
di].id == frameworkId) {
1947 return doChild(driverInfo.argc, driverInfo.argv,
1949 runningWorkflow,
ref,
1951 driverInfo.processingPolicies,
1952 driverInfo.defaultDriverClient,
1957 std::ostringstream ss;
1958 for (
auto& processor : workflow) {
1959 ss <<
" - " << processor.name <<
"\n";
1961 for (
auto& spec : runningWorkflow.devices) {
1962 ss <<
" - " << spec.name <<
"(" << spec.id <<
")"
1965 driverInfo.lastError = fmt::format(
1966 "Unable to find component with id {}."
1967 " Available options:\n{}",
1968 frameworkId, ss.str());
1969 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
1972 case DriverState::REDEPLOY_GUI:
1982 if (!driverConfig.
batch || getenv(
"DPL_DRIVER_REMOTE_GUI")) {
1984 uv_timer_stop(gui_timer);
1987 auto callback = debugGUI->
getGUIDebugger(infos, runningWorkflow.devices, allStates, dataProcessorInfos, metricsInfos, driverInfo, controls, driverControl);
1988 guiContext.
callback = [&serviceRegistry, &driverServices, &debugGUI, &infos, &runningWorkflow, &dataProcessorInfos, &metricsInfos, &driverInfo, &controls, &driverControl, callback]() {
1990 for (
auto& service : driverServices) {
1991 if (service.postRenderGUI) {
1992 service.postRenderGUI(serviceRegistry);
1996 guiContext.
window = window;
1999 gui_timer->data = &guiContext;
2002 guiDeployedOnce =
true;
2005 case DriverState::MERGE_CONFIGS: {
2007 controls.resize(runningWorkflow.devices.size());
2010 if (varmap.count(
"dpl-tracing-flags")) {
2011 for (
auto& control : controls) {
2013 control.tracingFlags = tracingFlags;
2016 deviceExecutions.resize(runningWorkflow.devices.size());
2020 const auto uniformOptions = {
2022 "--aod-memory-rate-limit",
2023 "--aod-writer-json",
2024 "--aod-writer-ntfmerge",
2025 "--aod-writer-resdir",
2026 "--aod-writer-resfile",
2027 "--aod-writer-resmode",
2028 "--aod-writer-maxfilesize",
2029 "--aod-writer-keep",
2030 "--aod-max-io-rate",
2031 "--aod-parent-access-level",
2032 "--aod-parent-base-path-replacement",
2033 "--driver-client-backend",
2034 "--fairmq-ipc-prefix",
2036 "--resources-monitoring",
2037 "--resources-monitoring-dump-interval",
2041 for (
auto& option : uniformOptions) {
2048 driverInfo.processingPolicies.termination == TerminationPolicy::WAIT,
2052 runningWorkflow.devices,
2056 driverInfo.uniqueWorkflowId);
2059 LOGP(error,
"unable to merge configurations in {}: {}", driverInfo.argv[0], err.what);
2060#ifdef DPL_ENABLE_BACKTRACE
2061 std::cerr <<
"\nStacktrace follows:\n\n";
2067 case DriverState::SCHEDULE: {
2072 LOG(info) <<
"Redeployment of configuration asked.";
2073 std::ostringstream forwardedStdin;
2075 infos.reserve(runningWorkflow.devices.size());
2078 unsigned parentCPU = -1;
2079 unsigned parentNode = -1;
2080#if defined(__linux__) && __has_include(<sched.h>)
2081 parentCPU = sched_getcpu();
2082#elif __has_include(<linux/getcpu.h>)
2083 getcpu(&parentCPU, &parentNode,
nullptr);
2084#elif __has_include(<cpuid.h>) && (__x86_64__ || __i386__)
2089 for (
auto& callback : preScheduleCallbacks) {
2090 callback(serviceRegistry, {varmap});
2092 childFds.resize(runningWorkflow.devices.size());
2093 for (
int di = 0;
di < (
int)runningWorkflow.devices.size(); ++
di) {
2094 auto& context = childFds[
di];
2097 if (driverInfo.mode == DriverMode::EMBEDDED || runningWorkflow.devices[
di].resource.hostname != driverInfo.deployHostname) {
2099 runningWorkflow.devices[
di], controls[
di], deviceExecutions[
di], infos, allStates);
2104 runningWorkflow.devices, driverInfo,
2105 controls, deviceExecutions, infos,
2107 serviceRegistry, varmap,
2108 childFds, parentCPU, parentNode);
2113 for (
auto& callback : postScheduleCallbacks) {
2114 callback(serviceRegistry, {varmap});
2116 assert(infos.empty() ==
false);
2121 uv_timer_init(loop, &metricDumpTimer);
2123 driverInfo.resourcesMonitoringDumpInterval * 1000,
2124 driverInfo.resourcesMonitoringDumpInterval * 1000);
2127 for (
const auto& processorInfo : dataProcessorInfos) {
2128 const auto& cmdLineArgs = processorInfo.cmdLineArgs;
2129 if (std::find(cmdLineArgs.begin(), cmdLineArgs.end(),
"--severity") != cmdLineArgs.end()) {
2130 for (
size_t counter = 0;
const auto& spec : runningWorkflow.devices) {
2131 if (spec.name.compare(processorInfo.name) == 0) {
2133 const auto logLevelIt = std::find(cmdLineArgs.begin(), cmdLineArgs.end(),
"--severity") + 1;
2134 if ((*logLevelIt).compare(
"debug") == 0) {
2135 info.logLevel = LogParsingHelpers::LogLevel::Debug;
2136 }
else if ((*logLevelIt).compare(
"detail") == 0) {
2137 info.logLevel = LogParsingHelpers::LogLevel::Debug;
2138 }
else if ((*logLevelIt).compare(
"info") == 0) {
2139 info.logLevel = LogParsingHelpers::LogLevel::Info;
2140 }
else if ((*logLevelIt).compare(
"warning") == 0) {
2141 info.logLevel = LogParsingHelpers::LogLevel::Warning;
2142 }
else if ((*logLevelIt).compare(
"error") == 0) {
2143 info.logLevel = LogParsingHelpers::LogLevel::Error;
2144 }
else if ((*logLevelIt).compare(
"important") == 0) {
2145 info.logLevel = LogParsingHelpers::LogLevel::Info;
2146 }
else if ((*logLevelIt).compare(
"alarm") == 0) {
2147 info.logLevel = LogParsingHelpers::LogLevel::Alarm;
2148 }
else if ((*logLevelIt).compare(
"critical") == 0) {
2149 info.logLevel = LogParsingHelpers::LogLevel::Critical;
2150 }
else if ((*logLevelIt).compare(
"fatal") == 0) {
2151 info.logLevel = LogParsingHelpers::LogLevel::Fatal;
2159 LOG(info) <<
"Redeployment of configuration done.";
2161 case DriverState::RUNNING:
2165 devicesManager->flush();
2169 if (guiTimerExpired ==
false) {
2170 O2_SIGNPOST_EVENT_EMIT(driver, sid,
"mainloop",
"Entering event loop with %{public}s", once ?
"UV_RUN_ONCE" :
"UV_RUN_NOWAIT");
2172 uv_run(loop, once ? UV_RUN_ONCE : UV_RUN_NOWAIT);
2176 if (guiQuitRequested ||
2177 (driverInfo.processingPolicies.termination == TerminationPolicy::QUIT && (
checkIfCanExit(infos) ==
true))) {
2182 LOG(info) <<
"Quitting";
2183 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
2184 }
else if (infos.size() != runningWorkflow.devices.size()) {
2189 driverInfo.states.push_back(DriverState::RUNNING);
2190 driverInfo.states.push_back(DriverState::REDEPLOY_GUI);
2191 driverInfo.states.push_back(DriverState::SCHEDULE);
2192 driverInfo.states.push_back(DriverState::MERGE_CONFIGS);
2193 }
else if (runningWorkflow.devices.empty() && driverConfig.
batch ==
true) {
2194 LOG(info) <<
"No device resulting from the workflow. Quitting.";
2196 driverInfo.states.push_back(DriverState::EXIT);
2197 }
else if (runningWorkflow.devices.empty() && driverConfig.
batch ==
false && !guiDeployedOnce) {
2199 driverInfo.states.push_back(DriverState::RUNNING);
2200 driverInfo.states.push_back(DriverState::REDEPLOY_GUI);
2202 driverInfo.states.push_back(DriverState::RUNNING);
2205 case DriverState::QUIT_REQUESTED: {
2206 std::time_t
result = std::time(
nullptr);
2210 guiQuitRequested =
true;
2217 force_step_timer.data = &infos;
2219 driverInfo.states.push_back(DriverState::HANDLE_CHILDREN);
2222 case DriverState::HANDLE_CHILDREN: {
2226 uv_run(loop, once ? UV_RUN_ONCE : UV_RUN_NOWAIT);
2231 static bool forcefulExitMessage =
true;
2232 if (forcefulExitMessage) {
2233 LOG(info) <<
"Forceful exit requested.";
2234 forcefulExitMessage =
false;
2240 driverInfo.sigchldRequested =
false;
2245 bool supposedToQuit = (guiQuitRequested || canExit ||
graceful_exit);
2247 if (allChildrenGone && (supposedToQuit || driverInfo.processingPolicies.termination == TerminationPolicy::QUIT)) {
2249 driverInfo.states.resize(0);
2250 driverInfo.states.push_back(DriverState::EXIT);
2251 }
else if (hasError && driverInfo.processingPolicies.error == TerminationPolicy::QUIT && !supposedToQuit) {
2253 force_exit_timer.data = &infos;
2254 static bool forceful_timer_started =
false;
2255 if (forceful_timer_started ==
false) {
2256 forceful_timer_started =
true;
2259 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
2260 }
else if (allChildrenGone ==
false && supposedToQuit) {
2261 driverInfo.states.push_back(DriverState::HANDLE_CHILDREN);
2265 case DriverState::EXIT: {
2267 if (driverInfo.resourcesMonitoringDumpInterval) {
2268 uv_timer_stop(&metricDumpTimer);
2270 LOG(info) <<
"Dumping performance metrics to performanceMetrics.json file";
2273 dumpRunSummary(serverContext, driverInfo, infos, runningWorkflow.devices);
2280 if (infos.empty()) {
2283 boost::property_tree::ptree finalConfig;
2284 assert(infos.size() == runningWorkflow.devices.size());
2285 for (
size_t di = 0;
di < infos.size(); ++
di) {
2286 auto& info = infos[
di];
2287 auto& spec = runningWorkflow.devices[
di];
2288 finalConfig.put_child(spec.name, info.currentConfig);
2290 LOG(info) <<
"Dumping used configuration in dpl-config.json";
2292 std::ofstream outDPLConfigFile(
"dpl-config.json", std::ios::out);
2293 if (outDPLConfigFile.is_open()) {
2294 boost::property_tree::write_json(outDPLConfigFile, finalConfig);
2296 LOGP(warning,
"Could not write out final configuration file. Read only run folder?");
2298 if (driverInfo.noSHMCleanup) {
2299 LOGP(warning,
"Not cleaning up shared memory.");
2303 return calculateExitCode(driverInfo, runningWorkflow.devices, infos);
2305 case DriverState::PERFORM_CALLBACKS:
2306 for (
auto& callback : driverControl.
callbacks) {
2307 callback(workflow, runningWorkflow.devices, deviceExecutions, dataProcessorInfos, commandInfo);
2312 LOG(error) <<
"Driver transitioned in an unknown state("
2313 <<
"current: " << (
int)current
2314 <<
", previous: " << (
int)previous
2315 <<
"). Shutting down.";
2316 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
2854 std::vector<ChannelConfigurationPolicy>
const& channelPolicies,
2855 std::vector<CompletionPolicy>
const& completionPolicies,
2856 std::vector<DispatchPolicy>
const& dispatchPolicies,
2857 std::vector<ResourcePolicy>
const& resourcePolicies,
2858 std::vector<CallbacksPolicy>
const& callbacksPolicies,
2859 std::vector<SendingPolicy>
const& sendingPolicies,
2860 std::vector<ConfigParamSpec>
const& currentWorkflowOptions,
2861 std::vector<ConfigParamSpec>
const& detectedParams,
2866 if (getenv(
"DPL_DRIVER_SIGNPOSTS")) {
2870 std::vector<std::string> currentArgs;
2871 std::vector<PluginInfo> plugins;
2874 for (
int ai = 1; ai < argc; ++ai) {
2875 currentArgs.emplace_back(argv[ai]);
2881 currentWorkflowOptions};
2885 bpo::options_description executorOptions(
"Executor options");
2886 const char* helpDescription =
"print help: short, full, executor, or processor name";
2888 executorOptions.add_options()
2889 (
"help,h", bpo::value<std::string>()->implicit_value(
"short"), helpDescription)
2890 (
"quiet,q", bpo::value<bool>()->zero_tokens()->default_value(
false),
"quiet operation")
2891 (
"stop,s", bpo::value<bool>()->zero_tokens()->default_value(
false),
"stop before device start")
2892 (
"single-step", bpo::value<bool>()->zero_tokens()->default_value(
false),
"start in single step mode")
2893 (
"batch,b", bpo::value<std::vector<std::string>>()->zero_tokens()->composing(),
"batch processing mode")
2894 (
"no-batch", bpo::value<bool>()->zero_tokens(),
"force gui processing mode")
2895 (
"no-cleanup", bpo::value<bool>()->zero_tokens()->default_value(
false),
"do not cleanup the shm segment")
2896 (
"hostname", bpo::value<std::string>()->default_value(
"localhost"),
"hostname to deploy")
2897 (
"resources", bpo::value<std::string>()->default_value(
""),
"resources allocated for the workflow")
2898 (
"start-port,p", bpo::value<unsigned short>()->default_value(22000),
"start port to allocate")
2899 (
"port-range,pr", bpo::value<unsigned short>()->default_value(1000),
"ports in range")
2900 (
"completion-policy,c", bpo::value<TerminationPolicy>(&processingPolicies.
termination)->default_value(TerminationPolicy::QUIT),
2901 "what to do when processing is finished: quit, wait")
2902 (
"error-policy", bpo::value<TerminationPolicy>(&processingPolicies.
error)->default_value(TerminationPolicy::QUIT),
2903 "what to do when a device has an error: quit, wait")
2904 (
"min-failure-level", bpo::value<LogParsingHelpers::LogLevel>(&minFailureLevel)->default_value(LogParsingHelpers::LogLevel::Fatal),
2905 "minimum message level which will be considered as fatal and exit with 1")
2906 (
"graphviz,g", bpo::value<bool>()->zero_tokens()->default_value(
false),
"produce graphviz output")
2907 (
"mermaid", bpo::value<std::string>()->default_value(
""),
"produce graph output in mermaid format in file under specified name or on stdout if argument is \"-\"")
2908 (
"timeout,t", bpo::value<uint64_t>()->default_value(0),
"forced exit timeout (in seconds)")
2909 (
"dds,D", bpo::value<std::string>()->default_value(
""),
"create DDS configuration")
2910 (
"dds-workflow-suffix,D", bpo::value<std::string>()->default_value(
""),
"suffix for DDS names")
2911 (
"dump-workflow,dump", bpo::value<bool>()->zero_tokens()->default_value(
false),
"dump workflow as JSON")
2912 (
"dump-workflow-file", bpo::value<std::string>()->default_value(
"-"),
"file to which do the dump")
2913 (
"driver-mode", bpo::value<DriverMode>(&driverMode)->default_value(DriverMode::STANDALONE), R
"(how to run the driver. default: "standalone". Valid: "embedded")")
2914 (
"run", bpo::value<bool>()->zero_tokens()->default_value(
false),
"run workflow merged so far. It implies --batch. Use --no-batch to see the GUI")
2915 (
"no-IPC", bpo::value<bool>()->zero_tokens()->default_value(
false),
"disable IPC topology optimization")
2916 (
"o2-control,o2", bpo::value<std::string>()->default_value(
""),
"dump O2 Control workflow configuration under the specified name")
2917 (
"resources-monitoring", bpo::value<unsigned short>()->default_value(0),
"enable cpu/memory monitoring for provided interval in seconds")
2918 (
"resources-monitoring-dump-interval", bpo::value<unsigned short>()->default_value(0),
"dump monitoring information to disk every provided seconds");
2923 (
"id,i", bpo::value<std::string>(),
"device id for child spawning")
2924 (
"channel-config", bpo::value<std::vector<std::string>>(),
"channel configuration")
2925 (
"control",
"control plugin")
2926 (
"log-color",
"logging color scheme")(
"color",
"logging color scheme");
2928 bpo::options_description visibleOptions;
2929 visibleOptions.add(executorOptions);
2931 auto physicalWorkflow = workflow;
2932 std::map<std::string, size_t> rankIndex;
2939 size_t workflowHashA = 0;
2940 std::hash<std::string> hash_fn;
2942 for (
auto& dp : workflow) {
2943 workflowHashA += hash_fn(dp.name);
2946 for (
auto& dp : workflow) {
2947 rankIndex.insert(std::make_pair(dp.name, workflowHashA));
2950 std::vector<DataProcessorInfo> dataProcessorInfos;
2954 std::vector<DataProcessorSpec> importedWorkflow;
2956 if (previousWorked ==
false) {
2960 size_t workflowHashB = 0;
2961 for (
auto& dp : importedWorkflow) {
2962 workflowHashB += hash_fn(dp.name);
2969 for (
auto& dp : importedWorkflow) {
2970 auto found = std::find_if(physicalWorkflow.begin(), physicalWorkflow.end(),
2972 if (found == physicalWorkflow.end()) {
2973 physicalWorkflow.push_back(dp);
2974 rankIndex.insert(std::make_pair(dp.name, workflowHashB));
2983 for (
auto& dp : physicalWorkflow) {
2985 if (std::find_if(dp.labels.begin(), dp.labels.end(), isExpendable) != dp.labels.end()) {
2986 for (
auto&
output : dp.outputs) {
2987 if (
output.lifetime == Lifetime::Timeframe) {
2988 output.lifetime = Lifetime::Sporadic;
2999 auto reader = std::find_if(physicalWorkflow.begin(), physicalWorkflow.end(), [](
DataProcessorSpec& spec) { return spec.name ==
"internal-dpl-aod-reader"; });
3000 if (reader != physicalWorkflow.end()) {
3003 for (
auto& service : driverServices) {
3004 if (service.injectTopology ==
nullptr) {
3008 service.injectTopology(node, configContext);
3010 for (
auto& dp : physicalWorkflow) {
3011 if (dp.name.rfind(
"internal-", 0) == 0) {
3012 rankIndex.insert(std::make_pair(dp.name, hash_fn(
"internal")));
3019 return a.name < b.name;
3022 for (
auto& dp : physicalWorkflow) {
3023 std::stable_sort(dp.inputs.begin(), dp.inputs.end(),
3024 [](
InputSpec const&
a,
InputSpec const&
b) { return DataSpecUtils::describe(a) < DataSpecUtils::describe(b); });
3025 std::stable_sort(dp.outputs.begin(), dp.outputs.end(),
3026 [](
OutputSpec const&
a,
OutputSpec const&
b) { return DataSpecUtils::describe(a) < DataSpecUtils::describe(b); });
3031 std::vector<std::pair<int, int>> edges;
3033 if (physicalWorkflow.size() > 1) {
3037 if (topoInfos.size() != physicalWorkflow.size()) {
3040 throw std::runtime_error(
"Unable to do topological sort of the resulting workflow. Do you have loops?\n" +
debugTopoInfo(physicalWorkflow, topoInfos, edges));
3044 auto aRank = std::make_tuple(a.layer, -workflow.at(a.index).outputs.size(), workflow.at(a.index).name);
3045 auto bRank = std::make_tuple(b.layer, -workflow.at(b.index).outputs.size(), workflow.at(b.index).name);
3046 return aRank < bRank;
3049 std::vector<int> dataProcessorOrder;
3050 dataProcessorOrder.resize(topoInfos.size());
3051 for (
size_t i = 0;
i < topoInfos.size(); ++
i) {
3052 dataProcessorOrder[topoInfos[
i].index] =
i;
3054 std::vector<int> newLocations;
3055 newLocations.resize(dataProcessorOrder.size());
3056 for (
size_t i = 0;
i < dataProcessorOrder.size(); ++
i) {
3057 newLocations[dataProcessorOrder[
i]] =
i;
3067 bpo::options_description od;
3068 od.add(visibleOptions);
3073 using namespace bpo::command_line_style;
3074 auto style = (allow_short | short_allow_adjacent | short_allow_next | allow_long | long_allow_adjacent | long_allow_next | allow_sticky | allow_dash_for_short);
3075 bpo::variables_map varmap;
3078 bpo::command_line_parser(argc, argv)
3083 }
catch (std::exception
const& e) {
3084 LOGP(error,
"error parsing options of {}: {}", argv[0], e.what());
3100 if (varmap.count(
"help")) {
3101 printHelp(varmap, executorOptions, physicalWorkflow, currentWorkflowOptions);
3107 if (varmap.count(
"severity")) {
3108 auto logLevel = varmap[
"severity"].as<std::string>();
3109 if (logLevel ==
"debug") {
3110 fair::Logger::SetConsoleSeverity(fair::Severity::debug);
3111 }
else if (logLevel ==
"detail") {
3112 fair::Logger::SetConsoleSeverity(fair::Severity::detail);
3113 }
else if (logLevel ==
"info") {
3114 fair::Logger::SetConsoleSeverity(fair::Severity::info);
3115 }
else if (logLevel ==
"warning") {
3116 fair::Logger::SetConsoleSeverity(fair::Severity::warning);
3117 }
else if (logLevel ==
"error") {
3118 fair::Logger::SetConsoleSeverity(fair::Severity::error);
3119 }
else if (logLevel ==
"important") {
3120 fair::Logger::SetConsoleSeverity(fair::Severity::important);
3121 }
else if (logLevel ==
"alarm") {
3122 fair::Logger::SetConsoleSeverity(fair::Severity::alarm);
3123 }
else if (logLevel ==
"critical") {
3124 fair::Logger::SetConsoleSeverity(fair::Severity::critical);
3125 }
else if (logLevel ==
"fatal") {
3126 fair::Logger::SetConsoleSeverity(fair::Severity::fatal);
3128 LOGP(error,
"Invalid log level '{}'", logLevel);
3135 auto evaluateBatchOption = [&varmap]() ->
bool {
3136 if (varmap.count(
"no-batch") > 0) {
3139 if (varmap.count(
"batch") == 0) {
3141 return isatty(fileno(stdout)) == 0;
3150 DriverInfo driverInfo{
3151 .sendingPolicies = sendingPolicies,
3152 .forwardingPolicies = forwardingPolicies,
3153 .callbacksPolicies = callbacksPolicies};
3154 driverInfo.states.reserve(10);
3155 driverInfo.sigintRequested =
false;
3156 driverInfo.sigchldRequested =
false;
3157 driverInfo.channelPolicies = channelPolicies;
3158 driverInfo.completionPolicies = completionPolicies;
3159 driverInfo.dispatchPolicies = dispatchPolicies;
3160 driverInfo.resourcePolicies = resourcePolicies;
3161 driverInfo.argc = argc;
3162 driverInfo.argv = argv;
3163 driverInfo.noSHMCleanup = varmap[
"no-cleanup"].as<
bool>();
3164 driverInfo.processingPolicies.termination = varmap[
"completion-policy"].as<
TerminationPolicy>();
3165 driverInfo.processingPolicies.earlyForward = varmap[
"early-forward-policy"].as<
EarlyForwardPolicy>();
3166 driverInfo.mode = varmap[
"driver-mode"].as<
DriverMode>();
3168 auto batch = evaluateBatchOption();
3171 .driverHasGUI = (batch ==
false) || getenv(
"DPL_DRIVER_REMOTE_GUI") !=
nullptr,
3174 if (varmap[
"error-policy"].defaulted() && driverConfig.batch ==
false) {
3175 driverInfo.processingPolicies.error = TerminationPolicy::WAIT;
3177 driverInfo.processingPolicies.error = varmap[
"error-policy"].as<
TerminationPolicy>();
3180 driverInfo.startTime = uv_hrtime();
3181 driverInfo.startTimeMsFromEpoch = std::chrono::duration_cast<std::chrono::milliseconds>(
3182 std::chrono::system_clock::now().time_since_epoch())
3184 driverInfo.timeout = varmap[
"timeout"].as<uint64_t>();
3185 driverInfo.deployHostname = varmap[
"hostname"].as<std::string>();
3186 driverInfo.resources = varmap[
"resources"].as<std::string>();
3187 driverInfo.resourcesMonitoringInterval = varmap[
"resources-monitoring"].as<
unsigned short>();
3188 driverInfo.resourcesMonitoringDumpInterval = varmap[
"resources-monitoring-dump-interval"].as<
unsigned short>();
3191 driverInfo.processorInfo = dataProcessorInfos;
3192 driverInfo.configContext = &configContext;
3199 std::string frameworkId;
3202 if (varmap.count(
"id")) {
3206 frameworkId = std::regex_replace(varmap[
"id"].as<std::string>(), std::regex{
"_dds.*"},
"");
3207 driverInfo.uniqueWorkflowId = fmt::format(
"{}", getppid());
3208 driverInfo.defaultDriverClient =
"stdout://";
3210 driverInfo.uniqueWorkflowId = fmt::format(
"{}", getpid());
3211 driverInfo.defaultDriverClient =
"ws://";