Skip to content

Commit aef5d05

Browse files
ktfktf
committed
DPL: add initial integration for O2 Control system
The configuration being generated is far from being complete, however it should be good enough to simplify integration.
1 parent 8665dc8 commit aef5d05

File tree

4 files changed

+158
-1
lines changed

4 files changed

+158
-1
lines changed

‎Framework/Core/CMakeLists.txt‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@ set(SRCS
4545
src/DeviceSpecHelpers.cxx
4646
src/DDSConfigHelpers.cxx
4747
src/Dispatcher.cxx
48+
src/O2ControlHelpers.cxx
4849
src/DriverControl.cxx
4950
src/DriverInfo.cxx
5051
src/FairOptionsRetriever.cxx
@@ -154,6 +155,7 @@ set(HEADERS
154155
include/Framework/RawBufferContext.h
155156
src/ComputingResource.h
156157
src/DDSConfigHelpers.h
158+
src/O2ControlHelpers.h
157159
src/DeviceSpecHelpers.h
158160
src/DriverControl.h
159161
src/DriverInfo.h

‎Framework/Core/src/O2ControlHelpers.cxx‎

Lines changed: 114 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,114 @@
1+
// Copyright CERN and copyright holders of ALICE O2. This software is
2+
// distributed under the terms of the GNU General Public License v3 (GPL
3+
// Version 3), copied verbatim in the file "COPYING".
4+
//
5+
// See http://alice-o2.web.cern.ch/license for full licensing information.
6+
//
7+
// In applying this license CERN does not waive the privileges and immunities
8+
// granted to it by virtue of its status as an Intergovernmental Organization
9+
// or submit itself to any jurisdiction.
10+
#include "O2ControlHelpers.h"
11+
#include "ChannelSpecHelpers.h"
12+
#include <map>
13+
#include <iostream>
14+
#include <cstring>
15+
16+
namespace o2
17+
{
18+
namespace framework
19+
{
20+
21+
void dumpDeviceSpec2O2Control(std::ostream& out,
22+
const std::vector<DeviceSpec>& specs,
23+
const std::vector<DeviceExecution>& executions)
24+
{
25+
out << R"(- o2:)"
26+
<< "\n";
27+
out << R"( tasks:)"
28+
<< "\n";
29+
assert(specs.size() == executions.size());
30+
31+
for (size_t di = 0; di < specs.size(); ++di) {
32+
auto& spec = specs[di];
33+
auto& execution = executions[di];
34+
35+
out << R"( - name: )" << spec.id << "\n";
36+
out << R"( control: )"
37+
<< "\n";
38+
out << R"( mode: "fairmq")"
39+
<< "\n";
40+
if (spec.outputChannels.empty()) {
41+
out << R"( bind: [])"
42+
<< "\n";
43+
} else {
44+
out << R"( bind: )"
45+
<< "\n";
46+
for (auto& channel : spec.outputChannels) {
47+
out << R"( - name: ")" << channel.name << "\"\n";
48+
out << R"( type: ")" << ChannelSpecHelpers::typeAsString(channel.type) << "\"\n";
49+
}
50+
}
51+
out << R"( command:)"
52+
<< "\n";
53+
out << R"( - shell: true)"
54+
<< "\n";
55+
out << R"( - value: )" << execution.args[0] << "\n";
56+
out << R"( - arguments:)"
57+
<< "\n";
58+
out << R"( - -b)"
59+
<< "\n";
60+
out << R"( - --monitoring-backend)"
61+
<< "\n";
62+
out << R"( - no-op://)"
63+
<< "\n";
64+
for (size_t ai = 1; ai < execution.args.size(); ++ai) {
65+
const char* option = execution.args[ai];
66+
const char* value = nullptr; // no value by default (i.e. a boolean)
67+
// If the subsequent option exists and does not start with -, we assume
68+
// it is an argument to the previous one.
69+
if (ai + 1 < execution.args.size() && execution.args[ai + 1][0] != '-') {
70+
value = execution.args[ai + 1];
71+
ai++;
72+
}
73+
if (!option) {
74+
break;
75+
}
76+
// Do not print out channel information
77+
if (strcmp(option, "--channel-config") == 0) {
78+
ai += 2;
79+
continue;
80+
} else if (strcmp(option, "--control") == 0) {
81+
continue;
82+
}
83+
out << R"( - )" << option << "\n";
84+
if (value) {
85+
out << R"( - )" << value << "\n";
86+
}
87+
}
88+
out << "\n";
89+
}
90+
out << " workflows:\n";
91+
out << " o2-workflow:\n";
92+
out << " name: \"o2-workflow-roles\"\n";
93+
out << " roles: \"\n";
94+
for (size_t di = 0; di < specs.size(); ++di) {
95+
auto& spec = specs[di];
96+
out << " - name: \"" << spec.name << "\"\n";
97+
out << " connect:\n";
98+
for (auto& channel : spec.inputChannels) {
99+
out << R"( - name: ")" << channel.name << "\"\n";
100+
// FIXME: Until we get a {{workflow}} placeholder.
101+
std::string sourceDevice = channel.name;
102+
sourceDevice.erase(0, 5);
103+
auto startSuffix = sourceDevice.find_last_of("_to_");
104+
sourceDevice = sourceDevice.substr(0, startSuffix - 3);
105+
out << R"( target: "{{parent}}.)" << sourceDevice << ":" << channel.name << "\"\n";
106+
out << R"( type: ")" << ChannelSpecHelpers::typeAsString(channel.type) << "\"\n";
107+
}
108+
out << " task:\n";
109+
out << " load: " << spec.name << "\n";
110+
}
111+
}
112+
113+
} // namespace framework
114+
} // namespace o2

