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