Line data Source code
1 : /**
2 : * @file SocketReaderModule.hpp 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 : #ifndef ASIOLIBS_PLUGINS_SOCKETREADERMODULE_HPP_
9 : #define ASIOLIBS_PLUGINS_SOCKETREADERMODULE_HPP_
10 :
11 : #include "appfwk/DAQModule.hpp"
12 :
13 : #include <boost/array.hpp>
14 : #include <boost/asio.hpp>
15 :
16 : #include <map>
17 : #include <memory>
18 : #include <string>
19 : #include <vector>
20 : #include <utility>
21 :
22 : namespace dunedaq::asiolibs {
23 :
24 : class SourceConcept;
25 :
26 : class SocketReaderModule : public dunedaq::appfwk::DAQModule
27 : {
28 : public:
29 : using remote_t = std::pair<std::string, uint32_t>; // NOLINT(build/unsigned)
30 : using remote_stream_pair_t = std::pair<remote_t, uint32_t>; // NOLINT(build/unsigned)
31 : using remote_source_map_t = std::map<remote_stream_pair_t, std::shared_ptr<SourceConcept>>;
32 :
33 : /**
34 : * @brief SocketReaderModule constructor
35 : * @param name DAQ module instance name
36 : */
37 : explicit SocketReaderModule(const std::string& name);
38 0 : ~SocketReaderModule() = default;
39 :
40 : SocketReaderModule(const SocketReaderModule&) = delete; ///< SocketReaderModule is not copy-constructible
41 : SocketReaderModule& operator=(const SocketReaderModule&) = delete; ///< SocketReaderModule is not copy-assignable
42 : SocketReaderModule(SocketReaderModule&&) = delete; ///< SocketReaderModule is not move-constructible
43 : SocketReaderModule& operator=(SocketReaderModule&&) = delete; ///< SocketReaderModule is not move-assignable
44 :
45 : /**
46 : * @brief Handles initialization on boot
47 : * @param mcfg DAQ configuration data
48 : */
49 : void init(const std::shared_ptr<appfwk::ConfigurationManager> mcfg) override;
50 :
51 : private:
52 : enum class SocketType
53 : {
54 : TCP,
55 : UDP,
56 : INVALID
57 : };
58 :
59 : struct SocketStats
60 : {
61 : /**
62 : * @brief Received packets
63 : */
64 : std::atomic<uint64_t> packets_received{ 0 }; // NOLINT(build/unsigned)
65 :
66 : /**
67 : * @brief Received bytes
68 : */
69 : std::atomic<uint64_t> bytes_received{ 0 }; // NOLINT(build/unsigned)
70 :
71 : /**
72 : * @brief Counts packets since last opmon data generation
73 : */
74 : std::atomic<int> stats_packet_count{ 0 };
75 : };
76 :
77 : struct ReaderInfo
78 : {
79 : /**
80 : * @brief Source IP address
81 : */
82 : std::string local_ip;
83 :
84 : /**
85 : * @brief Source port number
86 : */
87 : uint32_t local_port; // NOLINT(build/unsigned)
88 :
89 : /**
90 : * @brief Statistics of socket traffic
91 : */
92 : std::shared_ptr<SocketStats> socket_stats;
93 : };
94 :
95 : class TCPReader
96 : {
97 : public:
98 : /**
99 : * @brief Asynchronously creates and connects a TCP socket
100 : * @param io_context I/O context for socket creation
101 : * @param reader_info TCP reader info
102 : * @throws boost::system::system_error on failure
103 : */
104 : void configure(boost::asio::io_context& io_context, std::shared_ptr<ReaderInfo> reader_info);
105 :
106 : /**
107 : * @brief Asynchronously receives payloads from the socket in a loop
108 : * @param sources Data sources
109 : * @return Coroutine handle
110 : */
111 : boost::asio::awaitable<void> start(const remote_source_map_t& remote_to_source);
112 :
113 : /**
114 : * @brief Closes the socket
115 : */
116 : void stop();
117 :
118 : private:
119 : /**
120 : * @brief TCP socket
121 : */
122 : std::unique_ptr<boost::asio::ip::tcp::socket> m_socket;
123 :
124 : /**
125 : * @brief Connected remote
126 : */
127 : remote_t m_remote;
128 :
129 : /**
130 : * @brief Statistics of socket traffic
131 : */
132 : std::shared_ptr<SocketStats> m_socket_stats;
133 : };
134 :
135 : class UDPReader
136 : {
137 : public:
138 : /**
139 : * @brief Creates a UDP socket
140 : * @param io_context I/O context for socket creation
141 : * @param reader_info UDP reader info
142 : */
143 : void configure(boost::asio::io_context& io_context, std::shared_ptr<ReaderInfo> reader_info);
144 :
145 : /**
146 : * @brief Asynchronously receives payloads from the socket in a loop
147 : * @param sources Data sources
148 : * @return Coroutine handle
149 : */
150 : boost::asio::awaitable<void> start(const remote_source_map_t& remote_to_source);
151 :
152 : /**
153 : * @brief Closes the socket
154 : */
155 : void stop();
156 :
157 : private:
158 : /**
159 : * @brief UDP socket
160 : */
161 : std::unique_ptr<boost::asio::ip::udp::socket> m_socket;
162 :
163 : /**
164 : * @brief Statistics of socket traffic
165 : */
166 : std::shared_ptr<SocketStats> m_socket_stats;
167 : };
168 :
169 : // Commands
170 : void do_configure(const CommandData_t&);
171 : void do_start(const CommandData_t&);
172 : void do_stop(const CommandData_t&);
173 :
174 : void generate_opmon_data() override;
175 :
176 : /**
177 : * @brief Converts a socket type string to an enum
178 : * @param socket_type Socket type as a string
179 : * @return Corresponding SocketType enum
180 : */
181 : SocketType string_to_socket_type(const std::string& socket_type) const;
182 :
183 : /**
184 : * @brief I/O context for socket operations
185 : */
186 : boost::asio::io_context m_io_context;
187 :
188 : /**
189 : * @brief Prevents I/O context from exiting prematurely
190 : */
191 : boost::asio::executor_work_guard<boost::asio::io_context::executor_type> m_work_guard;
192 :
193 : /**
194 : * @brief Socket readers
195 : */
196 : std::vector<std::shared_ptr<std::variant<TCPReader, UDPReader>>> m_readers;
197 :
198 : /**
199 : * @brief Background thread to keep the I/O context running
200 : */
201 : std::jthread m_io_thread;
202 :
203 : /**
204 : * @brief Socket reader infos
205 : */
206 : std::vector<std::shared_ptr<ReaderInfo>> m_reader_infos;
207 :
208 : // Sinks (SourceConcepts)
209 : using sid_to_source_map_t = std::map<uint32_t, std::shared_ptr<SourceConcept>>; // NOLINT(build/unsigned)
210 : /**
211 : * @brief Data sources
212 : */
213 : sid_to_source_map_t m_sources;
214 :
215 : /**
216 : * @brief Map between a pair of {remote, stream} and the corresponding source
217 : */
218 : remote_source_map_t m_remote_to_source;
219 :
220 : };
221 :
222 : } // namespace dunedaq::asiolibs
223 :
224 : #endif // ASIOLIBS_PLUGINS_SOCKETREADERMODULE_HPP_
|
