Line data Source code
1 : /**
2 : * @file SocketWriterModule.hpp 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 : #ifndef ASIOLIBS_PLUGINS_SOCKETWRITERMODULE_HPP_
9 : #define ASIOLIBS_PLUGINS_SOCKETWRITERMODULE_HPP_
10 :
11 : #include "GenericReceiverConcept.hpp"
12 :
13 : #include "appfwk/DAQModule.hpp"
14 : #include "utilities/ReusableThread.hpp"
15 :
16 : #include "appmodel/SocketDataWriterModule.hpp"
17 :
18 : #include <boost/asio.hpp>
19 :
20 : #include <map>
21 : #include <memory>
22 : #include <queue>
23 : #include <string>
24 : #include <vector>
25 :
26 : namespace dunedaq::asiolibs {
27 :
28 : class SocketWriterModule : public dunedaq::appfwk::DAQModule
29 : {
30 : public:
31 : /**
32 : * @brief Default receiver timeout in ms
33 : */
34 : static constexpr auto default_receiver_timeout_ms = 10;
35 :
36 : /**
37 : * @brief SocketWriterModule constructor
38 : * @param name DAQ module instance name
39 : */
40 : explicit SocketWriterModule(const std::string& name);
41 0 : ~SocketWriterModule() = default;
42 :
43 : SocketWriterModule(const SocketWriterModule&) = delete; ///< SocketWriterModule is not copy-constructible
44 : SocketWriterModule& operator=(const SocketWriterModule&) = delete; ///< SocketWriterModule is not copy-assignable
45 : SocketWriterModule(SocketWriterModule&&) = delete; ///< SocketWriterModule is not move-constructible
46 : SocketWriterModule& operator=(SocketWriterModule&&) = delete; ///< SocketWriterModule is not move-assignable
47 :
48 : /**
49 : * @brief Handles initialization on boot
50 : * @param mcfg DAQ configuration data
51 : */
52 : void init(const std::shared_ptr<appfwk::ConfigurationManager> mcfg) override;
53 :
54 : private:
55 : enum class SocketType
56 : {
57 : TCP,
58 : UDP,
59 : INVALID
60 : };
61 :
62 : struct SocketStats
63 : {
64 : /**
65 : * @brief Total number of received payloads
66 : */
67 : std::atomic<uint64_t> sum_payloads{ 0 }; // NOLINT(build/unsigned)
68 :
69 : /**
70 : * @brief Incremental number of received payloads
71 : */
72 : std::atomic<uint64_t> num_payloads{ 0 }; // NOLINT(build/unsigned)
73 :
74 : /**
75 : * @brief Total number of received bytes
76 : */
77 : std::atomic<uint64_t> sum_bytes{ 0 }; // NOLINT(build/unsigned)
78 :
79 : /**
80 : * @brief Timeout on data inputs
81 : */
82 : std::atomic<uint64_t> queue_timeout_count{ 0 }; // NOLINT(build/unsigned)
83 :
84 : /**
85 : * @brief Rate of consumed packets
86 : */
87 : std::atomic<double> rate_payloads_consumed{ 0 };
88 :
89 : /**
90 : * @brief Counts packets since last opmon data generation
91 : */
92 : std::atomic<int> stats_packet_count{ 0 };
93 : };
94 :
95 : struct WriterInfo
96 : {
97 : /**
98 : * @brief Source IP address
99 : */
100 : std::string local_ip;
101 :
102 : /**
103 : * @brief Source port number
104 : */
105 : uint32_t local_port; // NOLINT(build/unsigned)
106 :
107 : /**
108 : * @brief Destination IP address
109 : */
110 : std::string remote_ip;
111 :
112 : /**
113 : * @brief Destination port number
114 : */
115 : uint32_t remote_port; // NOLINT(build/unsigned)
116 :
117 : /**
118 : * @brief Statistics of socket traffic
119 : */
120 : std::shared_ptr<SocketStats> socket_stats;
121 : };
122 :
123 : class TCPWriter
124 : {
125 : public:
126 : /**
127 : * @brief Creates and connects a TCP socket
128 : * @param io_context I/O context for socket creation and coroutine spawn
129 : * @param writer_info TCP writer info
130 : * @throws boost::system::system_error on failure
131 : */
132 : void configure(boost::asio::io_context& io_context, std::shared_ptr<WriterInfo> writer_info);
133 :
134 : /**
135 : * @brief Enqueues payload
136 : * @param payload Payload to send
137 : */
138 : void enqueue(GenericReceiverConcept::TypeErasedPayload&& payload);
139 :
140 : /**
141 : * @brief Closes the socket
142 : */
143 : void stop();
144 :
145 : /**
146 : * @brief Get socket statistics
147 : * @return Statistics of socket traffic
148 : */
149 : std::shared_ptr<SocketStats> get_socket_stats() const;
150 :
151 : private:
152 : /**
153 : * @brief Asynchronously sends payloads to the socket in a loop
154 : * @return Coroutine handle
155 : */
156 : boost::asio::awaitable<void> start();
157 :
158 : /**
159 : * @brief TCP socket
160 : */
161 : std::unique_ptr<boost::asio::ip::tcp::socket> m_socket;
162 :
163 : /**
164 : * @brief Statistics of socket traffic
165 : */
166 : std::shared_ptr<SocketStats> m_socket_stats;
167 :
168 : /**
169 : * @brief Payloads waiting to be sent (ensures no simultaneous async_write calls)
170 : */
171 : std::queue<GenericReceiverConcept::TypeErasedPayload> m_payloads;
172 :
173 : /**
174 : * @brief I/O context for socket operations
175 : */
176 : boost::asio::io_context* m_io_context;
177 :
178 : /**
179 : * @brief Ensures no race on queue
180 : */
181 : std::shared_ptr<boost::asio::strand<boost::asio::io_context::executor_type>> m_strand;
182 : };
183 :
184 : class UDPWriter
185 : {
186 : public:
187 : /**
188 : * @brief Creates a UDP socket
189 : * @param io_context I/O context for socket creation and coroutine spawn
190 : * @param writer_info UDP writer info
191 : */
192 : void configure(boost::asio::io_context& io_context, std::shared_ptr<WriterInfo> writer_info);
193 :
194 : /**
195 : * @brief Enqueues payload
196 : * @param payload Payload to send
197 : */
198 : void enqueue(GenericReceiverConcept::TypeErasedPayload&& payload);
199 :
200 : /**
201 : * @brief Closes the socket
202 : */
203 : void stop();
204 :
205 : /**
206 : * @brief Get socket statistics
207 : * @return Statistics of socket traffic
208 : */
209 : std::shared_ptr<SocketStats> get_socket_stats() const;
210 :
211 : private:
212 : /**
213 : * @brief Asynchronously sends payloads to the socket in a loop
214 : * @return Coroutine handle
215 : */
216 : boost::asio::awaitable<void> start();
217 :
218 : /**
219 : * @brief UDP socket
220 : */
221 : std::unique_ptr<boost::asio::ip::udp::socket> m_socket;
222 :
223 : /**
224 : * @brief Statistics of socket traffic
225 : */
226 : std::shared_ptr<SocketStats> m_socket_stats;
227 :
228 : /**
229 : * @brief Payloads waiting to be sent (ensures no simultaneous async_write calls)
230 : */
231 : std::queue<GenericReceiverConcept::TypeErasedPayload> m_payloads;
232 :
233 : /**
234 : * @brief I/O context for socket operations
235 : */
236 : boost::asio::io_context* m_io_context;
237 :
238 : /**
239 : * @brief Remote endpoint
240 : */
241 : boost::asio::ip::udp::endpoint m_remote_endpoint;
242 :
243 : /**
244 : * @brief Ensures no race on queue
245 : */
246 : std::shared_ptr<boost::asio::strand<boost::asio::io_context::executor_type>> m_strand;
247 : };
248 :
249 : // Commands
250 : void do_configure(const CommandData_t&);
251 : void do_start(const CommandData_t&);
252 : void do_stop(const CommandData_t&);
253 :
254 : void generate_opmon_data() override;
255 :
256 : /**
257 : * @brief Converts a socket type string to an enum
258 : * @param socket_type Socket type as a string
259 : * @return Corresponding SocketType enum
260 : */
261 : SocketType string_to_socket_type(const std::string& socket_type) const;
262 :
263 : /**
264 : * @brief Gets dal inputs
265 : * @param mdal SocketDataWriterModule dal
266 : */
267 : void get_dal_inputs(const dunedaq::appmodel::SocketDataWriterModule* mdal);
268 :
269 : /**
270 : * @brief Data consume thread function
271 : */
272 : void run_consume();
273 :
274 : /**
275 : * @brief I/O context for socket operations
276 : */
277 : boost::asio::io_context m_io_context;
278 :
279 : /**
280 : * @brief Prevents I/O context from exiting prematurely
281 : */
282 : boost::asio::executor_work_guard<boost::asio::io_context::executor_type> m_work_guard;
283 :
284 : /**
285 : * @brief Socket writer
286 : */
287 : std::shared_ptr<std::variant<TCPWriter, UDPWriter>> m_writer;
288 :
289 : /**
290 : * @brief Background thread to keep the I/O context running
291 : */
292 : std::jthread m_io_thread;
293 :
294 : /**
295 : * @brief Socket writer info
296 : */
297 : std::shared_ptr<WriterInfo> m_writer_info;
298 :
299 : // RECEIVER
300 : /**
301 : * @brief Generic receiver
302 : */
303 : std::shared_ptr<GenericReceiverConcept> m_receiver;
304 :
305 : /**
306 : * @brief Receiver timeout in ms
307 : */
308 : std::chrono::milliseconds m_receiver_timeout_ms{ default_receiver_timeout_ms };
309 :
310 : // CONSUMER
311 : /**
312 : * @brief Data consume thread
313 : */
314 : utilities::ReusableThread m_consumer_thread;
315 :
316 : /**
317 : * @brief Whether consumer thread should continue
318 : */
319 : std::atomic<bool> m_run_marker{ false };
320 :
321 : // RUN START T0
322 : /**
323 : * @brief Timestamp used to measure time between opmon reports
324 : */
325 : std::chrono::time_point<std::chrono::steady_clock> m_t0;
326 : };
327 :
328 : } // namespace dunedaq::asiolibs
329 :
330 : #endif // ASIOLIBS_PLUGINS_SOCKETWRITERMODULE_HPP_
|
