Added flow parser for 82599 10G card. Added exceptions for flow stat

errors

8 files changed, 206 insertions, 112 deletions

diff --git a/src/flow_stat.cpp b/src/flow_stat.cpp
index 778c92b9..43bde08b 100644
--- a/src/flow_stat.cpp
+++ b/src/flow_stat.cpp
@@ -34,7 +34,6 @@
 #define FLOW_STAT_ADD_ALL_PORTS 255
 
 static const uint16_t FREE_HW_ID = UINT16_MAX;
-static bool no_stat_supported = true;
 
 inline std::string methodName(const std::string& prettyFunction)
 {
@@ -107,7 +106,7 @@ int CFlowStatUserIdInfo::add_stream(uint8_t proto) {
 #endif
 
     if (proto != m_proto)
-        return -1;
+        throw TrexException("Can't use same pg_id for streams with different l4 protocol");
 
     m_ref_count++;
 
@@ -198,7 +197,7 @@ int CFlowStatUserIdMap::add_stream(uint32_t user_id, uint8_t proto) {
     if (! c_user_id) {
         c_user_id = add_user_id(user_id, proto);
         if (! c_user_id)
-            return -1;
+            throw TrexException("Failed adding statistic counter - Failure in add_stream");
         return 0;
     } else {
         return c_user_id->add_stream(proto);
@@ -214,7 +213,7 @@ int CFlowStatUserIdMap::del_stream(uint32_t user_id) {
 
     c_user_id = find_user_id(user_id);
     if (! c_user_id) {
-        return -1;
+        throw TrexException("Trying to delete stream which does not exist");
     }
 
     if (c_user_id->del_stream() == 0) {
@@ -237,13 +236,13 @@ int CFlowStatUserIdMap::start_stream(uint32_t user_id, uint16_t hw_id) {
     if (! c_user_id) {
         fprintf(stderr, "%s Error: Trying to associate hw id %d to user_id %d but it does not exist\n"
                 , __func__, hw_id, user_id);
-        return -1;
+        throw TrexException("Internal error: Trying to associate non exist group id");
     }
 
     if (c_user_id->is_hw_id()) {
-        fprintf(stderr, "%s Error: Trying to associate hw id %d to user_id %d but it is already associate to %u\n"
+        fprintf(stderr, "%s Error: Trying to associate hw id %d to user_id %d but it is already associated to %u\n"
                 , __func__, hw_id, user_id, c_user_id->get_hw_id());
-        return -1;
+        throw TrexException("Internal error: Trying to associate used packet group id to different hardware counter");
     }
     c_user_id->set_hw_id(hw_id);
     c_user_id->add_started_stream();
@@ -260,9 +259,9 @@ int CFlowStatUserIdMap::start_stream(uint32_t user_id) {
 
     c_user_id = find_user_id(user_id);
     if (! c_user_id) {
-        fprintf(stderr, "%s Error: Trying to start stream on user_id %d but it does not exist\n"
+        fprintf(stderr, "%s Error: Trying to start stream on pg_id %d but it does not exist\n"
                 , __func__, user_id);
-        return -1;
+        throw TrexException("Trying to start stream with non exist packet group id");
     }
 
     c_user_id->add_started_stream();
@@ -281,9 +280,9 @@ int CFlowStatUserIdMap::stop_stream(uint32_t user_id) {
 
     c_user_id = find_user_id(user_id);
     if (! c_user_id) {
-        fprintf(stderr, "%s Error: Trying to stop stream on user_id %d but it does not exist\n"
+        fprintf(stderr, "%s Error: Trying to stop stream on pg_id %d but it does not exist\n"
                 , __func__, user_id);
-        return -1;
+        throw TrexException("Trying to stop stream with non exist packet group id");
     }
 
     return c_user_id->stop_started_stream();
@@ -388,6 +387,32 @@ CFlowStatRuleMgr::CFlowStatRuleMgr() {
     m_max_hw_id = -1;
     m_num_started_streams = 0;
     m_ring_to_rx = NULL;
+    m_capabilities = 0;
+    m_parser = NULL;
+}
+
+CFlowStatRuleMgr::~CFlowStatRuleMgr() {
+    if (m_parser)
+        delete m_parser;
+}
+
+void CFlowStatRuleMgr::create() {
+    uint16_t num_counters, capabilities;
+    TrexStateless *tstateless = get_stateless_obj();
+    assert(tstateless);
+
+    m_api = tstateless->get_platform_api();
+    assert(m_api);
+    m_api->get_interface_stat_info(0, num_counters, capabilities);
+    m_api->get_port_num(m_num_ports);
+    for (uint8_t port = 0; port < m_num_ports; port++) {
+        assert(m_api->reset_hw_flow_stats(port) == 0);
+    }
+    m_ring_to_rx = CMsgIns::Ins()->getCpRx()->getRingCpToDp(0);
+    assert(m_ring_to_rx);
+    m_parser = m_api->get_flow_stat_parser();
+    assert(m_parser);
+    m_capabilities = capabilities;
 }
 
 std::ostream& operator<<(std::ostream& os, const CFlowStatRuleMgr& cf) {
@@ -397,38 +422,30 @@ std::ostream& operator<<(std::ostream& os, const CFlowStatRuleMgr& cf) {
     return os;
 }
 
-int CFlowStatRuleMgr::compile_stream(const TrexStream * stream, Cxl710Parser &parser) {
+int CFlowStatRuleMgr::compile_stream(const TrexStream * stream, CFlowStatParser *parser) {
 #ifdef __DEBUG_FUNC_ENTRY__
     std::cout << __METHOD_NAME__ << " user id:" << stream->m_rx_check.m_pg_id << " en:";
     std::cout << stream->m_rx_check.m_enabled << std::endl;
 #endif
 
-    // currently we support only IP ID rule types
-    // all our ports are the same type, so testing port 0 is enough
-    uint16_t num_counters, capabilities;
-    m_api->get_interface_stat_info(0, num_counters, capabilities);
-    if ((capabilities & TrexPlatformApi::IF_STAT_IPV4_ID) == 0) {
-        return -2;
-    }
-
-    if (parser.parse(stream->m_pkt.binary, stream->m_pkt.len) != 0) {
+    if (parser->parse(stream->m_pkt.binary, stream->m_pkt.len) != 0) {
         // if we could not parse the packet, but no stat count needed, it is probably OK.
         if (stream->m_rx_check.m_enabled) {
             fprintf(stderr, "Error: %s - Compilation failed\n", __func__);
-            return -1;
+            throw TrexException("Failed parsing given packet for flow stat. Probably bad packet format.");
         } else {
             return 0;
         }
     }
 
-    if (!parser.is_fdir_supported()) {
+    if (!parser->is_stat_supported()) {
         if (stream->m_stream_id <= 0) {
-            // rx stat not needed. Do nothing.
+            // flow stat not needed. Do nothing.
             return 0;
         } else {
-            // rx stat needed, but packet format is not supported
-            fprintf(stderr, "Error: %s - Unsupported packet format for rx stat\n", __func__);
-            return -1;
+            // flow stat needed, but packet format is not supported
+            fprintf(stderr, "Error: %s - Unsupported packet format for flow stat\n", __func__);
+            throw TrexException("Unsupported packet format for flow stat on given interface type");
         }
     }
     return 0;
@@ -439,47 +456,36 @@ int CFlowStatRuleMgr::add_stream(const TrexStream * stream) {
     std::cout << __METHOD_NAME__ << " user id:" << stream->m_rx_check.m_pg_id << std::endl;
 #endif
 
+    if (! stream->m_rx_check.m_enabled) {
+        return 0;
+    }
+
     // Init everything here, and not in the constructor, since we relay on other objects
     // By the time a stream is added everything else is initialized.
     if (! m_api ) {
-        TrexStateless *tstateless = get_stateless_obj();
-        m_api = tstateless->get_platform_api();
-        uint16_t num_counters, capabilities;
-        m_api->get_interface_stat_info(0, num_counters, capabilities);
-        if ((capabilities & TrexPlatformApi::IF_STAT_IPV4_ID) == 0) {
-            // All our interfaces are from the same type. If statistics not supported.
-            // no operation will work
-            return -1;
-        } else {
-            no_stat_supported = false;
-        }
-        m_api->get_port_num(m_num_ports);
-        for (uint8_t port = 0; port < m_num_ports; port++) {
-            assert(m_api->reset_hw_flow_stats(port) == 0);
-        }
-        m_ring_to_rx = CMsgIns::Ins()->getCpRx()->getRingCpToDp(0);
+        create();
     }
 
-    if (no_stat_supported)
-        return -ENOTSUP;
+    uint16_t rule_type = TrexPlatformApi::IF_STAT_IPV4_ID; // In the future need to get it from the stream;
 
-    Cxl710Parser parser;
-    int ret;
-
-    if (! stream->m_rx_check.m_enabled) {
-        return 0;
+    if ((m_capabilities & rule_type) == 0) {
+        fprintf(stderr, "Error: %s - rule type not supported by interface\n", __func__);
+        throw TrexException("Interface does not support given rule type");
     }
 
-    if ((ret = compile_stream(stream, parser)) < 0)
-        return ret;
+    // compile_stream throws exception if something goes wrong
+    compile_stream(stream, m_parser);
 
     uint8_t l4_proto;
-    if (parser.get_l4_proto(l4_proto) < 0) {
-        printf("Error: %s failed finding l4 proto\n", __func__);
-        return -1;
+    if (m_parser->get_l4_proto(l4_proto) < 0) {
+        fprintf(stderr, "Error: %s failed finding l4 proto\n", __func__);
+        throw TrexException("Failed determining l4 proto for packet");
     }
 
-    return m_user_id_map.add_stream(stream->m_rx_check.m_pg_id, l4_proto);
+    // throws exception if there is error
+    m_user_id_map.add_stream(stream->m_rx_check.m_pg_id, l4_proto);
+
+    return 0;
 }
 
 int CFlowStatRuleMgr::del_stream(const TrexStream * stream) {
@@ -487,8 +493,8 @@ int CFlowStatRuleMgr::del_stream(const TrexStream * stream) {
     std::cout << __METHOD_NAME__ << " user id:" << stream->m_rx_check.m_pg_id << std::endl;
 #endif
 
-    if (no_stat_supported)
-        return -ENOTSUP;
+    if (! m_api)
+        throw TrexException("Called del_stream, but no stream was added");
 
     if (! stream->m_rx_check.m_enabled) {
         return 0;
@@ -497,10 +503,13 @@ int CFlowStatRuleMgr::del_stream(const TrexStream * stream) {
     if (m_user_id_map.is_started(stream->m_rx_check.m_pg_id)) {
         std::cerr << "Error: Trying to delete flow statistics stream " << stream->m_rx_check.m_pg_id
                   << " which is not stopped." << std::endl;
-        return -1;
+        throw TrexException("Trying to delete stream which was not stopped");
     }
 
-    return m_user_id_map.del_stream(stream->m_rx_check.m_pg_id);
+    // Throws exception in case of error
+    m_user_id_map.del_stream(stream->m_rx_check.m_pg_id);
+
+    return 0;
 }
 
 // called on all streams, when stream start to transmit
@@ -514,33 +523,49 @@ int CFlowStatRuleMgr::start_stream(TrexStream * stream, uint16_t &ret_hw_id) {
     std::cout << __METHOD_NAME__ << " user id:" << stream->m_rx_check.m_pg_id << std::endl;
 #endif
 
-    Cxl710Parser parser;
     int ret;
-
-    if (no_stat_supported)
-        return -ENOTSUP;
-
-    if ((ret = compile_stream(stream, parser)) < 0)
-        return ret;
+    // Streams which does not need statistics might be started, before any stream that do
+    // need statistcs, so start_stream might be called before add_stream
+    if (! m_api ) {
+        create();
+    }
 
     // first handle streams that do not need rx stat
     if (! stream->m_rx_check.m_enabled) {
-        // no need for stat count
+        try {
+            compile_stream(stream, m_parser);
+        } catch (TrexException) {
+            // If no statistics needed, and we can't parse the stream, that's OK.
+            return 0;
+        }
+
         uint16_t ip_id;
-        if (parser.get_ip_id(ip_id) < 0) {
-            return 0; // if we could not find and ip id, no need to fix
+        if (m_parser->get_ip_id(ip_id) < 0) {
+            return 0; // if we could not find the ip id, no need to fix
         }
         // verify no reserved IP_ID used, and change if needed
         if (ip_id >= IP_ID_RESERVE_BASE) {
-            if (parser.set_ip_id(ip_id & 0xefff) < 0) {
-                return -1;
+            if (m_parser->set_ip_id(ip_id & 0xefff) < 0) {
+                throw TrexException("Stream IP ID in reserved range. Failed changing it");
             }
         }
         return 0;
     }
 
-    uint16_t hw_id;
     // from here, we know the stream need rx stat
+
+    // compile_stream throws exception if something goes wrong
+    if ((ret = compile_stream(stream, m_parser)) < 0)
+        return ret;
+
+    uint16_t hw_id;
+    uint16_t rule_type = TrexPlatformApi::IF_STAT_IPV4_ID; // In the future, need to get it from the stream;
+
+    if ((m_capabilities & rule_type) == 0) {
+        fprintf(stderr, "Error: %s - rule type not supported by interface\n", __func__);
+        throw TrexException("Interface does not support given rule type");
+    }
+
     if (m_user_id_map.is_started(stream->m_rx_check.m_pg_id)) {
         m_user_id_map.start_stream(stream->m_rx_check.m_pg_id); // just increase ref count;
         hw_id = m_user_id_map.get_hw_id(stream->m_rx_check.m_pg_id); // can't fail if we got here
@@ -548,19 +573,19 @@ int CFlowStatRuleMgr::start_stream(TrexStream * stream, uint16_t &ret_hw_id) {
         hw_id = m_hw_id_map.find_free_hw_id();
         if (hw_id == FREE_HW_ID) {
             printf("Error: %s failed finding free hw_id\n", __func__);
-            return -1;
+            throw TrexException("Failed allocating statistic counter. Probably all are used.");
         } else {
             if (hw_id > m_max_hw_id) {
                 m_max_hw_id = hw_id;
             }
             uint32_t user_id = stream->m_rx_check.m_pg_id;
-            m_user_id_map.start_stream(user_id, hw_id);
+            m_user_id_map.start_stream(user_id, hw_id); // ??? can throw exception. return hw_id
             m_hw_id_map.map(hw_id, user_id);
             add_hw_rule(hw_id, m_user_id_map.l4_proto(user_id));
         }
     }
 
-    parser.set_ip_id(IP_ID_RESERVE_BASE + hw_id);
+    m_parser->set_ip_id(IP_ID_RESERVE_BASE + hw_id);
 
     ret_hw_id = hw_id;
 
@@ -587,13 +612,13 @@ int CFlowStatRuleMgr::stop_stream(const TrexStream * stream) {
 #ifdef __DEBUG_FUNC_ENTRY__
     std::cout << __METHOD_NAME__ << " user id:" << stream->m_rx_check.m_pg_id << std::endl;
 #endif
-    if (no_stat_supported)
-        return -ENOTSUP;
-
     if (! stream->m_rx_check.m_enabled) {
         return 0;
     }
 
+    if (! m_api)
+        throw TrexException("Called stop_stream, but no stream was added");
+
     if (m_user_id_map.stop_stream(stream->m_rx_check.m_pg_id) == 0) {
         // last stream associated with the entry stopped transmittig.
         // remove user_id <--> hw_id mapping
@@ -601,7 +626,7 @@ int CFlowStatRuleMgr::stop_stream(const TrexStream * stream) {
         uint16_t hw_id = m_user_id_map.get_hw_id(stream->m_rx_check.m_pg_id);
         if (hw_id >= MAX_FLOW_STATS) {
             fprintf(stderr, "Error: %s got wrong hw_id %d from unmap\n", __func__, hw_id);
-            return -1;
+            throw TrexException("Internal error in stop_stream. Got bad hw_id");
         } else {
             // update counters, and reset before unmapping
             CFlowStatUserIdInfo *p_user_id = m_user_id_map.find_user_id(m_hw_id_map.get_user_id(hw_id));
diff --git a/src/flow_stat.h b/src/flow_stat.h
index 83f076de..ea33062d 100644
--- a/src/flow_stat.h
+++ b/src/flow_stat.h
@@ -104,7 +104,7 @@ typedef class tx_per_flow_t_ tx_per_flow_t;
 typedef class tx_per_flow_t_ rx_per_flow_t;
 
 class CPhyEthIF;
-class Cxl710Parser;
+class CFlowStatParser;
 
 class CFlowStatUserIdInfo {
  public:
@@ -198,6 +198,7 @@ class CFlowStatRuleMgr {
     };
 
     CFlowStatRuleMgr();
+    ~CFlowStatRuleMgr();
     friend std::ostream& operator<<(std::ostream& os, const CFlowStatRuleMgr& cf);
     int add_stream(const TrexStream * stream);
     int del_stream(const TrexStream * stream);
@@ -207,7 +208,8 @@ class CFlowStatRuleMgr {
     bool dump_json(std::string & json, bool baseline);
 
  private:
-    int compile_stream(const TrexStream * stream, Cxl710Parser &parser);
+    void create();
+    int compile_stream(const TrexStream * stream, CFlowStatParser *parser);
     int add_hw_rule(uint16_t hw_id, uint8_t proto);
     void send_start_stop_msg_to_rx(bool is_start);
 
@@ -219,6 +221,8 @@ class CFlowStatRuleMgr {
     int m_max_hw_id; // max hw id we ever used
     uint32_t m_num_started_streams; // How many started (transmitting) streams we have
     CNodeRing *m_ring_to_rx; // handle for sending messages to Rx core
+    CFlowStatParser *m_parser;
+    uint16_t m_capabilities;
 };
 
 #endif
diff --git a/src/flow_stat_parser.cpp b/src/flow_stat_parser.cpp
index 52824f73..8a77c82d 100644
--- a/src/flow_stat_parser.cpp
+++ b/src/flow_stat_parser.cpp
@@ -25,38 +25,36 @@
 #include <common/Network/Packet/EthernetHeader.h>
 #include <flow_stat_parser.h>
 
-Cxl710Parser::Cxl710Parser() {
-    reset();
-}
-
-void Cxl710Parser::reset() {
+void CFlowStatParser::reset() {
     m_ipv4 = 0;
     m_l4_proto = 0;
-    m_fdir_supported = false;
+    m_stat_supported = false;
 }
 
-int Cxl710Parser::parse(uint8_t *p, uint16_t len) {
+int CFlowStatParser::parse(uint8_t *p, uint16_t len) {
     EthernetHeader *ether = (EthernetHeader *)p;
 
+    reset();
+
     switch( ether->getNextProtocol() ) {
     case EthernetHeader::Protocol::IP :
         m_ipv4 = (IPHeader *)(p + 14);
-        m_fdir_supported = true;
+        m_stat_supported = true;
         break;
     case EthernetHeader::Protocol::VLAN :
         switch ( ether->getVlanProtocol() ){
         case EthernetHeader::Protocol::IP:
             m_ipv4 = (IPHeader *)(p + 18);
-            m_fdir_supported = true;
+            m_stat_supported = true;
             break;
         default:
-            m_fdir_supported = false;
+            m_stat_supported = false;
             return -1;
         }
 
         break;
     default:
-        m_fdir_supported = false;
+        m_stat_supported = false;
         return -1;
         break;
     }
@@ -64,7 +62,7 @@ int Cxl710Parser::parse(uint8_t *p, uint16_t len) {
     return 0;
 }
 
-int Cxl710Parser::get_ip_id(uint16_t &ip_id) {
+int CFlowStatParser::get_ip_id(uint16_t &ip_id) {
     if (! m_ipv4)
         return -1;
 
@@ -73,7 +71,7 @@ int Cxl710Parser::get_ip_id(uint16_t &ip_id) {
     return 0;
 }
 
-int Cxl710Parser::set_ip_id(uint16_t new_id) {
+int CFlowStatParser::set_ip_id(uint16_t new_id) {
     if (! m_ipv4)
         return -1;
 
@@ -84,7 +82,7 @@ int Cxl710Parser::set_ip_id(uint16_t new_id) {
     return 0;
 }
 
-int Cxl710Parser::get_l4_proto(uint8_t &proto) {
+int CFlowStatParser::get_l4_proto(uint8_t &proto) {
     if (! m_ipv4)
         return -1;
 
@@ -96,7 +94,7 @@ int Cxl710Parser::get_l4_proto(uint8_t &proto) {
 static const uint16_t TEST_IP_ID = 0xabcd;
 static const uint8_t TEST_L4_PROTO = 0x11;
 
-int Cxl710Parser::test() {
+int CFlowStatParser::test() {
     uint16_t ip_id = 0;
     uint8_t l4_proto;
     uint8_t test_pkt[] = {
@@ -124,14 +122,34 @@ int Cxl710Parser::test() {
     assert(m_ipv4->isChecksumOK() == true);
     assert(get_l4_proto(l4_proto) == 0);
     assert(l4_proto == TEST_L4_PROTO);
-    assert(m_fdir_supported == true);
+    assert(m_stat_supported == true);
 
     reset();
 
     // bad packet
     test_pkt[16] = 0xaa;
     assert (parse(test_pkt, sizeof(test_pkt)) == -1);
-    assert(m_fdir_supported == false);
+    assert(m_stat_supported == false);
+
+    return 0;
+}
+
+// In 82599 10G card we do not support VLANs
+int C82599Parser::parse(uint8_t *p, uint16_t len) {
+    EthernetHeader *ether = (EthernetHeader *)p;
+
+    reset();
+
+    switch( ether->getNextProtocol() ) {
+    case EthernetHeader::Protocol::IP :
+        m_ipv4 = (IPHeader *)(p + 14);
+        m_stat_supported = true;
+        break;
+    default:
+        m_stat_supported = false;
+        return -1;
+        break;
+    }
 
     return 0;
 }
diff --git a/src/flow_stat_parser.h b/src/flow_stat_parser.h
index 606a1bec..8c9e1418 100644
--- a/src/flow_stat_parser.h
+++ b/src/flow_stat_parser.h
@@ -19,19 +19,33 @@
   limitations under the License.
 */
 
-class Cxl710Parser {
+#ifndef __FLOW_STAT_PARSER_H__
+#define __FLOW_STAT_PARSER_H__
+
+// Basic flow stat parser. Relevant for xl710/x710/x350 cards
+#include "common/Network/Packet/IPHeader.h"
+
+class CFlowStatParser {
  public:
-    Cxl710Parser();
-    void reset();
-    int parse(uint8_t *pkt, uint16_t len);
-    bool is_fdir_supported() {return m_fdir_supported == true;};
-    int get_ip_id(uint16_t &ip_id);
-    int set_ip_id(uint16_t ip_id);
-    int get_l4_proto(uint8_t &proto);
-    int test();
+    virtual ~CFlowStatParser() {};
+    virtual void reset();
+    virtual int parse(uint8_t *pkt, uint16_t len);
+    virtual bool is_stat_supported() {return m_stat_supported == true;};
+    virtual int get_ip_id(uint16_t &ip_id);
+    virtual int set_ip_id(uint16_t ip_id);
+    virtual int get_l4_proto(uint8_t &proto);
+    virtual int test();
 
- private:
+ protected:
     IPHeader *m_ipv4;
-    bool m_fdir_supported;
+    bool m_stat_supported;
     uint8_t m_l4_proto;
 };
+
+class C82599Parser : public CFlowStatParser {
+ public:
+    ~C82599Parser() {};
+    int parse(uint8_t *pkt, uint16_t len);
+};
+
+#endif
diff --git a/src/gtest/trex_stateless_gtest.cpp b/src/gtest/trex_stateless_gtest.cpp
index c3dfcb95..a5cf3307 100644
--- a/src/gtest/trex_stateless_gtest.cpp
+++ b/src/gtest/trex_stateless_gtest.cpp
@@ -3581,7 +3581,7 @@ class rx_stat_pkt_parse  : public testing::Test {
 
 
 TEST_F(rx_stat_pkt_parse, x710_parser) {
-    Cxl710Parser parser;
+    CFlowStatParser parser;
 
     parser.test();
 }
diff --git a/src/internal_api/trex_platform_api.h b/src/internal_api/trex_platform_api.h
index dbca5a8a..90eaa7c7 100644
--- a/src/internal_api/trex_platform_api.h
+++ b/src/internal_api/trex_platform_api.h
@@ -26,6 +26,7 @@ limitations under the License.
 #include <vector>
 #include <string>
 #include <string.h>
+#include "flow_stat_parser.h"
 #include "trex_defs.h"
 
 /**
@@ -34,6 +35,7 @@ limitations under the License.
  * @author imarom (06-Oct-15)
  */
 
+
 class TrexPlatformGlobalStats {
 public:
     TrexPlatformGlobalStats() {
@@ -151,6 +153,7 @@ public:
     virtual bool get_promiscuous(uint8_t port_id) const = 0;
     virtual void flush_dp_messages() const = 0;
     virtual int get_active_pgids(flow_stat_active_t &result) const = 0;
+    virtual CFlowStatParser *get_flow_stat_parser() const = 0;
     virtual ~TrexPlatformApi() {}
 };
 
@@ -180,6 +183,7 @@ public:
     bool get_promiscuous(uint8_t port_id) const;
     void flush_dp_messages() const;
     int get_active_pgids(flow_stat_active_t &result) const;
+    CFlowStatParser *get_flow_stat_parser() const;
 };
 
 
@@ -241,6 +245,7 @@ public:
     void flush_dp_messages() const {
     }
     int get_active_pgids(flow_stat_active_t &result) const {return 0;}
+    CFlowStatParser *get_flow_stat_parser() const {return new CFlowStatParser();}
 
 private:
     int m_dp_core_count;
diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp
index ee408c63..46e9a95e 100644
--- a/src/main_dpdk.cpp
+++ b/src/main_dpdk.cpp
@@ -148,6 +148,7 @@ public:
     virtual int dump_fdir_global_stats(CPhyEthIF * _if, FILE *fd) { return -1;}
     virtual int get_stat_counters_num() {return 0;}
     virtual int get_rx_stat_capabilities() {return 0;}
+    virtual CFlowStatParser *get_flow_stat_parser();
 };
 
 
@@ -281,6 +282,7 @@ public:
     virtual int wait_for_stable_link();
     virtual int get_stat_counters_num() {return MAX_FLOW_STATS;}
     virtual int get_rx_stat_capabilities() {return TrexPlatformApi::IF_STAT_IPV4_ID;}
+    virtual CFlowStatParser *get_flow_stat_parser();
 };
 
 class CTRexExtendedDriverBase40G : public CTRexExtendedDriverBase10G {
@@ -332,9 +334,12 @@ public:
     // disabling flow control on 40G using DPDK API causes the interface to malfunction
     virtual bool flow_control_disable_supported(){return false;}
     virtual bool hw_rx_stat_supported(){return true;}
+    virtual CFlowStatParser *get_flow_stat_parser();
+
 private:
     virtual void add_del_rules(enum rte_filter_op op, uint8_t port_id, uint16_t type, uint8_t ttl, uint16_t ip_id, int queue, uint16_t stat_idx);
     virtual int configure_rx_filter_rules_statfull(CPhyEthIF * _if);
+
 private:
     uint8_t m_if_per_card;
 };
@@ -4537,6 +4542,12 @@ int CTRexExtendedDriverBase::configure_drop_queue(CPhyEthIF * _if) {
     return (rte_eth_dev_rx_queue_stop(port_id, 0));
 }
 
+CFlowStatParser *CTRexExtendedDriverBase::get_flow_stat_parser() {
+    CFlowStatParser *parser = new CFlowStatParser();
+    assert (parser);
+    return parser;
+}
+
 void wait_x_sec(int sec) {
     int i;
     printf(" wait %d sec ", sec);
@@ -4940,6 +4951,12 @@ int CTRexExtendedDriverBase10G::wait_for_stable_link(){
     return (0);
 }
 
+CFlowStatParser *CTRexExtendedDriverBase10G::get_flow_stat_parser() {
+    CFlowStatParser *parser = new C82599Parser();
+    assert (parser);
+    return parser;
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 void CTRexExtendedDriverBase40G::clear_extended_stats(CPhyEthIF * _if){
     rte_eth_stats_reset(_if->get_port_id());
@@ -5167,6 +5184,12 @@ int CTRexExtendedDriverBase40G::wait_for_stable_link(){
     return (0);
 }
 
+CFlowStatParser *CTRexExtendedDriverBase40G::get_flow_stat_parser() {
+    CFlowStatParser *parser = new CFlowStatParser();
+    assert (parser);
+    return parser;
+}
+
 /////////////////////////////////////////////////////////////////////
 
 
@@ -5407,3 +5430,8 @@ void TrexDpdkPlatformApi::flush_dp_messages() const {
 int TrexDpdkPlatformApi::get_active_pgids(flow_stat_active_t &result) const {
     return g_trex.m_trex_stateless->m_rx_flow_stat.get_active_pgids(result);
 }
+
+CFlowStatParser *TrexDpdkPlatformApi::get_flow_stat_parser() const {
+    return CTRexExtendedDriverDb::Ins()->get_drv()
+        ->get_flow_stat_parser();
+}
diff --git a/src/stateless/rx/trex_stateless_rx_core.cpp b/src/stateless/rx/trex_stateless_rx_core.cpp
index ab7c08d1..929ad7fa 100644
--- a/src/stateless/rx/trex_stateless_rx_core.cpp
+++ b/src/stateless/rx/trex_stateless_rx_core.cpp
@@ -101,7 +101,7 @@ void CRxCoreStateless::start() {
 }
 
 void CRxCoreStateless::handle_rx_pkt(CLatencyManagerPerPort *lp, rte_mbuf_t *m) {
-    Cxl710Parser parser;
+    CFlowStatParser parser;
 
     if (parser.parse(rte_pktmbuf_mtod(m, uint8_t *), m->pkt_len) == 0) {
         uint16_t ip_id;