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