Line data Source code
1 : /**
2 : * @file IfaceWrapper.hpp IfaceWrapper for holding resources of
3 : * a DPDK controlled NIC interface/port
4 : *
5 : * This is part of the DUNE DAQ , copyright 2020.
6 : * Licensing/copyright details are in the COPYING file that you should have
7 : * received with this code.
8 : */
9 : #ifndef DPDKLIBS_SRC_IFACEWRAPPER_HPP_
10 : #define DPDKLIBS_SRC_IFACEWRAPPER_HPP_
11 :
12 : //#include "dpdklibs/nicreader/Structs.hpp"
13 : #include "confmodel/NetworkDevice.hpp"
14 :
15 : #include "dpdklibs/EALSetup.hpp"
16 : #include "dpdklibs/udp/Utils.hpp"
17 : #include "dpdklibs/udp/PacketCtor.hpp"
18 : #include "dpdklibs/arp/ARP.hpp"
19 : #include "dpdklibs/ipv4_addr.hpp"
20 : #include "dpdklibs/XstatsHelper.hpp"
21 : #include "SourceConcept.hpp"
22 :
23 : #include <confmodel/Session.hpp>
24 : // #include <confmodel/NetworkDevice.hpp>
25 : #include "appmodel/DPDKReceiver.hpp"
26 : #include "appmodel/NWDetDataSender.hpp"
27 :
28 : #include <nlohmann/json.hpp>
29 :
30 : #include <ers/ers.hpp>
31 : #include "logging/Logging.hpp" // NOTE: if ISSUES ARE DECLARED BEFORE include logging/Logging.hpp, TLOG_DEBUG<<issue wont work.
32 :
33 : #include <memory>
34 : #include <sstream>
35 : #include <string>
36 : #include <set>
37 :
38 : #include <folly/ProducerConsumerQueue.h>
39 :
40 : namespace dunedaq {
41 :
42 2 : ERS_DECLARE_ISSUE( dpdklibs,
43 : MetricPublishFailed,
44 : "Field " << field << " was not reported",
45 : ((std::string)field)
46 : )
47 :
48 2 : ERS_DECLARE_ISSUE( dpdklibs,
49 : UnexpectedStreamID,
50 : "Unexpected stream ID " << src_id << " in UDP payoad. Total counter: " << counter,
51 : ((int)src_id)((size_t)counter)
52 : )
53 :
54 : namespace dpdklibs {
55 :
56 : class IfaceWrapper : public opmonlib::MonitorableObject
57 : {
58 : public:
59 : using sid_to_source_map_t = std::map<int, std::shared_ptr<SourceConcept>>;
60 :
61 : IfaceWrapper(uint iface_id, const appmodel::DPDKReceiver* receiver,
62 : const std::vector<const appmodel::NWDetDataSender*>& senders,
63 : const std::vector<const confmodel::DetectorStream*>& active_streams,
64 : sid_to_source_map_t& sources, std::atomic<bool>& run_marker);
65 : ~IfaceWrapper();
66 :
67 : IfaceWrapper(const IfaceWrapper&) = delete; ///< IfaceWrapper is not copy-constructible
68 : IfaceWrapper& operator=(const IfaceWrapper&) = delete; ///< IfaceWrapper is not copy-assginable
69 : IfaceWrapper(IfaceWrapper&&) = delete; ///< IfaceWrapper is not move-constructible
70 : IfaceWrapper& operator=(IfaceWrapper&&) = delete; ///< IfaceWrapper is not move-assignable
71 :
72 : //void init();
73 : void start();
74 : void stop();
75 :
76 : void generate_opmon_data() override;
77 :
78 : void allocate_mbufs();
79 : void setup_interface();
80 : void setup_flow_steering();
81 : void setup_xstats();
82 : void stop_xstats();
83 :
84 0 : void enable_flow() { m_lcore_enable_flow.store(true);}
85 0 : void disable_flow() { m_lcore_enable_flow.store(false);}
86 :
87 : const std::vector<uint16_t>& get_rte_cores() const { return m_rte_cores; }
88 :
89 : protected:
90 : //iface_conf_t m_cfg;
91 : int m_iface_id;
92 : std::string m_iface_id_str;
93 : bool m_configured;
94 :
95 : bool m_with_flow;
96 : bool m_prom_mode;
97 : std::vector<std::string> m_ip_addr;
98 : std::vector<rte_be32_t> m_ip_addr_bin;
99 : std::string m_mac_addr;
100 : int m_socket_id;
101 : int m_mtu;
102 : unsigned m_max_block_words;
103 : uint16_t m_rx_ring_size;
104 : uint16_t m_tx_ring_size;
105 : int m_num_mbufs;
106 : int m_burst_size;
107 : uint32_t m_lcore_sleep_ns;
108 : int m_mbuf_cache_size;
109 :
110 : private:
111 : int m_num_ip_sources;
112 : int m_num_rx_cores;
113 : std::set<std::string> m_ips;
114 : std::set<int> m_rx_qs;
115 : std::set<int> m_tx_qs;
116 : std::vector<uint16_t> m_rte_cores;
117 :
118 : // CPU core ID -> [queue -> ip]
119 : std::map<int, std::map<int, std::string>> m_rx_core_map;
120 : unsigned m_arp_rx_queue = 0; // RS TODO: make it configurable, and queue use conf check for exclusiveness!
121 :
122 : // Lcore stop signal
123 : std::atomic<bool> m_lcore_quit_signal{ false };
124 :
125 : std::atomic<bool> m_lcore_enable_flow{ false };
126 :
127 : // Mbufs and pools
128 : std::map<int, std::unique_ptr<rte_mempool>> m_mbuf_pools;
129 : std::map<int, struct rte_mbuf **> m_bufs; // by queue
130 :
131 : // Stats by queues
132 : std::map<int, std::atomic<std::size_t>> m_num_frames_rxq;
133 : std::map<int, std::atomic<std::size_t>> m_num_bytes_rxq;
134 : std::map<int, std::atomic<std::size_t>> m_num_full_bursts;
135 : std::map<int, std::atomic<uint16_t>> m_max_burst_size;
136 :
137 : // Stats by rte_workers
138 : std::map<int, std::atomic<std::size_t>> m_num_unhandled_non_ipv4;
139 : std::map<int, std::atomic<std::size_t>> m_num_unhandled_non_udp;
140 : std::map<int, std::atomic<std::size_t>> m_num_unhandled_non_jumbo_udp;
141 :
142 : // Unexpected stream ID count
143 : std::map<int, std::atomic<std::size_t>> m_num_unexid_frames;
144 :
145 :
146 : // DPDK HW stats
147 : dpdklibs::IfaceXstats m_iface_xstats;
148 :
149 : // stream -> source id map indexed by queue id
150 : // queue -> [stream_id -> sid]
151 : std::map<int, std::map<uint, uint>> m_stream_id_to_source_id;
152 : sid_to_source_map_t& m_sources;
153 : bool m_strict_parsing {true};
154 :
155 : // Run marker
156 : std::atomic<bool>& m_run_marker;
157 :
158 : // GARP
159 : std::unique_ptr<rte_mempool> m_garp_mbuf_pool;
160 : std::map<int, struct rte_mbuf **> m_garp_bufs;
161 : std::thread m_garp_thread;
162 : void garp_func();
163 : std::atomic<uint64_t> m_garps_sent{0};
164 :
165 : // ARP
166 : std::unique_ptr<rte_mempool> m_arp_mbuf_pool;
167 : std::map<int, struct rte_mbuf **> m_arp_bufs;
168 : std::thread m_arp_thread;
169 : void arp_func();
170 : std::atomic<uint64_t> m_arps_sent{0};
171 :
172 : // Lcore processor
173 : int rx_runner(void *arg __rte_unused);
174 : int arp_response_runner(void *arg __rte_unused);
175 :
176 : // Parse UDP payloads as DAQ frames
177 : void parse_udp_payload(int src_rx_q, char* payload, std::size_t size);
178 :
179 : // Pass through UDP payloads as is
180 : void passthrough_udp_payload(int src_rx_q, char* payload, std::size_t size);
181 :
182 : };
183 :
184 : } // namespace dpdklibs
185 : } // namespace dunedaq
186 :
187 : // #include "detail/IfaceWrapper.hxx"
188 :
189 : #endif // DPDKLIBS_SRC_IFACEWRAPPER_HPP_
|
