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