tx stats per stream

12 files changed, 287 insertions, 135 deletions

diff --git a/src/bp_sim.h b/src/bp_sim.h
index c9550dcf..68d74b9b 100755
--- a/src/bp_sim.h
+++ b/src/bp_sim.h
@@ -57,6 +57,7 @@ limitations under the License.
 #include <common/cgen_map.h>
 #include <arpa/inet.h>
 #include "platform_cfg.h"
+#include "flow_stat.h"
 
 #include <trex_stateless_dp_core.h>
 
@@ -258,9 +259,6 @@ void on_node_last(uint8_t plugin_id,CGenNode *     node);
 
 rte_mbuf_t * on_node_generate_mbuf(uint8_t plugin_id,CGenNode *     node,CFlowPktInfo * pkt_info);
 
-
-
-
 class CPreviewMode ;
 struct CGenNode;
 /* represent the virtual interface 
@@ -280,7 +278,7 @@ public:
     uint64_t   m_tx_drop;
     uint64_t   m_tx_queue_full;
     uint64_t   m_tx_alloc_error;
-
+    tx_per_flow_t m_tx_per_flow[MAX_FLOW_STATS];    
     CPerTxthreadTemplateInfo m_template;
 
 public:
diff --git a/src/debug.cpp b/src/debug.cpp
index 0ca34545..3f8ad2fa 100644
--- a/src/debug.cpp
+++ b/src/debug.cpp
@@ -359,10 +359,9 @@ int CTrexDebug::test_send(uint pkt_type) {
         exit(-1);
     }
 
-    // read first time to zero statistics
     for (port_id = 0; port_id < m_max_ports; port_id++) {
         CPhyEthIF * lp=&m_ports[port_id];
-        lp->get_rx_stats(NULL, -1, true);
+        lp->reset_hw_flow_stats();
     }
 
     printf("Sending packet:\n");
@@ -410,10 +409,10 @@ int CTrexDebug::test_send(uint pkt_type) {
         lp->dump_stats_extended(stdout);
     }
     for (port_id = 0; port_id < m_max_ports; port_id++) {
-        uint64_t fdir_stat[TREX_FDIR_STAT_SIZE];
+        uint64_t fdir_stat[MAX_FLOW_STATS];
         CPhyEthIF *lp = &m_ports[port_id];
-        if (lp->get_rx_stats(fdir_stat, -1, false) == 0)
-            rte_stat_dump_array(fdir_stat, "FDIR stat", TREX_FDIR_STAT_SIZE);
+        if (lp->get_flow_stats(fdir_stat, NULL, 0, MAX_FLOW_STATS, false) == 0)
+            rte_stat_dump_array(fdir_stat, "FDIR stat", MAX_FLOW_STATS);
     }
 
     return (0);
diff --git a/src/flow_stat.cpp b/src/flow_stat.cpp
index e97586f7..659de09b 100644
--- a/src/flow_stat.cpp
+++ b/src/flow_stat.cpp
@@ -33,8 +33,8 @@
 #define FLOW_STAT_ADD_ALL_PORTS 255
 
 static const uint16_t FREE_HW_ID = UINT16_MAX;
+static bool no_stat_supported = true;
 
-#ifdef __DEBUG_FUNC_ENTRY__
 inline std::string methodName(const std::string& prettyFunction)
 {
     size_t colons = prettyFunction.find("::");
@@ -45,6 +45,7 @@ inline std::string methodName(const std::string& prettyFunction)
 }
 
 #define __METHOD_NAME__ methodName(__PRETTY_FUNCTION__)
+#ifdef __DEBUG_FUNC_ENTRY__
 #define FUNC_ENTRY (std::cout << __METHOD_NAME__ << std::endl);
 #else
 #define FUNC_ENTRY
@@ -117,7 +118,7 @@ void CFlowStatUserIdInfo::reset_hw_id() {
         m_rx_counter_base[i] += m_rx_counter[i];
         m_rx_counter[i] = 0;
         m_tx_counter_base[i] += m_tx_counter[i];
-        m_tx_counter[i] = 0;
+        memset(&m_tx_counter[i], 0, sizeof(m_tx_counter[0]));
     }
 }
 /************** class CFlowStatUserIdMap ***************/
