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 LOG(info) <<
"QUIT_REQUESTED";
2207 guiQuitRequested =
true;
2214 force_step_timer.data = &infos;
2216 driverInfo.states.push_back(DriverState::HANDLE_CHILDREN);
2218 case DriverState::HANDLE_CHILDREN: {
2222 uv_run(loop, once ? UV_RUN_ONCE : UV_RUN_NOWAIT);
2227 static bool forcefulExitMessage =
true;
2228 if (forcefulExitMessage) {
2229 LOG(info) <<
"Forceful exit requested.";
2230 forcefulExitMessage =
false;
2236 driverInfo.sigchldRequested =
false;
2241 bool supposedToQuit = (guiQuitRequested || canExit ||
graceful_exit);
2243 if (allChildrenGone && (supposedToQuit || driverInfo.processingPolicies.termination == TerminationPolicy::QUIT)) {
2245 driverInfo.states.resize(0);
2246 driverInfo.states.push_back(DriverState::EXIT);
2247 }
else if (hasError && driverInfo.processingPolicies.error == TerminationPolicy::QUIT && !supposedToQuit) {
2249 force_exit_timer.data = &infos;
2250 static bool forceful_timer_started =
false;
2251 if (forceful_timer_started ==
false) {
2252 forceful_timer_started =
true;
2255 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
2256 }
else if (allChildrenGone ==
false && supposedToQuit) {
2257 driverInfo.states.push_back(DriverState::HANDLE_CHILDREN);
2261 case DriverState::EXIT: {
2263 if (driverInfo.resourcesMonitoringDumpInterval) {
2264 uv_timer_stop(&metricDumpTimer);
2266 LOG(info) <<
"Dumping performance metrics to performanceMetrics.json file";
2269 dumpRunSummary(serverContext, driverInfo, infos, runningWorkflow.devices);
2276 if (infos.empty()) {
2279 boost::property_tree::ptree finalConfig;
2280 assert(infos.size() == runningWorkflow.devices.size());
2281 for (
size_t di = 0;
di < infos.size(); ++
di) {
2282 auto& info = infos[
di];
2283 auto& spec = runningWorkflow.devices[
di];
2284 finalConfig.put_child(spec.name, info.currentConfig);
2286 LOG(info) <<
"Dumping used configuration in dpl-config.json";
2288 std::ofstream outDPLConfigFile(
"dpl-config.json", std::ios::out);
2289 if (outDPLConfigFile.is_open()) {
2290 boost::property_tree::write_json(outDPLConfigFile, finalConfig);
2292 LOGP(warning,
"Could not write out final configuration file. Read only run folder?");
2294 if (driverInfo.noSHMCleanup) {
2295 LOGP(warning,
"Not cleaning up shared memory.");
2299 return calculateExitCode(driverInfo, runningWorkflow.devices, infos);
2301 case DriverState::PERFORM_CALLBACKS:
2302 for (
auto& callback : driverControl.
callbacks) {
2303 callback(workflow, runningWorkflow.devices, deviceExecutions, dataProcessorInfos, commandInfo);
2308 LOG(error) <<
"Driver transitioned in an unknown state("
2309 <<
"current: " << (
int)current
2310 <<
", previous: " << (
int)previous
2311 <<
"). Shutting down.";
2312 driverInfo.states.push_back(DriverState::QUIT_REQUESTED);
2850 std::vector<ChannelConfigurationPolicy>
const& channelPolicies,
2851 std::vector<CompletionPolicy>
const& completionPolicies,
2852 std::vector<DispatchPolicy>
const& dispatchPolicies,
2853 std::vector<ResourcePolicy>
const& resourcePolicies,
2854 std::vector<CallbacksPolicy>
const& callbacksPolicies,
2855 std::vector<SendingPolicy>
const& sendingPolicies,
2856 std::vector<ConfigParamSpec>
const& currentWorkflowOptions,
2857 std::vector<ConfigParamSpec>
const& detectedParams,
2862 if (getenv(
"DPL_DRIVER_SIGNPOSTS")) {
2866 std::vector<std::string> currentArgs;
2867 std::vector<PluginInfo> plugins;
2870 for (
int ai = 1; ai < argc; ++ai) {
2871 currentArgs.emplace_back(argv[ai]);
2877 currentWorkflowOptions};
2881 bpo::options_description executorOptions(
"Executor options");
2882 const char* helpDescription =
"print help: short, full, executor, or processor name";
2884 executorOptions.add_options()
2885 (
"help,h", bpo::value<std::string>()->implicit_value(
"short"), helpDescription)
2886 (
"quiet,q", bpo::value<bool>()->zero_tokens()->default_value(
false),
"quiet operation")
2887 (
"stop,s", bpo::value<bool>()->zero_tokens()->default_value(
false),
"stop before device start")
2888 (
"single-step", bpo::value<bool>()->zero_tokens()->default_value(
false),
"start in single step mode")
2889 (
"batch,b", bpo::value<std::vector<std::string>>()->zero_tokens()->composing(),
"batch processing mode")
2890 (
"no-batch", bpo::value<bool>()->zero_tokens(),
"force gui processing mode")
2891 (
"no-cleanup", bpo::value<bool>()->zero_tokens()->default_value(
false),
"do not cleanup the shm segment")
2892 (
"hostname", bpo::value<std::string>()->default_value(
"localhost"),
"hostname to deploy")
2893 (
"resources", bpo::value<std::string>()->default_value(
""),
"resources allocated for the workflow")
2894 (
"start-port,p", bpo::value<unsigned short>()->default_value(22000),
"start port to allocate")
2895 (
"port-range,pr", bpo::value<unsigned short>()->default_value(1000),
"ports in range")
2896 (
"completion-policy,c", bpo::value<TerminationPolicy>(&processingPolicies.
termination)->default_value(TerminationPolicy::QUIT),
2897 "what to do when processing is finished: quit, wait")
2898 (
"error-policy", bpo::value<TerminationPolicy>(&processingPolicies.
error)->default_value(TerminationPolicy::QUIT),
2899 "what to do when a device has an error: quit, wait")
2900 (
"min-failure-level", bpo::value<LogParsingHelpers::LogLevel>(&minFailureLevel)->default_value(LogParsingHelpers::LogLevel::Fatal),
2901 "minimum message level which will be considered as fatal and exit with 1")
2902 (
"graphviz,g", bpo::value<bool>()->zero_tokens()->default_value(
false),
"produce graphviz output")
2903 (
"mermaid", bpo::value<std::string>()->default_value(
""),
"produce graph output in mermaid format in file under specified name or on stdout if argument is \"-\"")
2904 (
"timeout,t", bpo::value<uint64_t>()->default_value(0),
"forced exit timeout (in seconds)")
2905 (
"dds,D", bpo::value<std::string>()->default_value(
""),
"create DDS configuration")
2906 (
"dds-workflow-suffix,D", bpo::value<std::string>()->default_value(
""),
"suffix for DDS names")
2907 (
"dump-workflow,dump", bpo::value<bool>()->zero_tokens()->default_value(
false),
"dump workflow as JSON")
2908 (
"dump-workflow-file", bpo::value<std::string>()->default_value(
"-"),
"file to which do the dump")
2909 (
"driver-mode", bpo::value<DriverMode>(&driverMode)->default_value(DriverMode::STANDALONE), R
"(how to run the driver. default: "standalone". Valid: "embedded")")
2910 (
"run", bpo::value<bool>()->zero_tokens()->default_value(
false),
"run workflow merged so far. It implies --batch. Use --no-batch to see the GUI")
2911 (
"no-IPC", bpo::value<bool>()->zero_tokens()->default_value(
false),
"disable IPC topology optimization")
2912 (
"o2-control,o2", bpo::value<std::string>()->default_value(
""),
"dump O2 Control workflow configuration under the specified name")
2913 (
"resources-monitoring", bpo::value<unsigned short>()->default_value(0),
"enable cpu/memory monitoring for provided interval in seconds")
2914 (
"resources-monitoring-dump-interval", bpo::value<unsigned short>()->default_value(0),
"dump monitoring information to disk every provided seconds");
2919 (
"id,i", bpo::value<std::string>(),
"device id for child spawning")
2920 (
"channel-config", bpo::value<std::vector<std::string>>(),
"channel configuration")
2921 (
"control",
"control plugin")
2922 (
"log-color",
"logging color scheme")(
"color",
"logging color scheme");
2924 bpo::options_description visibleOptions;
2925 visibleOptions.add(executorOptions);
2927 auto physicalWorkflow = workflow;
2928 std::map<std::string, size_t> rankIndex;
2935 size_t workflowHashA = 0;
2936 std::hash<std::string> hash_fn;
2938 for (
auto& dp : workflow) {
2939 workflowHashA += hash_fn(dp.name);
2942 for (
auto& dp : workflow) {
2943 rankIndex.insert(std::make_pair(dp.name, workflowHashA));
2946 std::vector<DataProcessorInfo> dataProcessorInfos;
2950 std::vector<DataProcessorSpec> importedWorkflow;
2952 if (previousWorked ==
false) {
2956 size_t workflowHashB = 0;
2957 for (
auto& dp : importedWorkflow) {
2958 workflowHashB += hash_fn(dp.name);
2965 for (
auto& dp : importedWorkflow) {
2966 auto found = std::find_if(physicalWorkflow.begin(), physicalWorkflow.end(),
2968 if (found == physicalWorkflow.end()) {
2969 physicalWorkflow.push_back(dp);
2970 rankIndex.insert(std::make_pair(dp.name, workflowHashB));
2979 for (
auto& dp : physicalWorkflow) {
2981 if (std::find_if(dp.labels.begin(), dp.labels.end(), isExpendable) != dp.labels.end()) {
2982 for (
auto&
output : dp.outputs) {
2983 if (
output.lifetime == Lifetime::Timeframe) {
2984 output.lifetime = Lifetime::Sporadic;
2995 auto reader = std::find_if(physicalWorkflow.begin(), physicalWorkflow.end(), [](
DataProcessorSpec& spec) { return spec.name ==
"internal-dpl-aod-reader"; });
2996 if (reader != physicalWorkflow.end()) {
2999 for (
auto& service : driverServices) {
3000 if (service.injectTopology ==
nullptr) {
3004 service.injectTopology(node, configContext);
3006 for (
auto& dp : physicalWorkflow) {
3007 if (dp.name.rfind(
"internal-", 0) == 0) {
3008 rankIndex.insert(std::make_pair(dp.name, hash_fn(
"internal")));
3015 return a.name < b.name;
3018 for (
auto& dp : physicalWorkflow) {
3019 std::stable_sort(dp.inputs.begin(), dp.inputs.end(),
3020 [](
InputSpec const&
a,
InputSpec const&
b) { return DataSpecUtils::describe(a) < DataSpecUtils::describe(b); });
3021 std::stable_sort(dp.outputs.begin(), dp.outputs.end(),
3022 [](
OutputSpec const&
a,
OutputSpec const&
b) { return DataSpecUtils::describe(a) < DataSpecUtils::describe(b); });
3027 std::vector<std::pair<int, int>> edges;
3029 if (physicalWorkflow.size() > 1) {
3033 if (topoInfos.size() != physicalWorkflow.size()) {
3036 throw std::runtime_error(
"Unable to do topological sort of the resulting workflow. Do you have loops?\n" +
debugTopoInfo(physicalWorkflow, topoInfos, edges));
3040 auto aRank = std::make_tuple(a.layer, -workflow.at(a.index).outputs.size(), workflow.at(a.index).name);
3041 auto bRank = std::make_tuple(b.layer, -workflow.at(b.index).outputs.size(), workflow.at(b.index).name);
3042 return aRank < bRank;
3045 std::vector<int> dataProcessorOrder;
3046 dataProcessorOrder.resize(topoInfos.size());
3047 for (
size_t i = 0;
i < topoInfos.size(); ++
i) {
3048 dataProcessorOrder[topoInfos[
i].index] =
i;
3050 std::vector<int> newLocations;
3051 newLocations.resize(dataProcessorOrder.size());
3052 for (
size_t i = 0;
i < dataProcessorOrder.size(); ++
i) {
3053 newLocations[dataProcessorOrder[
i]] =
i;
3063 bpo::options_description od;
3064 od.add(visibleOptions);
3069 using namespace bpo::command_line_style;
3070 auto style = (allow_short | short_allow_adjacent | short_allow_next | allow_long | long_allow_adjacent | long_allow_next | allow_sticky | allow_dash_for_short);
3071 bpo::variables_map varmap;
3074 bpo::command_line_parser(argc, argv)
3079 }
catch (std::exception
const& e) {
3080 LOGP(error,
"error parsing options of {}: {}", argv[0], e.what());
3096 if (varmap.count(
"help")) {
3097 printHelp(varmap, executorOptions, physicalWorkflow, currentWorkflowOptions);
3103 if (varmap.count(
"severity")) {
3104 auto logLevel = varmap[
"severity"].as<std::string>();
3105 if (logLevel ==
"debug") {
3106 fair::Logger::SetConsoleSeverity(fair::Severity::debug);
3107 }
else if (logLevel ==
"detail") {
3108 fair::Logger::SetConsoleSeverity(fair::Severity::detail);
3109 }
else if (logLevel ==
"info") {
3110 fair::Logger::SetConsoleSeverity(fair::Severity::info);
3111 }
else if (logLevel ==
"warning") {
3112 fair::Logger::SetConsoleSeverity(fair::Severity::warning);
3113 }
else if (logLevel ==
"error") {
3114 fair::Logger::SetConsoleSeverity(fair::Severity::error);
3115 }
else if (logLevel ==
"important") {
3116 fair::Logger::SetConsoleSeverity(fair::Severity::important);
3117 }
else if (logLevel ==
"alarm") {
3118 fair::Logger::SetConsoleSeverity(fair::Severity::alarm);
3119 }
else if (logLevel ==
"critical") {
3120 fair::Logger::SetConsoleSeverity(fair::Severity::critical);
3121 }
else if (logLevel ==
"fatal") {
3122 fair::Logger::SetConsoleSeverity(fair::Severity::fatal);
3124 LOGP(error,
"Invalid log level '{}'", logLevel);
3131 auto evaluateBatchOption = [&varmap]() ->
bool {
3132 if (varmap.count(
"no-batch") > 0) {
3135 if (varmap.count(
"batch") == 0) {
3137 return isatty(fileno(stdout)) == 0;
3146 DriverInfo driverInfo{
3147 .sendingPolicies = sendingPolicies,
3148 .forwardingPolicies = forwardingPolicies,
3149 .callbacksPolicies = callbacksPolicies};
3150 driverInfo.states.reserve(10);
3151 driverInfo.sigintRequested =
false;
3152 driverInfo.sigchldRequested =
false;
3153 driverInfo.channelPolicies = channelPolicies;
3154 driverInfo.completionPolicies = completionPolicies;
3155 driverInfo.dispatchPolicies = dispatchPolicies;
3156 driverInfo.resourcePolicies = resourcePolicies;
3157 driverInfo.argc = argc;
3158 driverInfo.argv = argv;
3159 driverInfo.noSHMCleanup = varmap[
"no-cleanup"].as<
bool>();
3160 driverInfo.processingPolicies.termination = varmap[
"completion-policy"].as<
TerminationPolicy>();
3161 driverInfo.processingPolicies.earlyForward = varmap[
"early-forward-policy"].as<
EarlyForwardPolicy>();
3162 driverInfo.mode = varmap[
"driver-mode"].as<
DriverMode>();
3164 auto batch = evaluateBatchOption();
3167 .driverHasGUI = (batch ==
false) || getenv(
"DPL_DRIVER_REMOTE_GUI") !=
nullptr,
3170 if (varmap[
"error-policy"].defaulted() && driverConfig.batch ==
false) {
3171 driverInfo.processingPolicies.error = TerminationPolicy::WAIT;
3173 driverInfo.processingPolicies.error = varmap[
"error-policy"].as<
TerminationPolicy>();
3176 driverInfo.startTime = uv_hrtime();
3177 driverInfo.startTimeMsFromEpoch = std::chrono::duration_cast<std::chrono::milliseconds>(
3178 std::chrono::system_clock::now().time_since_epoch())
3180 driverInfo.timeout = varmap[
"timeout"].as<uint64_t>();
3181 driverInfo.deployHostname = varmap[
"hostname"].as<std::string>();
3182 driverInfo.resources = varmap[
"resources"].as<std::string>();
3183 driverInfo.resourcesMonitoringInterval = varmap[
"resources-monitoring"].as<
unsigned short>();
3184 driverInfo.resourcesMonitoringDumpInterval = varmap[
"resources-monitoring-dump-interval"].as<
unsigned short>();
3187 driverInfo.processorInfo = dataProcessorInfos;
3188 driverInfo.configContext = &configContext;
3195 std::string frameworkId;
3198 if (varmap.count(
"id")) {
3202 frameworkId = std::regex_replace(varmap[
"id"].as<std::string>(), std::regex{
"_dds.*"},
"");
3203 driverInfo.uniqueWorkflowId = fmt::format(
"{}", getppid());
3204 driverInfo.defaultDriverClient =
"stdout://";
3206 driverInfo.uniqueWorkflowId = fmt::format(
"{}", getpid());
3207 driverInfo.defaultDriverClient =
"ws://";