Line data Source code
1 : /**
2 : * @file DPDKReaderModule.cpp DPDKReaderModule DAQModule implementation
3 : *
4 : * This is part of the DUNE DAQ Application Framework, copyright 2020.
5 : * Licensing/copyright details are in the COPYING file that you should have
6 : * received with this code.
7 : */
8 : // #include "dpdklibs/nicreader/Nljs.hpp"
9 :
10 : #include "appfwk/ConfigurationManager.hpp"
11 :
12 : #include "appmodel/NetworkDetectorToDaqConnection.hpp"
13 :
14 : #include "appmodel/DPDKPortConfiguration.hpp"
15 : #include "appmodel/DPDKReaderConf.hpp"
16 : #include "appmodel/DataReaderModule.hpp"
17 : #include "confmodel/DetectorStream.hpp"
18 : #include "confmodel/HostCores.hpp"
19 : #include "confmodel/NetworkDevice.hpp"
20 : #include "confmodel/QueueWithSourceId.hpp"
21 :
22 : #include "logging/Logging.hpp"
23 :
24 : #include "datahandlinglibs/DataHandlingIssues.hpp"
25 : #include "datahandlinglibs/utils/BufferCopy.hpp"
26 :
27 : #include "CreateSource.hpp"
28 : #include "DPDKReaderModule.hpp"
29 : #include "dpdklibs/EALSetup.hpp"
30 : #include "dpdklibs/FlowControl.hpp"
31 : #include "dpdklibs/Issues.hpp"
32 : #include "dpdklibs/RTEIfaceSetup.hpp"
33 : #include "dpdklibs/udp/PacketCtor.hpp"
34 : #include "dpdklibs/udp/Utils.hpp"
35 :
36 : #include <chrono>
37 : #include <cinttypes>
38 : #include <ios>
39 : #include <memory>
40 : #include <sstream>
41 : #include <string>
42 : #include <thread>
43 : #include <utility>
44 : #include <vector>
45 :
46 : /**
47 : * @brief Name used by TRACE TLOG calls from this source file
48 : */
49 : #define TRACE_NAME "DPDKReaderModule" // NOLINT
50 :
51 : /**
52 : * @brief TRACE debug levels used in this source file
53 : */
54 : enum
55 : {
56 : TLVL_ENTER_EXIT_METHODS = 5,
57 : TLVL_WORK_STEPS = 10,
58 : TLVL_BOOKKEEPING = 15
59 : };
60 :
61 : namespace dunedaq {
62 : namespace dpdklibs {
63 :
64 0 : DPDKReaderModule::DPDKReaderModule(const std::string& name)
65 : : DAQModule(name)
66 0 : , m_run_marker{ false }
67 : {
68 0 : register_command("conf", &DPDKReaderModule::do_configure);
69 0 : register_command("start", &DPDKReaderModule::do_start);
70 0 : register_command("stop_trigger_sources", &DPDKReaderModule::do_stop);
71 0 : register_command("scrap", &DPDKReaderModule::do_scrap);
72 0 : }
73 :
74 0 : DPDKReaderModule::~DPDKReaderModule() {}
75 :
76 : inline void
77 : tokenize(std::string const& str, const char delim, std::vector<std::string>& out)
78 : {
79 : std::size_t start;
80 : std::size_t end = 0;
81 : while ((start = str.find_first_not_of(delim, end)) != std::string::npos) {
82 : end = str.find(delim, start);
83 : out.push_back(str.substr(start, end - start));
84 : }
85 : }
86 :
87 : void
88 0 : DPDKReaderModule::init(const std::shared_ptr<appfwk::ConfigurationManager> mcfg)
89 : {
90 0 : auto mdal = mcfg->get_dal<appmodel::DataReaderModule>(get_name());
91 0 : m_cfg = mcfg;
92 0 : if (mdal->get_raw_data_callbacks().empty()) {
93 0 : auto err = datahandlinglibs::InitializationError(ERS_HERE, "No outputs defined for NIC reader in configuration.");
94 0 : ers::fatal(err);
95 0 : throw err;
96 0 : }
97 :
98 : // Loop over output queues, extract source ids and create source model objects
99 0 : for (auto con : mdal->get_raw_data_callbacks()) {
100 : // TODO: add nullpointer check against misconfiguration
101 0 : auto ptr = m_sources[con->get_source_id()] = createSourceModel(con);
102 0 : register_node(con->UID(), ptr);
103 0 : }
104 0 : }
105 :
106 : void
107 0 : DPDKReaderModule::do_configure(const CommandData_t& /*args*/)
108 : {
109 0 : TLOG() << get_name() << ": Entering do_conf() method";
110 : // auto session = appfwk::ModuleManager::get()->get_session();
111 0 : auto mdal = m_cfg->get_dal<appmodel::DataReaderModule>(get_name());
112 0 : auto module_conf = mdal->get_configuration()->cast<appmodel::DPDKReaderConf>();
113 0 : auto res_set = mdal->get_connections();
114 : // EAL setup
115 0 : TLOG() << "Setting up EAL with params from config.";
116 0 : std::vector<std::string> eal_params;
117 0 : eal_params.push_back("eal_cmdline");
118 0 : eal_params.push_back("--proc-type=primary");
119 :
120 : // Construct the pcie devices allowed mask
121 0 : std::string first_pcie_addr;
122 0 : bool is_first_pcie_addr = true;
123 0 : std::deque<uint16_t> rte_cores;
124 :
125 0 : std::vector<const appmodel::NetworkDetectorToDaqConnection*> d2d_conns;
126 0 : for (auto res : res_set) {
127 0 : auto connection = res->cast<appmodel::NetworkDetectorToDaqConnection>();
128 0 : if (connection == nullptr) {
129 0 : datahandlinglibs::GenericConfigurationError err(
130 0 : ERS_HERE, "DetectorToDaqConnection configuration failed due expected but unavailable connection!");
131 0 : ers::fatal(err);
132 0 : throw err;
133 0 : }
134 0 : if (connection->is_excluded(*(m_cfg->get_session()))) {
135 0 : continue;
136 : }
137 :
138 0 : d2d_conns.push_back(connection);
139 :
140 0 : auto receiver = connection->receiver()->cast<appmodel::DPDKReceiver>();
141 0 : if (!receiver) {
142 0 : throw datahandlinglibs::InitializationError(
143 0 : ERS_HERE,
144 0 : fmt::format("Found {} of type {} in connection {} while expecting type DPDKReceiver",
145 0 : receiver->class_name(),
146 0 : receiver->UID(),
147 0 : connection->UID()));
148 : }
149 :
150 0 : auto net_device = receiver->get_uses()->cast<confmodel::NetworkDevice>();
151 :
152 0 : if (is_first_pcie_addr) {
153 0 : first_pcie_addr = net_device->get_pcie_addr();
154 : is_first_pcie_addr = false;
155 : }
156 0 : eal_params.push_back("-a");
157 0 : eal_params.push_back(net_device->get_pcie_addr());
158 :
159 0 : for (const auto* proc_res : receiver->get_configuration()->get_used_lcores()) {
160 0 : rte_cores.insert(rte_cores.end(), proc_res->get_cpu_cores().begin(), proc_res->get_cpu_cores().end());
161 : }
162 : }
163 :
164 0 : uint16_t main_core = rte_get_main_lcore();
165 0 : if (std::find(rte_cores.begin(), rte_cores.end(), main_core) != rte_cores.end()) {
166 0 : throw MainCoreConflict(ERS_HERE, main_core);
167 : }
168 0 : rte_cores.push_front(main_core);
169 :
170 0 : eal_params.push_back("-l");
171 0 : eal_params.push_back(fmt::format("{}", fmt::join(rte_cores, ",")));
172 :
173 : // Use the first pcie device id as file prefix
174 : // FIXME: Review this strategy - should work in most of cases, but it could be
175 : // confusing in configs with multiple connections
176 0 : eal_params.push_back(fmt::format("--file-prefix={}", first_pcie_addr));
177 :
178 0 : eal_params.push_back(module_conf->get_eal_args());
179 :
180 0 : ealutils::init_eal(eal_params);
181 :
182 : // Get available connections from EAL
183 0 : auto available_ifaces = ifaceutils::get_num_available_ifaces();
184 0 : TLOG() << "Number of available connections: " << available_ifaces;
185 0 : for (unsigned int ifc_id = 0; ifc_id < available_ifaces; ++ifc_id) {
186 0 : std::string mac_addr_str = ifaceutils::get_iface_mac_str(ifc_id);
187 0 : std::string pci_addr_str = ifaceutils::get_iface_pci_str(ifc_id);
188 0 : m_mac_to_id_map[mac_addr_str] = ifc_id;
189 : // TODO: remove
190 0 : m_pci_to_id_map[pci_addr_str] = ifc_id;
191 0 : TLOG() << "Available iface with MAC=" << mac_addr_str << " PCIe=" << pci_addr_str << " logical ID=" << ifc_id;
192 0 : }
193 :
194 0 : for (auto d2d_conn : d2d_conns) {
195 0 : auto dpdk_receiver = d2d_conn->get_net_receiver()->cast<appmodel::DPDKReceiver>();
196 :
197 : // Prepare the list of active senders (network transmitters) and active streams
198 0 : std::vector<const appmodel::NWDetDataSender*> nw_senders;
199 0 : std::vector<const confmodel::DetectorStream*> active_streams;
200 :
201 0 : for (auto nw_sender : d2d_conn->get_net_senders()) {
202 0 : TLOG() << "Sender " << nw_sender->UID() << "is " << nw_sender->is_excluded(*(m_cfg->get_session()));
203 :
204 0 : if (!nw_sender->is_excluded(*(m_cfg->get_session()))) {
205 0 : nw_senders.push_back(nw_sender);
206 :
207 0 : for (auto det_stream : nw_sender->get_streams()) {
208 0 : if (det_stream->is_excluded(*(m_cfg->get_session())))
209 0 : continue;
210 :
211 0 : active_streams.push_back(det_stream);
212 : }
213 : }
214 : }
215 :
216 0 : auto net_device = dpdk_receiver->get_uses();
217 :
218 0 : if ((m_mac_to_id_map.count(net_device->get_mac_address()) == 0) ||
219 0 : (m_pci_to_id_map.count(net_device->get_pcie_addr()) == 0)) {
220 0 : TLOG() << "No available interface with MAC=" << net_device->get_mac_address();
221 0 : throw datahandlinglibs::InitializationError(
222 0 : ERS_HERE, "DPDKReaderModule configuration failed due expected but unavailable interface!");
223 : }
224 :
225 0 : uint iface_id = m_mac_to_id_map[net_device->get_mac_address()];
226 0 : auto ptr = m_ifaces[iface_id] =
227 0 : std::make_shared<IfaceWrapper>(iface_id, dpdk_receiver, nw_senders, active_streams, m_sources, m_run_marker);
228 0 : register_node(fmt::format("interface-{}", iface_id), ptr);
229 0 : ptr->allocate_mbufs();
230 0 : ptr->setup_interface();
231 0 : ptr->setup_flow_steering();
232 0 : ptr->setup_xstats();
233 0 : }
234 :
235 0 : if (!m_run_marker.load()) {
236 0 : set_running(true);
237 0 : TLOG() << "Starting iface wrappers.";
238 0 : for (auto& [iface_id, iface] : m_ifaces) {
239 0 : iface->start();
240 : }
241 : } else {
242 0 : TLOG_DEBUG(5) << "iface wrappers are already running!";
243 : }
244 0 : }
245 :
246 : void
247 0 : DPDKReaderModule::do_start(const CommandData_t&)
248 : {
249 :
250 : // Setup callbacks on all sourcemodels
251 0 : for (auto& [sourceid, source] : m_sources) {
252 0 : source->acquire_callback();
253 : }
254 :
255 0 : for (auto& [iface_id, iface] : m_ifaces) {
256 0 : iface->enable_flow();
257 : }
258 0 : }
259 :
260 : void
261 0 : DPDKReaderModule::do_stop(const CommandData_t&)
262 : {
263 0 : for (auto& [iface_id, iface] : m_ifaces) {
264 0 : iface->disable_flow();
265 : }
266 0 : }
267 :
268 : void
269 0 : DPDKReaderModule::do_scrap(const CommandData_t&)
270 : {
271 0 : TLOG() << get_name() << ": Entering do_scrap() method";
272 0 : if (m_run_marker.load()) {
273 0 : TLOG() << "Raising stop through variables!";
274 0 : set_running(false);
275 0 : TLOG() << "Stopping iface wrappers.";
276 0 : for (auto& [iface_id, iface] : m_ifaces) {
277 0 : iface->stop_xstats();
278 0 : iface->stop();
279 : }
280 0 : ealutils::wait_for_lcores();
281 0 : TLOG() << "Stoppped DPDK lcore processors and internal threads...";
282 : } else {
283 0 : TLOG_DEBUG(5) << "DPDK lcore processor is already stopped!";
284 : }
285 0 : TLOG() << get_name() << ": do_scrap called. Tearing down EAL.";
286 0 : ealutils::finish_eal();
287 0 : }
288 :
289 : void
290 0 : DPDKReaderModule::set_running(bool should_run)
291 : {
292 0 : bool was_running = m_run_marker.exchange(should_run);
293 0 : TLOG_DEBUG(5) << "Active state was toggled from " << was_running << " to " << should_run;
294 0 : }
295 :
296 : } // namespace dpdklibs
297 : } // namespace dunedaq
298 :
299 0 : DEFINE_DUNE_DAQ_MODULE(dunedaq::dpdklibs::DPDKReaderModule)
|
