Line data Source code
1 : /**
2 : * @file SocketWriterModule.cpp Boost.Asio-based socket writer 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 "SocketWriterModule.hpp"
10 :
11 : #include "CreateGenericReceiver.hpp"
12 :
13 : #include "appfwk/ConfigurationManager.hpp"
14 : #include "appmodel/NWDetDataReceiver.hpp"
15 : #include "appmodel/SocketDataSender.hpp"
16 : #include "appmodel/SocketDataWriterModule.hpp"
17 : #include "appmodel/SocketDetectorToDaqConnection.hpp"
18 : #include "confmodel/DetectorStream.hpp"
19 : #include "confmodel/NetworkInterface.hpp"
20 : #include "confmodel/QueueWithSourceId.hpp"
21 :
22 : #include "datahandlinglibs/DataHandlingIssues.hpp"
23 :
24 : #include "asiolibs/opmon/SocketWriterModule.pb.h"
25 :
26 : #include "asiolibs/AsioIssues.hpp"
27 :
28 : #include <memory>
29 : #include <string>
30 : #include <utility>
31 : #include <vector>
32 :
33 : namespace dunedaq::asiolibs {
34 :
35 0 : SocketWriterModule::SocketWriterModule(const std::string& name)
36 : : DAQModule(name)
37 0 : , m_work_guard(boost::asio::make_work_guard(m_io_context))
38 : {
39 0 : register_command("conf", &SocketWriterModule::do_configure);
40 0 : register_command("start", &SocketWriterModule::do_start);
41 0 : register_command("stop_trigger_sources", &SocketWriterModule::do_stop);
42 0 : }
43 :
44 : void
45 0 : SocketWriterModule::get_dal_inputs(const dunedaq::appmodel::SocketDataWriterModule* mdal)
46 : {
47 0 : auto* con = mdal->get_inputs()[0]; // there's only 1 input
48 0 : auto* queue = con->cast<confmodel::Queue>();
49 0 : if (queue == nullptr) {
50 0 : auto err = datahandlinglibs::InitializationError(ERS_HERE, "Input is not of type Queue.");
51 0 : ers::fatal(err);
52 0 : throw err;
53 0 : }
54 :
55 0 : auto connection_name = queue->UID();
56 :
57 0 : const auto recv_timeout_ms = con->get_recv_timeout_ms();
58 0 : if (recv_timeout_ms == 0) {
59 0 : ers::warning(InvalidRawReceiverTimeout(ERS_HERE, m_receiver_timeout_ms.count()));
60 : } else {
61 0 : m_receiver_timeout_ms = std::chrono::milliseconds(recv_timeout_ms);
62 : }
63 :
64 0 : m_receiver = createGenericReceiver(connection_name);
65 : // Input connection sensibility check
66 0 : if (m_receiver == nullptr) {
67 : // ers::error(datahandlinglibs::ConfigurationError(ERS_HERE, sid, "Non callback mode, and receiver is unset!"));
68 : // FIXME (DTE)
69 : }
70 0 : }
71 :
72 : void
73 0 : SocketWriterModule::init(const std::shared_ptr<appfwk::ConfigurationManager> mcfg)
74 : {
75 0 : auto* mdal = mcfg->get_dal<appmodel::SocketDataWriterModule>(get_name());
76 :
77 0 : get_dal_inputs(mdal);
78 :
79 0 : auto* d2d_conn = mdal->get_connection();
80 0 : auto* socket_d2d_conn = d2d_conn->cast<appmodel::SocketDetectorToDaqConnection>();
81 0 : if (socket_d2d_conn == nullptr) {
82 0 : auto err =
83 0 : datahandlinglibs::InitializationError(ERS_HERE, "Connection is not of type SocketDetectorToDaqConnection.");
84 0 : ers::fatal(err);
85 0 : throw err;
86 0 : }
87 :
88 0 : auto nw_receiver = socket_d2d_conn->get_net_receiver();
89 0 : auto remote_ip = nw_receiver->get_uses()->get_ip_address()[0];
90 :
91 0 : auto* nw_sender = socket_d2d_conn->get_net_senders()[0]; // there's only 1 sender
92 :
93 0 : auto* socket_sender = nw_sender->cast<appmodel::SocketDataSender>();
94 0 : if (socket_sender == nullptr) {
95 0 : auto err = datahandlinglibs::InitializationError(ERS_HERE, "Sender is not of type SocketDataSender.");
96 0 : ers::fatal(err);
97 0 : throw err;
98 0 : }
99 :
100 0 : auto local_ip = socket_sender->get_uses()->get_ip_address()[0];
101 0 : auto local_port = socket_sender->get_local_port();
102 0 : auto remote_port = socket_sender->get_remote_port();
103 :
104 0 : m_writer_info =
105 0 : std::make_shared<WriterInfo>(local_ip, local_port, remote_ip, remote_port, std::make_shared<SocketStats>());
106 :
107 0 : if (string_to_socket_type(socket_sender->get_socket_type()) == SocketType::TCP) {
108 0 : m_writer = std::make_shared<std::variant<TCPWriter, UDPWriter>>(TCPWriter{});
109 : } else {
110 0 : m_writer = std::make_shared<std::variant<TCPWriter, UDPWriter>>(UDPWriter{});
111 : }
112 0 : }
113 :
114 : SocketWriterModule::SocketType
115 0 : SocketWriterModule::string_to_socket_type(const std::string& socket_type) const
116 : {
117 0 : if (socket_type == "TCP") {
118 : return SocketWriterModule::SocketType::TCP;
119 0 : } else if (socket_type == "UDP") {
120 0 : return SocketWriterModule::SocketType::UDP;
121 : }
122 : return SocketWriterModule::SocketType::INVALID;
123 : }
124 :
125 : void
126 0 : SocketWriterModule::do_configure(const CommandData_t&)
127 : {
128 0 : std::visit([this](auto& writer) { writer.configure(m_io_context, m_writer_info); }, *m_writer);
129 0 : }
130 :
131 : void
132 0 : SocketWriterModule::do_start(const CommandData_t&)
133 : {
134 : // Reset opmon variables
135 0 : m_writer_info->socket_stats->sum_payloads = 0;
136 0 : m_writer_info->socket_stats->num_payloads = 0;
137 0 : m_writer_info->socket_stats->sum_bytes = 0;
138 0 : m_writer_info->socket_stats->stats_packet_count = 0;
139 :
140 0 : m_t0 = std::chrono::steady_clock::now();
141 :
142 0 : m_run_marker.store(true);
143 0 : m_consumer_thread.set_work(&SocketWriterModule::run_consume, this);
144 :
145 0 : m_io_thread = std::jthread([this] { m_io_context.run(); });
146 0 : }
147 :
148 : void
149 0 : SocketWriterModule::run_consume()
150 : {
151 0 : TLOG() << "Consumer thread started..."; // TODO (DTE): Make debug logs
152 :
153 0 : while (m_run_marker.load()) {
154 : // Try to acquire data
155 0 : if (auto opt_payload = m_receiver->try_receive(m_receiver_timeout_ms)) {
156 0 : std::visit([&](auto& writer) { writer.enqueue(std::move(*opt_payload)); }, *m_writer);
157 : } else {
158 0 : std::visit([&](auto& writer) { ++writer.get_socket_stats()->queue_timeout_count; }, *m_writer);
159 0 : }
160 : }
161 :
162 0 : TLOG() << "Consumer thread joins... ";
163 0 : }
164 :
165 : void
166 0 : SocketWriterModule::do_stop(const CommandData_t&)
167 : {
168 0 : m_run_marker.store(false);
169 0 : while (!m_consumer_thread.get_readiness()) {
170 0 : std::this_thread::sleep_for(std::chrono::milliseconds(10));
171 : }
172 :
173 0 : std::visit([this](auto& writer) { writer.stop(); }, *m_writer);
174 :
175 0 : m_work_guard.reset();
176 0 : }
177 :
178 : void
179 0 : SocketWriterModule::generate_opmon_data()
180 : {
181 0 : opmon::SocketWriterStats stats;
182 0 : stats.set_sum_payloads(m_writer_info->socket_stats->sum_payloads.load());
183 0 : stats.set_num_payloads(m_writer_info->socket_stats->num_payloads.exchange(0));
184 0 : stats.set_sum_bytes(m_writer_info->socket_stats->sum_bytes.load());
185 :
186 0 : auto now = std::chrono::steady_clock::now();
187 0 : int new_packets = m_writer_info->socket_stats->stats_packet_count.exchange(0);
188 0 : double seconds = std::chrono::duration_cast<std::chrono::microseconds>(now - m_t0).count() / 1000000.;
189 0 : m_t0 = now;
190 :
191 0 : stats.set_rate_payloads_consumed(new_packets / seconds / 1000.);
192 :
193 0 : publish(std::move(stats),
194 0 : { { "socket-writer", m_writer_info->local_ip + ":" + std::to_string(m_writer_info->local_port) } });
195 0 : }
196 :
197 : void
198 0 : SocketWriterModule::TCPWriter::configure(boost::asio::io_context& io_context, std::shared_ptr<WriterInfo> writer_info)
199 : {
200 0 : m_io_context = &io_context;
201 0 : m_strand = std::make_shared<boost::asio::strand<boost::asio::io_context::executor_type>>(io_context.get_executor());
202 :
203 0 : m_socket_stats = writer_info->socket_stats;
204 :
205 0 : boost::asio::ip::address local_address;
206 0 : boost::asio::ip::address remote_address;
207 :
208 0 : try {
209 0 : local_address = boost::asio::ip::make_address(writer_info->local_ip);
210 0 : remote_address = boost::asio::ip::make_address(writer_info->remote_ip);
211 0 : } catch (const boost::system::system_error& e) {
212 0 : TLOG() << "Failed to configure TCP socket: " << e.what() << " (Local IP: " << writer_info->local_ip
213 0 : << ", Remote IP: " << writer_info->remote_ip << ")";
214 0 : throw;
215 0 : }
216 :
217 0 : while (true) {
218 0 : try {
219 0 : m_socket = std::make_unique<boost::asio::ip::tcp::socket>(io_context);
220 0 : m_socket->open(boost::asio::ip::tcp::v4());
221 0 : m_socket->bind(boost::asio::ip::tcp::endpoint(local_address, writer_info->local_port));
222 0 : m_socket->connect(boost::asio::ip::tcp::endpoint(remote_address, writer_info->remote_port));
223 0 : break;
224 0 : } catch (const boost::system::system_error& e) {
225 0 : TLOG() << "Failed to create TCP socket: " << e.what() << ". Retrying in 1 second..."
226 0 : << " (Local IP: " << writer_info->local_ip << ", Local port: " << writer_info->local_port
227 0 : << ", Remote IP: " << writer_info->remote_ip << ", Remote port: " << writer_info->remote_port << ")";
228 0 : std::this_thread::sleep_for(std::chrono::seconds(1));
229 0 : }
230 : }
231 :
232 0 : TLOG() << "Established TCP connection from " << m_socket->local_endpoint().address().to_string() << ":"
233 0 : << m_socket->local_endpoint().port() << " to " << m_socket->remote_endpoint().address().to_string() << ":"
234 0 : << m_socket->remote_endpoint().port();
235 0 : }
236 :
237 : void
238 0 : SocketWriterModule::TCPWriter::enqueue(GenericReceiverConcept::TypeErasedPayload&& payload)
239 : {
240 0 : boost::asio::post(*m_strand, [this, payload = std::move(payload)]() mutable {
241 0 : m_payloads.push(std::move(payload));
242 0 : if (m_payloads.size() == 1) { // queue was empty
243 0 : boost::asio::co_spawn(*m_strand, start(), boost::asio::detached);
244 : }
245 0 : });
246 0 : }
247 :
248 : boost::asio::awaitable<void>
249 0 : SocketWriterModule::TCPWriter::start()
250 : {
251 : while (!m_payloads.empty()) {
252 : auto payload = std::move(m_payloads.front());
253 :
254 : auto bytes_sent = co_await boost::asio::async_write(
255 : *m_socket, boost::asio::buffer(payload.data, payload.size), boost::asio::use_awaitable);
256 : m_payloads.pop();
257 :
258 : ++m_socket_stats->num_payloads;
259 : ++m_socket_stats->sum_payloads;
260 : m_socket_stats->sum_bytes.fetch_add(bytes_sent);
261 : ++m_socket_stats->stats_packet_count;
262 : }
263 0 : }
264 :
265 : void
266 0 : SocketWriterModule::TCPWriter::stop()
267 : {
268 0 : if (m_socket && m_socket->is_open()) {
269 0 : m_socket->close();
270 : }
271 0 : TLOG() << "Closed TCP connection";
272 0 : }
273 :
274 : std::shared_ptr<SocketWriterModule::SocketStats>
275 0 : SocketWriterModule::TCPWriter::get_socket_stats() const
276 : {
277 0 : return m_socket_stats;
278 : }
279 :
280 : void
281 0 : SocketWriterModule::UDPWriter::configure(boost::asio::io_context& io_context, std::shared_ptr<WriterInfo> writer_info)
282 : {
283 0 : m_io_context = &io_context;
284 0 : m_strand = std::make_shared<boost::asio::strand<boost::asio::io_context::executor_type>>(io_context.get_executor());
285 :
286 0 : m_socket_stats = writer_info->socket_stats;
287 :
288 0 : try {
289 0 : boost::asio::ip::udp::endpoint writer_endpoint(boost::asio::ip::address::from_string(writer_info->local_ip),
290 0 : writer_info->local_port);
291 :
292 0 : m_socket = std::make_unique<boost::asio::ip::udp::socket>(io_context, writer_endpoint);
293 :
294 0 : m_remote_endpoint = boost::asio::ip::udp::endpoint(boost::asio::ip::address::from_string(writer_info->remote_ip),
295 0 : writer_info->remote_port);
296 :
297 0 : TLOG() << "Created UDP socket on " << m_socket->local_endpoint().address() << ":"
298 0 : << m_socket->local_endpoint().port();
299 :
300 0 : } catch (const boost::system::system_error& e) {
301 0 : TLOG() << "Failed to create UDP socket: " << e.what() << " (Local IP: " << writer_info->local_ip
302 0 : << ", Local port: " << writer_info->local_port << ", Remote IP: " << writer_info->remote_ip
303 0 : << ", Remote port: " << writer_info->remote_port << ")";
304 0 : throw;
305 0 : }
306 0 : }
307 :
308 : void
309 0 : SocketWriterModule::UDPWriter::enqueue(GenericReceiverConcept::TypeErasedPayload&& payload)
310 : {
311 0 : boost::asio::post(*m_strand, [this, payload = std::move(payload)]() mutable {
312 0 : m_payloads.push(std::move(payload));
313 0 : if (m_payloads.size() == 1) { // queue was empty
314 0 : boost::asio::co_spawn(*m_strand, start(), boost::asio::detached);
315 : }
316 0 : });
317 0 : }
318 :
319 : boost::asio::awaitable<void>
320 0 : SocketWriterModule::UDPWriter::start()
321 : {
322 : while (!m_payloads.empty()) {
323 : auto payload = std::move(m_payloads.front());
324 :
325 : const auto bytes_sent = co_await m_socket->async_send_to(
326 : boost::asio::buffer(payload.data, payload.size), m_remote_endpoint, boost::asio::use_awaitable);
327 : m_payloads.pop();
328 :
329 : ++m_socket_stats->num_payloads;
330 : ++m_socket_stats->sum_payloads;
331 : m_socket_stats->sum_bytes.fetch_add(bytes_sent);
332 : ++m_socket_stats->stats_packet_count;
333 : }
334 0 : }
335 :
336 : void
337 0 : SocketWriterModule::UDPWriter::stop()
338 : {
339 0 : if (m_socket && m_socket->is_open()) {
340 0 : m_socket->close();
341 : }
342 0 : TLOG() << "Closed UDP socket";
343 0 : }
344 :
345 : std::shared_ptr<SocketWriterModule::SocketStats>
346 0 : SocketWriterModule::UDPWriter::get_socket_stats() const
347 : {
348 0 : return m_socket_stats;
349 : }
350 :
351 : } // namespace dunedaq::asiolibs
352 :
353 0 : DEFINE_DUNE_DAQ_MODULE(dunedaq::asiolibs::SocketWriterModule)
|