‎Framework/Core/src/O2ControlHelpers.h‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
// Copyright CERN and copyright holders of ALICE O2. This software is
2+
// distributed under the terms of the GNU General Public License v3 (GPL
3+
// Version 3), copied verbatim in the file "COPYING".
4+
//
5+
// See http://alice-o2.web.cern.ch/license for full licensing information.
6+
//
7+
// In applying this license CERN does not waive the privileges and immunities
8+
// granted to it by virtue of its status as an Intergovernmental Organization
9+
// or submit itself to any jurisdiction.
10+
#ifndef FRAMEWORK_O2CONTROLHELPERS_H
11+
#define FRAMEWORK_O2CONTROLHELPERS_H
12+
13+
#include "Framework/DeviceSpec.h"
14+
#include "Framework/DeviceExecution.h"
15+
#include <vector>
16+
#include <iosfwd>
17+
18+
namespace o2
19+
{
20+
namespace framework
21+
{
22+
23+
void dumpDeviceSpec2O2Control(std::ostream& out,
24+
std::vector<DeviceSpec> const& specs,
25+
std::vector<DeviceExecution> const& executions);
26+
27+
} // namespace framework
28+
} // namespace o2
29+
#endif // FRAMEWORK_O2CONTROLHELPERS_H

‎Framework/Core/src/runDataProcessing.cxx‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
#include "Framework/WorkflowSpec.h"
3434

3535
#include "DDSConfigHelpers.h"
36+
#include "O2ControlHelpers.h"
3637
#include "DeviceSpecHelpers.h"
3738
#include "DriverControl.h"
3839
#include "DriverInfo.h"
@@ -966,6 +967,16 @@ void initialiseDriverControl(bpo::variables_map const& varmap, DriverControl& co
966967
DriverState::SCHEDULE, //
967968
DriverState::MATERIALISE_WORKFLOW //
968969
};
970+
} else if (varmap["o2-control"].as<bool>()) {
971+
control.callbacks = { [](DeviceSpecs const specs, DeviceExecutions const& executions) {
972+
dumpDeviceSpec2O2Control(std::cout, specs, executions);
973+
} };
974+
control.forcedTransitions = {
975+
DriverState::EXIT, //
976+
DriverState::PERFORM_CALLBACKS, //
977+
DriverState::SCHEDULE, //
978+
DriverState::MATERIALISE_WORKFLOW //
979+
};
969980
} else if (varmap.count("id")) {
970981
// FIXME: for the time being each child needs to recalculate the workflow,
971982
// so that it can understand what it needs to do. This is obviously
@@ -1013,7 +1024,8 @@ int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& workflow,
10131024
"what to do when processing is finished") //
10141025
("graphviz,g", bpo::value<bool>()->zero_tokens()->default_value(false), "produce graph output") //
10151026
("timeout,t", bpo::value<double>()->default_value(0), "timeout after which to exit") //
1016-
("dds,D", bpo::value<bool>()->zero_tokens()->default_value(false), "create DDS configuration");
1027+
("dds,D", bpo::value<bool>()->zero_tokens()->default_value(false), "create DDS configuration") //
1028+
("o2-control,o2", bpo::value<bool>()->zero_tokens()->default_value(false), "create O2 Control configuration");
10171029
// some of the options must be forwarded by default to the device
10181030
executorOptions.add(DeviceSpecHelpers::getForwardedDeviceOptions());
10191031

0 commit comments

Comments
 (0)