@@ -378,6 +379,7 @@ void CFlowStatHwIdMap::unmap(uint16_t hw_id) {
 /************** class CFlowStatRuleMgr ***************/
 CFlowStatRuleMgr::CFlowStatRuleMgr() {
     m_api = NULL;
+    m_max_hw_id = -1;
 }
 
 std::ostream& operator<<(std::ostream& os, const CFlowStatRuleMgr& cf) {
@@ -396,11 +398,11 @@ int CFlowStatRuleMgr::compile_stream(const TrexStream * stream, Cxl710Parser &pa
     // 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);    
+    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 we could not parse the packet, but no stat count needed, it is probably OK.
         if (stream->m_rx_check.m_enabled) {
@@ -432,10 +434,24 @@ int CFlowStatRuleMgr::add_stream(const TrexStream * stream) {
     if (! m_api ) {
         TrexStateless *tstateless = get_stateless_obj();
         m_api = tstateless->get_platform_api();
-        //    m_api = get_stateless_obj()->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);
+        }
     }
-    
+
+    if (no_stat_supported)
+        return -ENOTSUP;
+
     Cxl710Parser parser;
     int ret;
 
@@ -460,6 +476,9 @@ int CFlowStatRuleMgr::del_stream(const TrexStream * stream) {
     std::cout << __METHOD_NAME__ << " user id:" << stream->m_rx_check.m_user_id << std::endl;
 #endif
 
+    if (no_stat_supported)
+        return -ENOTSUP;
+
     if (! stream->m_rx_check.m_enabled) {
         return 0;
     }
@@ -473,7 +492,7 @@ int CFlowStatRuleMgr::del_stream(const TrexStream * stream) {
 // If stream does not need flow stat counting, make sure it does not interfere with
 // other streams that do need stat counting.
 // Might change the IP ID of the stream packet
-int CFlowStatRuleMgr::start_stream(TrexStream * stream) {
+int CFlowStatRuleMgr::start_stream(TrexStream * stream, uint16_t &ret_hw_id) {
 #ifdef __DEBUG_FUNC_ENTRY__
     std::cout << __METHOD_NAME__ << " user id:" << stream->m_rx_check.m_user_id << std::endl;
 #endif
@@ -481,9 +500,8 @@ int CFlowStatRuleMgr::start_stream(TrexStream * stream) {
     Cxl710Parser parser;
     int ret;
 
-    if (! m_api ) {
-        return 0;
-    }
+    if (no_stat_supported)
+        return -ENOTSUP;
 
     if ((ret = compile_stream(stream, parser)) < 0)
         return ret;
@@ -509,6 +527,9 @@ int CFlowStatRuleMgr::start_stream(TrexStream * stream) {
         m_user_id_map.start_stream(stream->m_rx_check.m_user_id); // just increase ref count;
     } else {
         uint16_t hw_id = m_hw_id_map.find_free_hw_id();
+        if (hw_id > m_max_hw_id) {
+            m_max_hw_id = hw_id;
+        }
         if (hw_id == FREE_HW_ID) {
             printf("Error: %s failed finding free hw_id\n", __func__);
             return -1;
@@ -523,10 +544,16 @@ int CFlowStatRuleMgr::start_stream(TrexStream * stream) {
     uint16_t hw_id = m_user_id_map.get_hw_id(stream->m_rx_check.m_user_id); // can't fail if we got here
     parser.set_ip_id(IP_ID_RESERVE_BASE + hw_id);
 
+    ret_hw_id = hw_id;
+
+#ifdef __DEBUG_FUNC_ENTRY__
+    std::cout << "exit:" << __METHOD_NAME__ << " hw_id:" << ret_hw_id << std::endl;
+#endif
+
     return 0;
 }
 
-int CFlowStatRuleMgr::add_hw_rule(uint16_t hw_id, uint8_t proto) {    
+int CFlowStatRuleMgr::add_hw_rule(uint16_t hw_id, uint8_t proto) {
     for (int port = 0; port < m_num_ports; port++) {
         m_api->add_rx_flow_stat_rule(port, FLOW_STAT_RULE_TYPE_IPV4_ID, proto, hw_id);
     }
@@ -538,12 +565,12 @@ int CFlowStatRuleMgr::stop_stream(const TrexStream * stream) {
 #ifdef __DEBUG_FUNC_ENTRY__
     std::cout << __METHOD_NAME__ << " user id:" << stream->m_rx_check.m_user_id << std::endl;
 #endif
+    if (no_stat_supported)
+        return -ENOTSUP;
+
     if (! stream->m_rx_check.m_enabled) {
         return 0;
     }
-    if (! m_api ) {
-        return 0;
-    }
 
     if (m_user_id_map.stop_stream(stream->m_rx_check.m_user_id) == 0) {
         // last stream associated with the entry stopped transmittig.
@@ -557,12 +584,13 @@ int CFlowStatRuleMgr::stop_stream(const TrexStream * stream) {
             // update counters, and reset before unmapping
             class CFlowStatUserIdInfo *p_user_id = m_user_id_map.find_user_id(m_hw_id_map.get_user_id(hw_id));
             assert(p_user_id != NULL);
-            uint64_t counter;
+            uint64_t rx_counter;
+            tx_per_flow_t tx_counter;
             for (uint8_t port = 0; port < m_num_ports; port++) {
                 m_api->del_rx_flow_stat_rule(port, FLOW_STAT_RULE_TYPE_IPV4_ID, proto, hw_id);
-                m_api->get_rx_stats(port, &counter, hw_id, true);
-                p_user_id->set_rx_counter(port, counter);
-                p_user_id->set_tx_counter(port, counter); //??? until tx work, just set for same value
+                m_api->get_flow_stats(port, &rx_counter, (void *)&tx_counter, hw_id, hw_id, true);
+                p_user_id->set_rx_counter(port, rx_counter);
+                p_user_id->set_tx_counter(port, tx_counter);
             }
             m_hw_id_map.unmap(hw_id);
         }
@@ -572,7 +600,8 @@ int CFlowStatRuleMgr::stop_stream(const TrexStream * stream) {
 
 // return false if no counters changed since last run. true otherwise
 bool CFlowStatRuleMgr::dump_json(std::string & json) {
-    uint64_t stats[TREX_FDIR_STAT_SIZE];
+    uint64_t rx_stats[MAX_FLOW_STATS];
+    tx_per_flow_t tx_stats[MAX_FLOW_STATS];
     Json::FastWriter writer;
     Json::Value root;
     bool ret = false;
@@ -587,15 +616,24 @@ bool CFlowStatRuleMgr::dump_json(std::string & json) {
     // read hw counters, and update
     data_section["timestamp"] = Json::Value::UInt64(os_get_hr_tick_64());
     for (uint8_t port = 0; port < m_num_ports; port++) {
-        int rc = m_api->get_rx_stats(port, stats, -1, false);
-        if (rc == -1) {
-            continue;
-        }
-
-        for (int i = 0; i < TREX_FDIR_STAT_SIZE; i++) {
-            if (stats[i] != 0) {
-                m_user_id_map.find_user_id(m_hw_id_map.get_user_id(i))->set_rx_counter(port, stats[i]);
-                m_user_id_map.find_user_id(m_hw_id_map.get_user_id(i))->set_tx_counter(port, stats[i]);  //??? until tx work, just set for same value
+        m_api->get_flow_stats(port, rx_stats, (void *)tx_stats, 0, m_max_hw_id, false);
+        for (int i = 0; i <= m_max_hw_id; i++) {
+            if (rx_stats[i] != 0) {
+                class CFlowStatUserIdInfo *p_user_id = m_user_id_map.find_user_id(m_hw_id_map.get_user_id(i));
+                if (likely(p_user_id != NULL)) {
+                    p_user_id->set_rx_counter(port, rx_stats[i]);
+                } else {
+                    std::cerr <<  __METHOD_NAME__ << i << ":Could not count " << rx_stats[i] << " rx packets, because no mapping was found" << std::endl;
+                }
+            }
+            if (tx_stats[i].get_pkts() != 0) {
+                tx_per_flow_t tx_pkts = tx_stats[i];
+                class CFlowStatUserIdInfo *p_user_id = m_user_id_map.find_user_id(m_hw_id_map.get_user_id(i));
+                if (likely(p_user_id != NULL)) {
+                    p_user_id->set_tx_counter(port, tx_pkts);
+                } else {
+                    std::cerr <<  __METHOD_NAME__ << i << ":Could not count tx " << tx_pkts << " because no mapping was found" << std::endl;
+                }
             }
         }
     }
@@ -608,11 +646,14 @@ bool CFlowStatRuleMgr::dump_json(std::string & json) {
         std::string str_user_id = static_cast<std::ostringstream*>( &(std::ostringstream()
                                                                       << user_id) )->str();
         for (uint8_t port = 0; port < m_num_ports; port++) {
-            if ((user_id_info->get_tx_counter(port) != 0) || (user_id_info->get_rx_counter(port) != 0)) {
-                std::string str_port = static_cast<std::ostringstream*>( &(std::ostringstream()
-                                                                           << port) )->str();
+            std::string str_port = static_cast<std::ostringstream*>( &(std::ostringstream() << port) )->str();
+            if (user_id_info->get_rx_counter(port) != 0) {
                 data_section[str_user_id]["rx"][str_port] = Json::Value::UInt64(user_id_info->get_rx_counter(port));
-                data_section[str_user_id]["tx"][str_port] = Json::Value::UInt64(user_id_info->get_tx_counter(port));
+                ret = true;
+            }
+            if (user_id_info->get_tx_counter(port).get_pkts() != 0) {
+                data_section[str_user_id]["tx-pkts"][str_port] = Json::Value::UInt64(user_id_info->get_tx_counter(port).get_pkts());
+                data_section[str_user_id]["tx-bytes"][str_port] = Json::Value::UInt64(user_id_info->get_tx_counter(port).get_bytes());
                 ret = true;
             }
         }
diff --git a/src/flow_stat.h b/src/flow_stat.h
index 444daab0..4ea59a2c 100644
--- a/src/flow_stat.h
+++ b/src/flow_stat.h
@@ -25,8 +25,9 @@
 #include <string>
 #include <map>
 #include "trex_defs.h"
+#include "trex_stream.h"
+#include <internal_api/trex_platform_api.h>
 
-#define MAX_FLOW_STATS 128
 // range reserved for rx stat measurement is from IP_ID_RESERVE_BASE to 0xffff
 // Do not change this value. In i350 cards, we filter according to first byte of IP ID
 // In other places, we identify packets by if (ip_id > IP_ID_RESERVE_BASE)
@@ -35,6 +36,65 @@
 typedef std::map<uint32_t, uint16_t> flow_stat_map_t;
 typedef std::map<uint32_t, uint16_t>::iterator flow_stat_map_it_t;
 
+class tx_per_flow_t_ {
+ public:
+    tx_per_flow_t_() {
+        clear();
+    }
+    inline uint64_t get_bytes() {
+        return m_bytes;
+    }
+    inline uint64_t get_pkts() {
+        return m_pkts;
+    }
+    inline void set_bytes(uint64_t bytes) {
+        m_bytes = bytes;;
+    }
+    inline void get_pkts(uint64_t pkts) {
+        m_pkts = pkts;
+    }
+    inline void add_bytes(uint64_t bytes) {
+        m_bytes += bytes;;
+    }
+    inline void add_pkts(uint64_t pkts) {
+        m_pkts += pkts;
+    }
+    inline void clear() {
+        m_bytes = 0;
+        m_pkts = 0;
+    }
+    inline tx_per_flow_t_ operator+ (const tx_per_flow_t_ &t_in) {
+        tx_per_flow_t_ t_out;
+        t_out.m_bytes = this->m_bytes + t_in.m_bytes;
+        t_out.m_pkts = this->m_pkts + t_in.m_pkts;
+        return t_out;
+    }
+
+    inline tx_per_flow_t_ operator- (const tx_per_flow_t_ &t_in) {
+        tx_per_flow_t_ t_out;
+        t_out.m_bytes = this->m_bytes - t_in.m_bytes;
+        t_out.m_pkts = this->m_pkts - t_in.m_pkts;
+        return t_out;
+    }
+
+    inline tx_per_flow_t_ operator+= (const tx_per_flow_t_ &t_in) {
+        m_bytes += t_in.m_bytes;
+        m_pkts += t_in.m_pkts;
+        return *this;
+    }
+
+    friend std::ostream& operator<<(std::ostream& os, const class tx_per_flow_t_ &t) {
+        os  << "p:" << t.m_pkts << " b:" << t.m_bytes;
+        return os;
+    }
+
+ private:
+    uint64_t m_bytes;
+    uint64_t m_pkts;
+};
+
+typedef class tx_per_flow_t_ tx_per_flow_t;
+
 class CPhyEthIF;
 class Cxl710Parser;
 
@@ -44,8 +104,8 @@ class CFlowStatUserIdInfo {
     friend std::ostream& operator<<(std::ostream& os, const CFlowStatUserIdInfo& cf);
     void set_rx_counter(uint8_t port, uint64_t val) {m_rx_counter[port] = val;}
     uint64_t get_rx_counter(uint8_t port) {return m_rx_counter[port] + m_rx_counter_base[port];}
-    void set_tx_counter(uint8_t port, uint64_t val) {m_tx_counter[port] = val;}
-    uint64_t get_tx_counter(uint8_t port) {return m_tx_counter[port] + m_tx_counter_base[port];}
+    void set_tx_counter(uint8_t port, tx_per_flow_t val) {m_tx_counter[port] = val;}
+    tx_per_flow_t get_tx_counter(uint8_t port) {return m_tx_counter[port] + m_tx_counter_base[port];}
     void set_hw_id(uint16_t hw_id) {m_hw_id = hw_id;}
     uint64_t get_hw_id() {return m_hw_id;}
     void reset_hw_id();
@@ -62,9 +122,9 @@ class CFlowStatUserIdInfo {
     uint64_t m_rx_counter[TREX_MAX_PORTS]; // How many packets received with this user id since stream start
     // How many packets received with this user id, since stream creation, before stream start.
     uint64_t m_rx_counter_base[TREX_MAX_PORTS];
-    uint64_t m_tx_counter[TREX_MAX_PORTS]; // How many packets transmitted with this user id since stream start
+    tx_per_flow_t m_tx_counter[TREX_MAX_PORTS]; // How many packets transmitted with this user id since stream start
     // How many packets transmitted with this user id, since stream creation, before stream start.
-    uint64_t m_tx_counter_base[TREX_MAX_PORTS];
+    tx_per_flow_t m_tx_counter_base[TREX_MAX_PORTS];
     uint16_t m_hw_id;     // Associated hw id. UINT16_MAX if no associated hw id.
     uint8_t m_proto;      // protocol (UDP, TCP, other), associated with this user id.
     uint8_t m_ref_count;  // How many streams with this ref count exists
@@ -121,7 +181,7 @@ class CFlowStatRuleMgr {
     friend std::ostream& operator<<(std::ostream& os, const CFlowStatRuleMgr& cf);
     int add_stream(const TrexStream * stream);
     int del_stream(const TrexStream * stream);
-    int start_stream(TrexStream * stream);
+    int start_stream(TrexStream * stream, uint16_t &ret_hw_id);
     int stop_stream(const TrexStream * stream);
     bool dump_json(std::string & json);
 
@@ -134,6 +194,7 @@ class CFlowStatRuleMgr {
     class CFlowStatUserIdMap m_user_id_map; // map user ids to hw ids
     uint8_t m_num_ports; // How many ports are being used
     const TrexPlatformApi *m_api;
+    int m_max_hw_id; // max hw id we ever used
 };
 
 #endif
diff --git a/src/internal_api/trex_platform_api.h b/src/internal_api/trex_platform_api.h
index 847611e4..40ce374d 100644
--- a/src/internal_api/trex_platform_api.h
+++ b/src/internal_api/trex_platform_api.h
@@ -141,7 +141,8 @@ public:
     virtual void publish_async_data_now(uint32_t key) const = 0;
     virtual uint8_t get_dp_core_count() const = 0;
     virtual void get_interface_stat_info(uint8_t interface_id, uint16_t &num_counters, uint16_t &capabilities) const =0;
-    virtual int get_rx_stats(uint8_t port_id, uint64_t *stats, int index, bool reset) const = 0;
+    virtual int get_flow_stats(uint8_t port_id, uint64_t *stats, void *tx_stats, int min, int max, bool reset) const = 0;
+    virtual int reset_hw_flow_stats(uint8_t port_id) const = 0;
     virtual void get_port_num(uint8_t &port_num) const = 0;
     virtual int add_rx_flow_stat_rule(uint8_t port_id, uint8_t type, uint16_t proto, uint16_t id) const = 0;
     virtual int del_rx_flow_stat_rule(uint8_t port_id, uint8_t type, uint16_t proto, uint16_t id) const = 0;
@@ -168,7 +169,8 @@ public:
     void publish_async_data_now(uint32_t key) const;
     uint8_t get_dp_core_count() const;
     void get_interface_stat_info(uint8_t interface_id, uint16_t &num_counters, uint16_t &capabilities) const;
-    int get_rx_stats(uint8_t port_id, uint64_t *stats, int index, bool reset) const;
+    int get_flow_stats(uint8_t port_id, uint64_t *stats, void *tx_stats, int min, int max, bool reset) const;
+    int reset_hw_flow_stats(uint8_t port_id) const;
     void get_port_num(uint8_t &port_num) const;
     int add_rx_flow_stat_rule(uint8_t port_id, uint8_t type, uint16_t proto, uint16_t id) const;
     int del_rx_flow_stat_rule(uint8_t port_id, uint8_t type, uint16_t proto, uint16_t id) const;
diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp
index 5c7be19e..af6efe1d 100644
--- a/src/main_dpdk.cpp
+++ b/src/main_dpdk.cpp
@@ -106,6 +106,8 @@ extern "C" void i40e_set_trex_mode(int mode);
 #define RTE_TEST_TX_DESC_DEFAULT 512
 #define RTE_TEST_RX_DESC_DROP    0
 
+static int max_stat_hw_id_seen = 0;
+
 static inline int get_vm_one_queue_enable(){
     return (CGlobalInfo::m_options.preview.get_vm_one_queue_enable() ?1:0);
 }
@@ -145,7 +147,7 @@ public:
     virtual int  wait_for_stable_link()=0;
     virtual void wait_after_link_up(){};
     virtual bool flow_control_disable_supported(){return true;}
-    virtual int get_rx_stats(CPhyEthIF * _if, uint32_t *stats, uint32_t *prev_stats, int index) {return -1;}
+    virtual int get_rx_stats(CPhyEthIF * _if, uint32_t *stats, uint32_t *prev_stats, int min, int max) {return -1;}
     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;}
@@ -318,9 +320,9 @@ public:
     }
     virtual void get_extended_stats(CPhyEthIF * _if,CPhyEthIFStats *stats);
     virtual void clear_extended_stats(CPhyEthIF * _if);
-    int get_rx_stats(CPhyEthIF * _if, uint32_t *stats, uint32_t *prev_stats, int index);
+    int get_rx_stats(CPhyEthIF * _if, uint32_t *stats, uint32_t *prev_stats, int min, int max);
     int dump_fdir_global_stats(CPhyEthIF * _if, FILE *fd);
-    int get_stat_counters_num() {return TREX_FDIR_STAT_SIZE;}
+    int get_stat_counters_num() {return MAX_FLOW_STATS;}
     int get_rx_stat_capabilities() {return TrexPlatformApi::IF_STAT_IPV4_ID;}
     virtual int wait_for_stable_link();
     // disabling flow control on 40G using DPDK API causes the interface to malfunction
@@ -1191,7 +1193,6 @@ void CPhyEthIFStats::Clear(){
     imcasts = 0;
     rx_nombuf = 0;
     memset(m_rx_per_flow, 0, sizeof(m_rx_per_flow));
-    m_fdir_stats_first_time = true;
 }
 
 
@@ -1560,52 +1561,6 @@ int CPhyEthIF::dump_fdir_global_stats(FILE *fd) {
     return get_ex_drv()->dump_fdir_global_stats(this, fd);
 }
 
-// get/reset flow director counters
-// return 0 if OK. -1 if operation not supported.
-// stats - If not NULL, returning counter numbers in it.
-// index - If non negative, get only counter with this index
-// reset - If true, reset counter value after reading
-int CPhyEthIF::get_rx_stats(uint64_t *stats, int index, bool reset) {
-    uint32_t diff_stats[TREX_FDIR_STAT_SIZE];
-    int start, len;
-
-    if (index >= 0) {
-        start = index;
-        len = 1;
-    } else {
-        start = 0;
-        len = TREX_FDIR_STAT_SIZE;
-    }
-    
-    if (get_ex_drv()->get_rx_stats(this, diff_stats, m_stats.m_fdir_prev_stats, index) < 0) {
-        return -1;
-    }
-
-    // First time, just syncing the counters
-    if (m_stats.m_fdir_stats_first_time) {
-        m_stats.m_fdir_stats_first_time = false;
-        if (stats) {
-            memset(stats, 0, sizeof(uint64_t) * TREX_FDIR_STAT_SIZE);
-        }
-        return 0;
-    }
-
-    for (int i = start; i < (start + len); i++) {
-        if ( reset ) {
-            // return value so far, and reset
-            stats[i] = m_stats.m_rx_per_flow[i] + diff_stats[i];
-            m_stats.m_rx_per_flow[i] = 0;
-        } else {
-            m_stats.m_rx_per_flow[i] += diff_stats[i];
-            if (stats != NULL) {
-                stats[i] = m_stats.m_rx_per_flow[i];
-            }
-        }
-    }
-    
-    return 0;
-}
-
 void dump_hw_state(FILE *fd,struct ixgbe_hw_stats *hs ){
 
     #define DP_A1(f) if (hs->f) fprintf(fd," %-40s : %llu \n",#f, (unsigned long long)hs->f)
@@ -2037,7 +1992,6 @@ void CCoreEthIF::update_mac_addr(CGenNode * node,uint8_t *p){
 
 int CCoreEthIFStateless::send_node(CGenNode * no){
     CGenNodeStateless * node_sl=(CGenNodeStateless *) no;
-
     /* check that we have mbuf  */
     rte_mbuf_t *    m=node_sl->get_cache_mbuf();
     pkt_dir_t dir=(pkt_dir_t)node_sl->get_mbuf_cache_dir();
@@ -2051,6 +2005,15 @@ int CCoreEthIFStateless::send_node(CGenNode * no){
         assert(m);
     }
 
+    if (unlikely(node_sl->is_stat_needed())) {
+        uint16_t hw_id = node_sl->get_stat_hw_id();
+        if (hw_id > max_stat_hw_id_seen) {
+            max_stat_hw_id_seen = hw_id;
+        }
+        tx_per_flow_t *lp_s = &lp_stats->m_tx_per_flow[hw_id];
+        lp_s->add_pkts(1);
+        lp_s->add_bytes(m->pkt_len);
+    }
     send_pkt(lp_port,m,lp_stats);
 
     return (0);
@@ -2292,7 +2255,9 @@ public:
     uint64_t ibytes;
     uint64_t ierrors;
     uint64_t oerrors;     
-
+    tx_per_flow_t m_tx_per_flow[MAX_FLOW_STATS];
+    tx_per_flow_t m_prev_tx_per_flow[MAX_FLOW_STATS];
+    
     float     m_total_tx_bps;
     float     m_total_tx_pps;
 
@@ -2325,7 +2290,7 @@ public:
     uint64_t  m_total_nat_active  ;
     uint64_t  m_total_nat_open    ;
     uint64_t  m_total_nat_learn_error    ;
-
+    
     CPerTxthreadTemplateInfo m_template;
 
     float     m_socket_util;
@@ -2683,6 +2648,8 @@ public:
     void dump_template_info(std::string & json);
     bool sanity_check();
     void update_stats(void);
+    tx_per_flow_t get_flow_tx_stats(uint8_t port, uint16_t hw_id);
+    void clear_flow_tx_stats(uint8_t port, uint16_t index);
     void get_stats(CGlobalStats & stats);
     void dump_post_test_stats(FILE *fd);
     void dump_config(FILE *fd);
@@ -3334,6 +3301,13 @@ void CGlobalTRex::update_stats(){
 
 }
 
+tx_per_flow_t CGlobalTRex::get_flow_tx_stats(uint8_t port, uint16_t index) {
+    return m_stats.m_port[port].m_tx_per_flow[index] - m_stats.m_port[port].m_prev_tx_per_flow[index];
+}
+
+void CGlobalTRex::clear_flow_tx_stats(uint8_t port, uint16_t index) {
+    m_stats.m_port[port].m_prev_tx_per_flow[index] = m_stats.m_port[port].m_tx_per_flow[index];
+}
 
 void CGlobalTRex::get_stats(CGlobalStats & stats){
 
@@ -3383,6 +3357,9 @@ void CGlobalTRex::get_stats(CGlobalStats & stats){
         total_tx_pps +=_if->get_last_tx_pps_rate();
         total_rx_pps +=_if->get_last_rx_pps_rate();
 
+        for (uint16_t flow = 0; flow <= max_stat_hw_id_seen; flow++) {
+            stats.m_port[i].m_tx_per_flow[flow].clear();
+        }
     }
 
     uint64_t total_open_flows=0;
@@ -3400,10 +3377,8 @@ void CGlobalTRex::get_stats(CGlobalStats & stats){
     uint64_t total_nat_open     =0;
     uint64_t total_nat_learn_error=0;
 
-
     CFlowGenListPerThread   * lpt;
     stats.m_template.Clear();
-
     for (i=0; i<get_cores_tx(); i++) {
         lpt = m_fl.m_threads_info[i];
         total_open_flows +=   lpt->m_stats.m_total_open_flows ;
@@ -3414,7 +3389,7 @@ void CGlobalTRex::get_stats(CGlobalStats & stats){
         stats.m_total_queue_full +=lpt->m_node_gen.m_v_if->m_stats[0].m_tx_queue_full+
                                lpt->m_node_gen.m_v_if->m_stats[1].m_tx_queue_full;
 
-        stats.m_total_queue_drop =lpt->m_node_gen.m_v_if->m_stats[0].m_tx_drop+
+        stats.m_total_queue_drop +=lpt->m_node_gen.m_v_if->m_stats[0].m_tx_drop+
                                lpt->m_node_gen.m_v_if->m_stats[1].m_tx_drop;
 
         stats.m_template.Add(&lpt->m_node_gen.m_v_if->m_stats[0].m_template);
@@ -3431,6 +3406,13 @@ void CGlobalTRex::get_stats(CGlobalStats & stats){
         total_nat_active   +=lpt->m_stats.m_nat_lookup_add_flow_id - lpt->m_stats.m_nat_lookup_remove_flow_id;
         total_nat_open     +=lpt->m_stats.m_nat_lookup_add_flow_id;
         total_nat_learn_error   +=lpt->m_stats.m_nat_flow_learn_error;
+        uint8_t port0 = lpt->getDualPortId() *2;
+        for (uint16_t flow = 0; flow <= max_stat_hw_id_seen; flow++) {
+            stats.m_port[port0].m_tx_per_flow[flow] +=
+                lpt->m_node_gen.m_v_if->m_stats[0].m_tx_per_flow[flow];
+            stats.m_port[port0 + 1].m_tx_per_flow[flow] +=
+                lpt->m_node_gen.m_v_if->m_stats[1].m_tx_per_flow[flow];
+        }
     }
 
     stats.m_total_nat_time_out = total_nat_time_out;
@@ -3905,6 +3887,56 @@ int CGlobalTRex::start_master_statefull() {
 
 static CGlobalTRex g_trex;
 
+// The HW counters start from some random values. The driver give us the diffs from previous,
+// each time we do get_rx_stats. We need to make one first call, at system startup,
+// and ignore the returned diffs
+int CPhyEthIF::reset_hw_flow_stats() {
+    uint32_t diff_stats[MAX_FLOW_STATS];
+
+    if (get_ex_drv()->get_rx_stats(this, diff_stats, m_stats.m_fdir_prev_stats, 0, MAX_FLOW_STATS) < 0) {
+        return -1;
+    }
+
+    return 0;
+}
+
+// get/reset flow director counters
+// return 0 if OK. -1 if operation not supported.
+// stats - If not NULL, returning counter numbers in it.
+// index - If non negative, get only counter with this index
+// reset - If true, reset counter value after reading
+int CPhyEthIF::get_flow_stats(uint64_t *rx_stats, tx_per_flow_t *tx_stats, int min, int max, bool reset) {
+    uint32_t diff_stats[MAX_FLOW_STATS];
+    
+    if (get_ex_drv()->get_rx_stats(this, diff_stats, m_stats.m_fdir_prev_stats, min, max) < 0) {
+        return -1;
+    }
+
+    for (int i = min; i <= max; i++) {
+        if ( reset ) {
+            // return value so far, and reset
+            if (rx_stats != NULL) {
+                rx_stats[i] = m_stats.m_rx_per_flow[i] + diff_stats[i];
+            }
+            if (tx_stats != NULL) {
+                tx_stats[i] = g_trex.get_flow_tx_stats(m_port_id, i);
+            }
+            m_stats.m_rx_per_flow[i] = 0;
+            g_trex.clear_flow_tx_stats(m_port_id, i);
+        } else {
+            m_stats.m_rx_per_flow[i] += diff_stats[i];
+            if (rx_stats != NULL) {
+                rx_stats[i] = m_stats.m_rx_per_flow[i];
+            }
+            if (tx_stats != NULL) {
+                tx_stats[i] = g_trex.get_flow_tx_stats(m_port_id, i);
+            }
+        }
+    }
+    
+    return 0;
+}
+
 bool CCoreEthIF::process_rx_pkt(pkt_dir_t   dir,
                                 rte_mbuf_t * m){
 
@@ -4818,9 +4850,9 @@ void CTRexExtendedDriverBase40G::add_del_rules(enum rte_filter_op op, uint8_t po
 
 // type - rule type. Currently we only support rules in IP ID.
 // proto - Packet protocol: UDP or TCP
-// id - Counter id in HW. We assume it is in the range 0..TREX_FDIR_STAT_SIZE
+// id - Counter id in HW. We assume it is in the range 0..MAX_FLOW_STATS
 int CTRexExtendedDriverBase40G::add_del_rx_flow_stat_rule(uint8_t port_id, enum rte_filter_op op, uint8_t type, uint16_t proto, uint16_t id) {
-    uint32_t rule_id = (port_id % m_if_per_card) * TREX_FDIR_STAT_SIZE + id;
+    uint32_t rule_id = (port_id % m_if_per_card) * MAX_FLOW_STATS + id;
     uint16_t rte_type = RTE_ETH_FLOW_NONFRAG_IPV4_OTHER;
     
     switch(proto) {
@@ -4868,20 +4900,12 @@ int CTRexExtendedDriverBase40G::configure_rx_filter_rules(CPhyEthIF * _if) {
 
 // instead of adding this to rte_ethdev.h
 extern "C" int rte_eth_fdir_stats_get(uint8_t port_id, uint32_t *stats, uint32_t start, uint32_t len);
-int CTRexExtendedDriverBase40G::get_rx_stats(CPhyEthIF * _if, uint32_t *stats, uint32_t *prev_stats, int index) {
-    uint32_t hw_stats[TREX_FDIR_STAT_SIZE];
+int CTRexExtendedDriverBase40G::get_rx_stats(CPhyEthIF * _if, uint32_t *stats, uint32_t *prev_stats, int min, int max) {
+    uint32_t hw_stats[MAX_FLOW_STATS];
     uint32_t port_id = _if->get_port_id();
-    uint32_t len, start, loop_start;
-
-    if (index >= 0) {
-        len = 1;
-        start = (port_id % m_if_per_card) * TREX_FDIR_STAT_SIZE + index;
-        loop_start = index;
-    } else {
-        start = (port_id % m_if_per_card) * TREX_FDIR_STAT_SIZE;
-        len = TREX_FDIR_STAT_SIZE;
-        loop_start = 0;
-    }
+    uint32_t start = (port_id % m_if_per_card) * MAX_FLOW_STATS + min;
+    uint32_t len = max - min + 1;
+    uint32_t loop_start = min;
 
     rte_eth_fdir_stats_get(port_id, hw_stats, start, len);
     for (int i = loop_start; i <  loop_start + len; i++) {
@@ -5163,8 +5187,12 @@ TrexDpdkPlatformApi::get_interface_stat_info(uint8_t interface_id, uint16_t &num
     capabilities = CTRexExtendedDriverDb::Ins()->get_drv()->get_rx_stat_capabilities();
 }
 
-int TrexDpdkPlatformApi::get_rx_stats(uint8 port_id, uint64_t *stats, int index, bool reset) const {
-    return g_trex.m_ports[port_id].get_rx_stats(stats, index, reset);
+int TrexDpdkPlatformApi::get_flow_stats(uint8 port_id, uint64_t *rx_stats, void *tx_stats, int min, int max, bool reset) const {
+    return g_trex.m_ports[port_id].get_flow_stats(rx_stats, (tx_per_flow_t *)tx_stats, min, max, reset);
+}
+
+int TrexDpdkPlatformApi::reset_hw_flow_stats(uint8_t port_id) const {
+    return g_trex.m_ports[port_id].reset_hw_flow_stats();
 }
 
 int TrexDpdkPlatformApi::add_rx_flow_stat_rule(uint8_t port_id, uint8_t type, uint16_t proto, uint16_t id) const {
diff --git a/src/main_dpdk.h b/src/main_dpdk.h
index 7357c0f4..33615636 100644
--- a/src/main_dpdk.h
+++ b/src/main_dpdk.h
@@ -38,10 +38,9 @@ class CPhyEthIFStats {
     uint64_t oerrors;   /**< Total number of failed transmitted packets. */
     uint64_t imcasts;   /**< Total number of multicast received packets. */
     uint64_t rx_nombuf; /**< Total number of RX mbuf allocation failures. */
-    uint64_t  m_rx_per_flow [TREX_FDIR_STAT_SIZE]; // Per flow RX statistics
+    uint64_t  m_rx_per_flow [MAX_FLOW_STATS]; // Per flow RX statistics
     // Previous fdir stats values read from HW. Since on xl710 this is 32 bit, we save old value, to handle wrap around.
-    uint32_t  m_fdir_prev_stats [TREX_FDIR_STAT_SIZE];
-    bool m_fdir_stats_first_time;
+    uint32_t  m_fdir_prev_stats [MAX_FLOW_STATS];
  public:
     void Clear();
     void Dump(FILE *fd);
@@ -53,7 +52,6 @@ class CPhyEthIF  {
     CPhyEthIF (){
         m_port_id=0;
         m_rx_queue=0;
-        m_stats.m_fdir_stats_first_time = true;
     }
     bool Create(uint8_t portid){
         m_port_id      = portid;
@@ -74,7 +72,8 @@ class CPhyEthIF  {
     void macaddr_get(struct ether_addr *mac_addr);
     void get_stats(CPhyEthIFStats *stats);
     int dump_fdir_global_stats(FILE *fd);
-    int get_rx_stats(uint64_t *stats, int index, bool reset);
+    int reset_hw_flow_stats();
+    int get_flow_stats(uint64_t *rx_stats, tx_per_flow_t *tx_stats, int min, int max, bool reset);
     void get_stats_1g(CPhyEthIFStats *stats);
     void rx_queue_setup(uint16_t rx_queue_id,
                         uint16_t nb_rx_desc,
@@ -116,6 +115,7 @@ class CPhyEthIF  {
     float get_last_rx_pps_rate(){
         return (m_last_rx_pps);
     }
+
     CPhyEthIFStats     & get_stats(){
         return ( m_stats );
     }
diff --git a/src/stateless/cp/trex_stream.h b/src/stateless/cp/trex_stream.h
index 161e9592..1abf0c04 100644
--- a/src/stateless/cp/trex_stream.h
+++ b/src/stateless/cp/trex_stream.h
@@ -514,7 +514,7 @@ public:
         bool      m_seq_enabled;
         bool      m_latency;
         uint32_t  m_user_id;
-
+        uint16_t  m_hw_id;
     } m_rx_check;
 
     uint32_t   m_burst_total_pkts; /* valid in case of burst stSINGLE_BURST,stMULTI_BURST*/
diff --git a/src/stateless/cp/trex_streams_compiler.cpp b/src/stateless/cp/trex_streams_compiler.cpp
index 7c91754c..be5002da 100644
--- a/src/stateless/cp/trex_streams_compiler.cpp
+++ b/src/stateless/cp/trex_streams_compiler.cpp
@@ -476,7 +476,8 @@ TrexStreamsCompiler::compile_stream(TrexStream *stream,
     }
 
     TrexStream *fixed_rx_flow_stat_stream = stream->clone(true);
-    get_stateless_obj()->m_rx_flow_stat.start_stream(fixed_rx_flow_stat_stream);
+
+    get_stateless_obj()->m_rx_flow_stat.start_stream(fixed_rx_flow_stat_stream, fixed_rx_flow_stat_stream->m_rx_check.m_hw_id); //???? check for errors
 
     /* can this stream be split to many cores ? */
     if (!stream->is_splitable(dp_core_count)) {
diff --git a/src/stateless/dp/trex_stateless_dp_core.cpp b/src/stateless/dp/trex_stateless_dp_core.cpp
index 549f923f..0f578b99 100644
--- a/src/stateless/dp/trex_stateless_dp_core.cpp
+++ b/src/stateless/dp/trex_stateless_dp_core.cpp
@@ -565,7 +565,6 @@ void
 TrexStatelessDpCore::add_stream(TrexStatelessDpPerPort * lp_port,
                                 TrexStream * stream,
                                 TrexStreamsCompiledObj *comp) {
-
     CGenNodeStateless *node = m_core->create_node_sl();
 
     /* add periodic */
@@ -581,7 +580,6 @@ TrexStatelessDpCore::add_stream(TrexStatelessDpPerPort * lp_port,
 
     node->m_next_stream=0; /* will be fixed later */
 
-
     if ( stream->m_self_start ){
         /* if self start it is in active mode */
         node->m_state =CGenNodeStateless::ss_ACTIVE;
@@ -597,7 +595,12 @@ TrexStatelessDpCore::add_stream(TrexStatelessDpPerPort * lp_port,
     node->m_src_port =0;
     node->m_original_packet_data_prefix = 0;
 
-
+    if (stream->m_rx_check.m_enabled) {
+        node->set_stat_needed();
+        uint8_t hw_id = stream->m_rx_check.m_hw_id;
+        assert (hw_id < MAX_FLOW_STATS);
+        node->set_stat_hw_id(hw_id);
+    }
 
     /* set socket id */
     node->set_socket_id(m_core->m_node_gen.m_socket_id);
diff --git a/src/stateless/dp/trex_stream_node.h b/src/stateless/dp/trex_stream_node.h
index b366a770..104e4d3b 100644
--- a/src/stateless/dp/trex_stream_node.h
+++ b/src/stateless/dp/trex_stream_node.h
@@ -62,7 +62,8 @@ public:
 
         SL_NODE_CONST_MBUF                 =4,
 
-        SL_NODE_VAR_PKT_SIZE               =8
+        SL_NODE_VAR_PKT_SIZE               =8,
+        SL_NODE_STATS_NEEDED               = 0x10
 
     };
 
@@ -82,7 +83,8 @@ private:
 
     double              m_next_time_offset; /* in sec */
     uint16_t            m_action_counter;
-    uint16_t            m_pad11;
+    uint8_t             m_stat_hw_id; // hw id used to count rx and tx stats
+    uint8_t             m_pad11;
     uint32_t            m_pad12;
 
     stream_state_t      m_state;
@@ -269,6 +271,22 @@ public:
         return ( m_socket_id );
     }
 
+    void set_stat_hw_id(uint16_t hw_id) {
+        m_stat_hw_id = hw_id;
+    }
+
+    socket_id_t get_stat_hw_id() {
+        return ( m_stat_hw_id );
+    }
+
+    inline void set_stat_needed() {
+        m_flags |= SL_NODE_STATS_NEEDED;
+    }
+
+    inline bool is_stat_needed() {
+        return ((m_flags & SL_NODE_STATS_NEEDED) != 0);
+    }
+
     inline void set_mbuf_cache_dir(pkt_dir_t  dir){
         if (dir) {
             m_flags |=NODE_FLAGS_DIR;
diff --git a/src/trex_defs.h b/src/trex_defs.h
index bb8510fa..ace3618d 100644
--- a/src/trex_defs.h
+++ b/src/trex_defs.h
@@ -17,7 +17,8 @@ limitations under the License.
 #define __TREX_DEFS_H__
 
 #define TREX_MAX_PORTS 12
-#define TREX_FDIR_STAT_SIZE 128
+
+#define MAX_FLOW_STATS 128
 
 #ifndef UINT8_MAX
     #define UINT8_MAX 255