1+ /* *
2+ * \file dpdk-ring.cpp
3+ * \brief DPDK ring input interface for ipfixprobe (secondary DPDK app).
4+ * \author Jaroslav Pesek <[email protected] > 5+ * \date 2023
6+ */
7+ /*
8+ * Copyright (C) 2023 CESNET
9+ *
10+ * LICENSE TERMS
11+ *
12+ * Redistribution and use in source and binary forms, with or without
13+ * modification, are permitted provided that the following conditions
14+ * are met:
15+ * 1. Redistributions of source code must retain the above copyright
16+ * notice, this list of conditions and the following disclaimer.
17+ * 2. Redistributions in binary form must reproduce the above copyright
18+ * notice, this list of conditions and the following disclaimer in
19+ * the documentation and/or other materials provided with the
20+ * distribution.
21+ * 3. Neither the name of the Company nor the names of its contributors
22+ * may be used to endorse or promote products derived from this
23+ * software without specific prior written permission.
24+ *
25+ */
26+ #include < cstring>
27+ #include < mutex>
28+ #include < rte_ethdev.h>
29+ #include < rte_version.h>
30+ #include < unistd.h>
31+ #include < rte_eal.h>
32+ #include < rte_errno.h>
33+
34+ #include " dpdk-ring.h"
35+ #include " parser.hpp"
36+
37+ namespace ipxp {
38+ __attribute__ ((constructor)) static void register_this_plugin ()
39+ {
40+ static PluginRecord rec = PluginRecord (" dpdk-ring" return new DpdkRingReader (); });
41+ register_plugin (&rec);
42+ }
43+
44+ DpdkRingCore *DpdkRingCore::m_instance = nullptr ;
45+
46+ DpdkRingCore &DpdkRingCore::getInstance ()
47+ {
48+ if (!m_instance) {
49+ m_instance = new DpdkRingCore ();
50+ }
51+ return *m_instance;
52+ }
53+
54+ DpdkRingCore::~DpdkRingCore ()
55+ {
56+ rte_eal_cleanup ();
57+ m_instance = nullptr ;
58+ }
59+
60+ void DpdkRingCore::deinit ()
61+ {
62+ if (m_instance) {
63+ delete m_instance;
64+ m_instance = nullptr ;
65+ }
66+ }
67+
68+ void DpdkRingCore::configure (const char * params) {
69+ if (isConfigured) {
70+ return ;
71+ }
72+
73+ try {
74+ parser.parse (params);
75+ } catch (ParserError& e) {
76+ throw PluginError (e.what ());
77+ }
78+
79+ configureEal (parser.eal_params ());
80+ isConfigured = true ;
81+ }
82+
83+ std::vector<char *> DpdkRingCore::convertStringToArgvFormat (const std::string& ealParams)
84+ {
85+ // set first value as program name (argv[0])
86+ std::vector<char *> args = {" ipfixprobe"
87+ std::istringstream iss (ealParams);
88+ std::string token;
89+
90+ while (iss >> token) {
91+ char *arg = new char [token.size () + 1 ];
92+ copy (token.begin (), token.end (), arg);
93+ arg[token.size ()] = ' \0 '
94+ args.push_back (arg);
95+ }
96+ return args;
97+ }
98+
99+ void DpdkRingCore::configureEal (const std::string& ealParams)
100+ {
101+ std::vector<char *> args = convertStringToArgvFormat (ealParams);
102+
103+ if (rte_eal_init (args.size (), args.data ()) < 0 ) {
104+ rte_exit (EXIT_FAILURE, " Cannot initialize RTE_EAL: %s\n " rte_strerror (rte_errno));
105+ }
106+ }
107+
108+ DpdkRingReader::DpdkRingReader ()
109+ : m_dpdkRingCore(DpdkRingCore::getInstance())
110+ {
111+ pkts_read_ = 0 ;
112+ }
113+
114+ DpdkRingReader::~DpdkRingReader ()
115+ {
116+ m_dpdkRingCore.deinit ();
117+ }
118+
119+ void DpdkRingReader::createRteMbufs (uint16_t mbufsSize)
120+ {
121+ try {
122+ mbufs_.resize (mbufsSize);
123+ } catch (const std::exception& e) {
124+ throw PluginError (e.what ());
125+ }
126+ }
127+
128+ void DpdkRingReader::init (const char * params)
129+ {
130+ m_dpdkRingCore.configure (params);
131+ DpdkRingOptParser parser;
132+ try {
133+ parser.parse (params);
134+ } catch (ParserError& e) {
135+ throw PluginError (e.what ());
136+ }
137+ createRteMbufs (m_dpdkRingCore.parser .pkt_buffer_size ());
138+ m_ring = rte_ring_lookup (parser.ring_name ().c_str ());
139+ if (!m_ring) {
140+ throw PluginError (" Cannot find ring with name: " ring_name ());
141+ } else {
142+ is_reader_ready = true ;
143+ }
144+ }
145+
146+ struct timeval DpdkRingReader::getTimestamp (rte_mbuf* mbuf)
147+ {
148+ struct timeval tv;
149+ auto now = std::chrono::system_clock::now ();
150+ auto now_t = std::chrono::system_clock::to_time_t (now);
151+
152+ auto dur = now - std::chrono::system_clock::from_time_t (now_t );
153+ auto micros = std::chrono::duration_cast<std::chrono::microseconds>(dur).count ();
154+
155+ tv.tv_sec = now_t ;
156+ tv.tv_usec = micros;
157+ return tv;
158+ }
159+
160+ InputPlugin::Result DpdkRingReader::get (PacketBlock& packets)
161+ {
162+ while (is_reader_ready == false ) {
163+ usleep (1000 );
164+ }
165+
166+ parser_opt_t opt {&packets, false , false , 0 };
167+
168+ packets.cnt = 0 ;
169+ for (auto i = 0 ; i < pkts_read_; i++) {
170+ rte_pktmbuf_free (mbufs_[i]);
171+ }
172+ pkts_read_ = rte_ring_dequeue_burst (
173+ m_ring,
174+ reinterpret_cast <void **>(mbufs_.data ()),
175+ mbufs_.capacity (),
176+ nullptr );
177+ if (pkts_read_ == 0 ) {
178+ return Result::TIMEOUT;
179+ }
180+ for (auto i = 0 ; i < pkts_read_; i++) {
181+ parse_packet (&opt,
182+ getTimestamp (mbufs_[i]),
183+ rte_pktmbuf_mtod (mbufs_[i], const std::uint8_t *),
184+ rte_pktmbuf_data_len (mbufs_[i]),
185+ rte_pktmbuf_data_len (mbufs_[i]));
186+ m_seen++;
187+ m_parsed++;
188+ }
189+ return Result::PARSED;
190+ }
191+ } // namespace ipxp
0 commit comments