Line data Source code
1 : /**
2 : * @file SocketReaderModule.cpp Boost.Asio-based socket reader plugin for low-bandwidth devices
3 : *
4 : * This is part of the DUNE DAQ , copyright 2020.
5 : * Licensing/copyright details are in the COPYING file that you should have
6 : * received with this code.
7 : */
8 :
9 : #include "SocketReaderModule.hpp"
10 :
11 : #include "CreateSource.hpp"
12 :
13 : #include "appmodel/DataReaderModule.hpp"
14 : #include "appmodel/NWDetDataReceiver.hpp"
15 : #include "appmodel/SocketDataSender.hpp"
16 : #include "appmodel/SocketDetectorToDaqConnection.hpp"
17 : #include "confmodel/DetectorStream.hpp"
18 : #include "confmodel/GeoId.hpp"
19 : #include "confmodel/NetworkInterface.hpp"
20 : #include "confmodel/QueueWithSourceId.hpp"
21 :
22 : #include "datahandlinglibs/DataHandlingIssues.hpp"
23 :
24 : #include "asiolibs/opmon/SocketReaderModule.pb.h"
25 :
26 : #include <memory>
27 : #include <string>
28 : #include <utility>
29 : #include <vector>
30 :
31 : namespace dunedaq::asiolibs {
32 :
33 0 : SocketReaderModule::SocketReaderModule(const std::string& name)
34 : : DAQModule(name)
35 0 : , m_work_guard(boost::asio::make_work_guard(m_io_context))
36 : {
37 0 : register_command("conf", &SocketReaderModule::do_configure);
38 0 : register_command("start", &SocketReaderModule::do_start);
39 0 : register_command("stop_trigger_sources", &SocketReaderModule::do_stop);
40 0 : }
41 :
42 : void
43 0 : SocketReaderModule::init(const std::shared_ptr<appfwk::ConfigurationManager> mcfg)
44 : {
45 0 : auto* mdal = mcfg->get_dal<appmodel::DataReaderModule>(get_name());
46 :
47 0 : if (mdal->get_raw_data_callbacks().empty()) {
48 0 : auto err =
49 0 : datahandlinglibs::InitializationError(ERS_HERE, "No outputs defined for socket reader in configuration.");
50 0 : ers::fatal(err);
51 0 : throw err;
52 0 : }
53 :
54 : // Loop over output queues, extract source ids and create source model objects
55 0 : for (auto con : mdal->get_raw_data_callbacks()) {
56 : // TODO: add nullpointer check against misconfiguration
57 0 : auto ptr = m_sources[con->get_source_id()] = createSourceModel(con);
58 0 : register_node(con->UID(), ptr);
59 0 : }
60 :
61 0 : auto* d2d_conn = mdal->get_connections()[0]; // there's only 1 connection
62 0 : auto* socket_d2d_conn = d2d_conn->cast<appmodel::SocketDetectorToDaqConnection>();
63 0 : if (socket_d2d_conn == nullptr) {
64 0 : auto err =
65 0 : datahandlinglibs::InitializationError(ERS_HERE, "Connection is not of type SocketDetectorToDaqConnection.");
66 0 : ers::fatal(err);
67 0 : throw err;
68 0 : }
69 :
70 0 : auto nw_receiver = socket_d2d_conn->get_net_receiver();
71 0 : auto local_ip = nw_receiver->get_uses()->get_ip_address()[0];
72 :
73 0 : for (auto nw_sender : socket_d2d_conn->get_net_senders()) {
74 0 : if (nw_sender->is_excluded(*(mcfg->get_session()))) {
75 0 : continue;
76 : }
77 :
78 0 : auto* socket_sender = nw_sender->cast<appmodel::SocketDataSender>();
79 0 : if (socket_sender == nullptr) {
80 0 : auto err = datahandlinglibs::InitializationError(ERS_HERE, "Sender is not of type SocketDataSender.");
81 0 : ers::fatal(err);
82 0 : throw err;
83 0 : }
84 :
85 0 : auto local_port = socket_sender->get_remote_port();
86 :
87 0 : m_reader_infos.emplace_back(std::make_shared<ReaderInfo>(local_ip, local_port, std::make_shared<SocketStats>()));
88 :
89 0 : if (string_to_socket_type(socket_sender->get_socket_type()) == SocketType::TCP) {
90 0 : m_readers.push_back(std::make_shared<std::variant<TCPReader, UDPReader>>(TCPReader{}));
91 : } else {
92 0 : m_readers.push_back(std::make_shared<std::variant<TCPReader, UDPReader>>(UDPReader{}));
93 : }
94 :
95 0 : auto remote_ip = socket_sender->get_uses()->get_ip_address()[0];
96 0 : auto remote_port = socket_sender->get_local_port();
97 :
98 : // Loop over streams
99 0 : for (auto det_stream : socket_sender->get_streams()) {
100 0 : if (det_stream->is_excluded(*(mcfg->get_session()))) {
101 0 : continue;
102 : }
103 :
104 0 : remote_stream_pair_t sender_stream_pair = { { remote_ip, remote_port },
105 0 : det_stream->get_geo_id()->get_stream_id() };
106 0 : m_remote_to_source[sender_stream_pair] = m_sources[det_stream->get_source_id()];
107 0 : }
108 0 : }
109 0 : }
110 :
111 : SocketReaderModule::SocketType
112 0 : SocketReaderModule::string_to_socket_type(const std::string& socket_type) const
113 : {
114 0 : if (socket_type == "TCP") {
115 : return SocketReaderModule::SocketType::TCP;
116 0 : } else if (socket_type == "UDP") {
117 0 : return SocketReaderModule::SocketType::UDP;
118 : }
119 : return SocketReaderModule::SocketType::INVALID;
120 : }
121 :
122 : void
123 0 : SocketReaderModule::do_configure(const CommandData_t&)
124 : {
125 0 : for (std::size_t i = 0; i < m_readers.size(); ++i) {
126 0 : auto reader_info = m_reader_infos[i];
127 0 : std::visit([this, reader_info](auto& reader) { reader.configure(m_io_context, reader_info); }, *m_readers[i]);
128 0 : }
129 0 : }
130 :
131 : void
132 0 : SocketReaderModule::do_start(const CommandData_t&)
133 : {
134 0 : for (const auto& reader_info : m_reader_infos) {
135 : // Reset opmon variables
136 0 : reader_info->socket_stats->packets_received = 0;
137 0 : reader_info->socket_stats->bytes_received = 0;
138 0 : reader_info->socket_stats->stats_packet_count = 0;
139 : }
140 :
141 : // Setup callbacks on all sourcemodels
142 0 : for (auto& [_, source] : m_sources) {
143 0 : source->acquire_callback();
144 : }
145 :
146 0 : m_io_thread = std::jthread([this] { m_io_context.run(); });
147 :
148 0 : for (auto& reader : m_readers) {
149 0 : boost::asio::co_spawn(m_io_context,
150 0 : std::visit([this](auto& reader) { return reader.start(m_remote_to_source); }, *reader),
151 : boost::asio::detached);
152 : }
153 0 : }
154 :
155 : void
156 0 : SocketReaderModule::do_stop(const CommandData_t&)
157 : {
158 0 : for (auto& reader : m_readers) {
159 0 : std::visit([](auto& reader) { reader.stop(); }, *reader);
160 : }
161 :
162 0 : m_work_guard.reset();
163 0 : }
164 :
165 : void
166 0 : SocketReaderModule::generate_opmon_data()
167 : {
168 0 : for (const auto& reader_info : m_reader_infos) {
169 0 : opmon::SocketReaderStats stats;
170 0 : stats.set_packets_received(reader_info->socket_stats->packets_received.load());
171 0 : stats.set_bytes_received(reader_info->socket_stats->bytes_received.load());
172 0 : publish(std::move(stats),
173 0 : { { "socket-reader", reader_info->local_ip + ":" + std::to_string(reader_info->local_port) } });
174 0 : }
175 0 : }
176 :
177 : void
178 0 : SocketReaderModule::TCPReader::configure(boost::asio::io_context& io_context, std::shared_ptr<ReaderInfo> reader_info)
179 : {
180 0 : m_socket_stats = reader_info->socket_stats;
181 :
182 0 : m_socket = std::make_unique<boost::asio::ip::tcp::socket>(io_context);
183 :
184 0 : boost::asio::ip::address local_address;
185 :
186 0 : try {
187 0 : local_address = boost::asio::ip::make_address(reader_info->local_ip);
188 0 : } catch (const boost::system::system_error& e) {
189 0 : TLOG() << "Failed to configure TCP socket: " << e.what() << " (Local IP: " << reader_info->local_ip << ")";
190 0 : throw;
191 0 : }
192 :
193 0 : boost::asio::ip::tcp::acceptor acceptor(io_context,
194 0 : boost::asio::ip::tcp::endpoint(local_address, reader_info->local_port));
195 :
196 0 : TLOG() << "Waiting for TCP connection at " << reader_info->local_ip << ":" << reader_info->local_port;
197 :
198 0 : acceptor.accept(*m_socket);
199 :
200 0 : m_remote = { m_socket->remote_endpoint().address().to_string(), m_socket->remote_endpoint().port() };
201 :
202 0 : TLOG() << "Established TCP connection from " << m_socket->local_endpoint().address().to_string() << ":"
203 0 : << m_socket->local_endpoint().port() << " to " << m_remote.first << ":" << m_remote.second;
204 0 : }
205 :
206 : boost::asio::awaitable<void>
207 0 : SocketReaderModule::TCPReader::start(const remote_source_map_t& remote_to_source)
208 : {
209 : const auto buffer_size = remote_to_source.begin()->second->get_expected_frame_size();
210 : std::vector<char> buffer(buffer_size);
211 :
212 : while (m_socket->is_open()) {
213 : const auto bytes_received =
214 : co_await boost::asio::async_read(*m_socket, boost::asio::buffer(buffer), boost::asio::use_awaitable);
215 :
216 : const auto* daq_header = reinterpret_cast<const dunedaq::detdataformats::DAQEthHeader*>(buffer.data());
217 :
218 : auto stream_id = static_cast<unsigned>(daq_header->stream_id);
219 : remote_stream_pair_t sender_stream_pair = { m_remote, stream_id };
220 :
221 : auto src_it = remote_to_source.find(sender_stream_pair);
222 : if (src_it == remote_to_source.end()) {
223 0 : TLOG() << "Unexpected sender-stream combination! (" << m_remote.first << ":" << m_remote.second << ", "
224 : << stream_id << ")";
225 : continue;
226 : }
227 :
228 : src_it->second->handle_daq_frame(buffer.data());
229 :
230 : ++m_socket_stats->packets_received;
231 : m_socket_stats->bytes_received.fetch_add(bytes_received);
232 : ++m_socket_stats->stats_packet_count;
233 : }
234 0 : }
235 :
236 : void
237 0 : SocketReaderModule::TCPReader::stop()
238 : {
239 0 : if (m_socket && m_socket->is_open()) {
240 0 : m_socket->close();
241 : }
242 0 : TLOG() << "Closed TCP connection";
243 0 : }
244 :
245 : void
246 0 : SocketReaderModule::UDPReader::configure(boost::asio::io_context& io_context, std::shared_ptr<ReaderInfo> reader_info)
247 : {
248 0 : m_socket_stats = reader_info->socket_stats;
249 :
250 0 : try {
251 0 : boost::asio::ip::udp::endpoint writer_endpoint(boost::asio::ip::address::from_string(reader_info->local_ip),
252 0 : reader_info->local_port);
253 :
254 0 : m_socket = std::make_unique<boost::asio::ip::udp::socket>(io_context, writer_endpoint);
255 :
256 0 : TLOG() << "Created UDP socket on " << m_socket->local_endpoint().address() << ":"
257 0 : << m_socket->local_endpoint().port();
258 :
259 0 : } catch (const boost::system::system_error& e) {
260 0 : TLOG() << "Failed to configure UDP socket: " << e.what() << " (Local IP: " << reader_info->local_ip
261 0 : << ", Local port: " << reader_info->local_port << ")";
262 0 : throw;
263 0 : }
264 0 : }
265 :
266 : boost::asio::awaitable<void>
267 0 : SocketReaderModule::UDPReader::start(const remote_source_map_t& remote_to_source)
268 : {
269 : const auto buffer_size = remote_to_source.begin()->second->get_expected_frame_size();
270 : std::vector<char> buffer(buffer_size);
271 : boost::asio::ip::udp::endpoint sender_endpoint;
272 :
273 : while (m_socket->is_open()) {
274 : std::size_t bytes_received =
275 : co_await m_socket->async_receive_from(boost::asio::buffer(buffer), sender_endpoint, boost::asio::use_awaitable);
276 :
277 : const auto* daq_header = reinterpret_cast<const dunedaq::detdataformats::DAQEthHeader*>(buffer.data());
278 :
279 : auto stream_id = static_cast<unsigned>(daq_header->stream_id);
280 :
281 : remote_t remote = { sender_endpoint.address().to_string(), sender_endpoint.port() };
282 : remote_stream_pair_t sender_stream_pair = { remote, stream_id };
283 :
284 : auto src_it = remote_to_source.find(sender_stream_pair);
285 :
286 : if (src_it == remote_to_source.end()) {
287 0 : TLOG() << "Unexpected sender-stream combination! (" << remote.first << ":" << remote.second << ", " << stream_id
288 : << ")";
289 : continue;
290 : }
291 :
292 : if (bytes_received == buffer_size) [[likely]] {
293 : src_it->second->handle_daq_frame(buffer.data());
294 : } else {
295 0 : TLOG() << "Payload is smaller than " << buffer_size << " (" << bytes_received << ")";
296 : }
297 :
298 : ++m_socket_stats->packets_received;
299 : m_socket_stats->bytes_received.fetch_add(bytes_received);
300 : ++m_socket_stats->stats_packet_count;
301 : }
302 0 : }
303 :
304 : void
305 0 : SocketReaderModule::UDPReader::stop()
306 : {
307 0 : if (m_socket && m_socket->is_open()) {
308 0 : m_socket->close();
309 : }
310 0 : TLOG() << "Closed UDP socket";
311 0 : }
312 :
313 : } // namespace dunedaq::asiolibs
314 :
315 0 : DEFINE_DUNE_DAQ_MODULE(dunedaq::asiolibs::SocketReaderModule)
|
