Additions and fixes to per flow stats

3 files changed, 114 insertions, 43 deletions

diff --git a/src/flow_stat.cpp b/src/flow_stat.cpp
index ef32284b..266acb3f 100644
--- a/src/flow_stat.cpp
+++ b/src/flow_stat.cpp
@@ -23,7 +23,7 @@
 #include <iostream>
 #include <assert.h>
 #include <os_time.h>
-#include <internal_api/trex_platform_api.h>
+#include "internal_api/trex_platform_api.h"
 #include "trex_stateless.h"
 #include "trex_stream.h"
 #include "flow_stat_parser.h"
@@ -61,9 +61,14 @@ CFlowStatUserIdInfo::CFlowStatUserIdInfo(uint8_t proto) {
     m_proto = proto;
     m_ref_count = 1;
     m_trans_ref_count = 0;
+    m_was_sent = false;
+    for (int i = 0; i < TREX_MAX_PORTS; i++) {
+        m_rx_changed[i] = false;
+        m_tx_changed[i] = false;
+    }
 }
 
-std::ostream& operator<<(std::ostream& os, const class CFlowStatUserIdInfo& cf) {
+std::ostream& operator<<(std::ostream& os, const CFlowStatUserIdInfo& cf) {
     os << "hw_id:" << cf.m_hw_id << " proto:" << (uint16_t) cf.m_proto << " ref("
        << (uint16_t) cf.m_ref_count << "," << (uint16_t) cf.m_trans_ref_count << ")";
     os << " rx count (";
@@ -138,7 +143,7 @@ std::ostream& operator<<(std::ostream& os, const CFlowStatUserIdMap& cf) {
 }
 
 uint16_t CFlowStatUserIdMap::get_hw_id(uint32_t user_id) {
-    class CFlowStatUserIdInfo *cf = find_user_id(user_id);
+    CFlowStatUserIdInfo *cf = find_user_id(user_id);
 
     if (cf == NULL) {
         return FREE_HW_ID;
@@ -147,7 +152,7 @@ uint16_t CFlowStatUserIdMap::get_hw_id(uint32_t user_id) {
     }
 }
 
-class CFlowStatUserIdInfo *
+CFlowStatUserIdInfo *
 CFlowStatUserIdMap::find_user_id(uint32_t user_id) {
     flow_stat_user_id_map_it_t it = m_map.find(user_id);
 
@@ -158,17 +163,17 @@ CFlowStatUserIdMap::find_user_id(uint32_t user_id) {
     }
 }
 
-class CFlowStatUserIdInfo *
+CFlowStatUserIdInfo *
 CFlowStatUserIdMap::add_user_id(uint32_t user_id, uint8_t proto) {
 #ifdef __DEBUG_FUNC_ENTRY__
     std::cout << __METHOD_NAME__ << " user id:" << user_id << " proto:" << (uint16_t)proto
               << std::endl;
 #endif
 
-    class CFlowStatUserIdInfo *new_id = new CFlowStatUserIdInfo(proto);
+    CFlowStatUserIdInfo *new_id = new CFlowStatUserIdInfo(proto);
     if (new_id != NULL) {
         std::pair<flow_stat_user_id_map_it_t, bool> ret;
-        ret = m_map.insert(std::pair<uint32_t, class CFlowStatUserIdInfo *>(user_id, new_id));
+        ret = m_map.insert(std::pair<uint32_t, CFlowStatUserIdInfo *>(user_id, new_id));
         if (ret.second == false) {
             printf("%s Error: Trying to add user id %d which already exist\n", __func__, user_id);
             delete new_id;
@@ -186,7 +191,7 @@ int CFlowStatUserIdMap::add_stream(uint32_t user_id, uint8_t proto) {
               << std::endl;
 #endif
 
-    class CFlowStatUserIdInfo *c_user_id;
+    CFlowStatUserIdInfo *c_user_id;
 
     c_user_id = find_user_id(user_id);
     if (! c_user_id) {
@@ -204,7 +209,7 @@ int CFlowStatUserIdMap::del_stream(uint32_t user_id) {
     std::cout << __METHOD_NAME__ << " user id:" << user_id << std::endl;
 #endif
 
-    class CFlowStatUserIdInfo *c_user_id;
+    CFlowStatUserIdInfo *c_user_id;
 
     c_user_id = find_user_id(user_id);
     if (! c_user_id) {
@@ -225,7 +230,7 @@ int CFlowStatUserIdMap::start_stream(uint32_t user_id, uint16_t hw_id) {
     std::cout << __METHOD_NAME__ << " user id:" << user_id << " hw_id:" << hw_id << std::endl;
 #endif
 
-    class CFlowStatUserIdInfo *c_user_id;
+    CFlowStatUserIdInfo *c_user_id;
 
     c_user_id = find_user_id(user_id);
     if (! c_user_id) {
@@ -235,7 +240,7 @@ int CFlowStatUserIdMap::start_stream(uint32_t user_id, uint16_t hw_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 %ld\n"
+        fprintf(stderr, "%s Error: Trying to associate hw id %d to user_id %d but it is already associate to %u\n"
                 , __func__, hw_id, user_id, c_user_id->get_hw_id());
         return -1;
     }
@@ -250,7 +255,7 @@ int CFlowStatUserIdMap::start_stream(uint32_t user_id) {
     std::cout << __METHOD_NAME__ << " user id:" << user_id << std::endl;
 #endif
 
-    class CFlowStatUserIdInfo *c_user_id;
+    CFlowStatUserIdInfo *c_user_id;
 
     c_user_id = find_user_id(user_id);
     if (! c_user_id) {
@@ -271,7 +276,7 @@ int CFlowStatUserIdMap::stop_stream(uint32_t user_id) {
     std::cout << __METHOD_NAME__ << " user id:" << user_id << std::endl;
 #endif
 
-    class CFlowStatUserIdInfo *c_user_id;
+    CFlowStatUserIdInfo *c_user_id;
 
     c_user_id = find_user_id(user_id);
     if (! c_user_id) {
@@ -284,7 +289,7 @@ int CFlowStatUserIdMap::stop_stream(uint32_t user_id) {
 }
 
 bool CFlowStatUserIdMap::is_started(uint32_t user_id) {
-    class CFlowStatUserIdInfo *c_user_id;
+    CFlowStatUserIdInfo *c_user_id;
 
     c_user_id = find_user_id(user_id);
     if (! c_user_id) {
@@ -295,7 +300,7 @@ bool CFlowStatUserIdMap::is_started(uint32_t user_id) {
 }
 
 uint8_t CFlowStatUserIdMap::l4_proto(uint32_t user_id) {
-    class CFlowStatUserIdInfo *c_user_id;
+    CFlowStatUserIdInfo *c_user_id;
 
     c_user_id = find_user_id(user_id);
     if (! c_user_id) {
@@ -310,7 +315,7 @@ uint16_t CFlowStatUserIdMap::unmap(uint32_t user_id) {
     std::cout << __METHOD_NAME__ << " user id:" << user_id << std::endl;
 #endif
 
-    class CFlowStatUserIdInfo *c_user_id;
+    CFlowStatUserIdInfo *c_user_id;
 
     c_user_id = find_user_id(user_id);
     if (! c_user_id) {
@@ -583,15 +588,21 @@ int CFlowStatRuleMgr::stop_stream(const TrexStream * stream) {
             return -1;
         } else {
             // 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));
+            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 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_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);
+                if (p_user_id->get_rx_counter(port) != rx_counter) {
+                    p_user_id->set_rx_counter(port, rx_counter);
+                    p_user_id->set_need_to_send_rx(port);
+                }
+                if (p_user_id->get_tx_counter(port) != tx_counter) {
+                    p_user_id->set_tx_counter(port, tx_counter);
+                    p_user_id->set_need_to_send_tx(port);
+                }
             }
             m_user_id_map.unmap(stream->m_rx_check.m_pg_id);
             m_hw_id_map.unmap(hw_id);
@@ -611,37 +622,48 @@ int CFlowStatRuleMgr::get_active_pgids(flow_stat_active_t &result) {
 }
 
 // return false if no counters changed since last run. true otherwise
-bool CFlowStatRuleMgr::dump_json(std::string & json) {
+bool CFlowStatRuleMgr::dump_json(std::string & json, bool force_sync) {
     uint64_t rx_stats[MAX_FLOW_STATS];
     tx_per_flow_t tx_stats[MAX_FLOW_STATS];
     Json::FastWriter writer;
     Json::Value root;
 
-    if (m_user_id_map.is_empty()) {
-        return false;
-    }
     root["name"] = "flow_stats";
     root["type"] = 0;
     Json::Value &data_section = root["data"];
+    data_section["timestamp"] = Json::Value::UInt64(os_get_hr_tick_64());
+
+    if (m_user_id_map.is_empty()) {
+        if (force_sync) {
+            json = writer.write(root);
+            return true;
+        } else
+            return false;
+    }
 
     // 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++) {
         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));
+                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]);
+                    if (p_user_id->get_rx_counter(port) != rx_stats[i]) {
+                        p_user_id->set_rx_counter(port, rx_stats[i]);
+                        p_user_id->set_need_to_send_rx(port);
+                    }
                 } 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));
+                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);
+                    if (p_user_id->get_tx_counter(port) != tx_pkts) {
+                        p_user_id->set_tx_counter(port, tx_pkts);
+                        p_user_id->set_need_to_send_tx(port);
+                    }
                 } else {
                     std::cerr <<  __METHOD_NAME__ << i << ":Could not count tx " << tx_pkts << " because no mapping was found" << std::endl;
                 }
@@ -656,13 +678,18 @@ bool CFlowStatRuleMgr::dump_json(std::string & json) {
         uint32_t user_id = it->first;
         std::string str_user_id = static_cast<std::ostringstream*>( &(std::ostringstream()
                                                                       << user_id) )->str();
+        if (! user_id_info->was_sent()) {
+            data_section[str_user_id]["first_time"] = true;
+            user_id_info->set_was_sent(true);
+        }
         for (uint8_t port = 0; port < m_num_ports; port++) {
             std::string str_port = static_cast<std::ostringstream*>( &(std::ostringstream() << int(port) ) )->str();
-
-            if (user_id_info->get_rx_counter(port) != 0) {
+            if (user_id_info->need_to_send_rx(port) || force_sync) {
+                user_id_info->set_no_need_to_send_rx(port);
                 data_section[str_user_id]["rx_pkts"][str_port] = Json::Value::UInt64(user_id_info->get_rx_counter(port));
             }
-            if (user_id_info->get_tx_counter(port).get_pkts() != 0) {
+            if (user_id_info->need_to_send_tx(port) || force_sync) {
+                user_id_info->set_no_need_to_send_tx(port);
                 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());
             }
diff --git a/src/flow_stat.h b/src/flow_stat.h
index 6f7671fd..6966b116 100644
--- a/src/flow_stat.h
+++ b/src/flow_stat.h
@@ -83,7 +83,13 @@ class tx_per_flow_t_ {
         return *this;
     }
 
-    friend std::ostream& operator<<(std::ostream& os, const class tx_per_flow_t_ &t) {
+    inline bool operator!= (const tx_per_flow_t_ &t_in) {
+        if ((m_bytes != t_in.m_bytes) || (m_pkts != t_in.m_pkts))
+            return true;
+        return false;
+    }
+
+    friend std::ostream& operator<<(std::ostream& os, const tx_per_flow_t_ &t) {
         os  << "p:" << t.m_pkts << " b:" << t.m_bytes;
         return os;
     }
@@ -107,7 +113,7 @@ class CFlowStatUserIdInfo {
     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;}
+    uint16_t get_hw_id() {return m_hw_id;}
     void reset_hw_id();
     bool is_hw_id() {return (m_hw_id != UINT16_MAX);}
     uint64_t get_proto() {return m_proto;}
@@ -117,8 +123,18 @@ class CFlowStatUserIdInfo {
     void add_started_stream() {m_trans_ref_count++;}
     int stop_started_stream() {m_trans_ref_count--; return m_trans_ref_count;}
     bool is_started() {return (m_trans_ref_count != 0);}
+    bool need_to_send_rx(uint8_t port) {return m_rx_changed[port];}
+    bool need_to_send_tx(uint8_t port) {return m_tx_changed[port];}
+    void set_no_need_to_send_rx(uint8_t port) {m_rx_changed[port] = false;}
+    void set_no_need_to_send_tx(uint8_t port) {m_tx_changed[port] = false;}
+    void set_need_to_send_rx(uint8_t port) {m_rx_changed[port] = true;}
+    void set_need_to_send_tx(uint8_t port) {m_tx_changed[port] = true;}
+    bool was_sent() {return m_was_sent == true;}
+    void set_was_sent(bool val) {m_was_sent = val;}
 
  private:
+    bool m_rx_changed[TREX_MAX_PORTS]; // Which RX counters changed since we last published
+    bool m_tx_changed[TREX_MAX_PORTS]; // Which TX counters changed since we last published
     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];
@@ -129,6 +145,7 @@ class CFlowStatUserIdInfo {
     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
     uint8_t m_trans_ref_count;  // How many streams with this ref count currently transmit
+    bool m_was_sent; // Did we send this info to clients once?
 };
 
 typedef std::map<uint32_t, class CFlowStatUserIdInfo *> flow_stat_user_id_map_t;
@@ -140,8 +157,8 @@ class CFlowStatUserIdMap {
     friend std::ostream& operator<<(std::ostream& os, const CFlowStatUserIdMap& cf);
     bool is_empty() {return (m_map.empty() == true);};
     uint16_t get_hw_id(uint32_t user_id);
-    class CFlowStatUserIdInfo * find_user_id(uint32_t user_id);
-    class CFlowStatUserIdInfo * add_user_id(uint32_t user_id, uint8_t proto);
+    CFlowStatUserIdInfo * find_user_id(uint32_t user_id);
+    CFlowStatUserIdInfo * add_user_id(uint32_t user_id, uint8_t proto);
     int add_stream(uint32_t user_id, uint8_t proto);
     int del_stream(uint32_t user_id);
     int start_stream(uint32_t user_id, uint16_t hw_id);
@@ -185,15 +202,15 @@ class CFlowStatRuleMgr {
     int start_stream(TrexStream * stream, uint16_t &ret_hw_id);
     int stop_stream(const TrexStream * stream);
     int get_active_pgids(flow_stat_active_t &result);
-    bool dump_json(std::string & json);
+    bool dump_json(std::string & json, bool force_sync);
 
  private:
     int compile_stream(const TrexStream * stream, Cxl710Parser &parser);
     int add_hw_rule(uint16_t hw_id, uint8_t proto);
 
  private:
-    class CFlowStatHwIdMap m_hw_id_map; // map hw ids to user ids
-    class CFlowStatUserIdMap m_user_id_map; // map user ids to hw ids
+    CFlowStatHwIdMap m_hw_id_map; // map hw ids to user ids
+    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
diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp
index 0d40215a..12670969 100644
--- a/src/main_dpdk.cpp
+++ b/src/main_dpdk.cpp
@@ -2640,7 +2640,7 @@ public:
     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);
+    tx_per_flow_t 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);
@@ -3307,8 +3307,32 @@ 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) {
+// read stats. Return read value, and clear.
+tx_per_flow_t CGlobalTRex::clear_flow_tx_stats(uint8_t port, uint16_t index) {
+    uint8_t port0;
+    CFlowGenListPerThread * lpt;
+    tx_per_flow_t ret;
+
+    m_stats.m_port[port].m_tx_per_flow[index].clear();
+
+    for (int i=0; i < get_cores_tx(); i++) {
+        lpt = m_fl.m_threads_info[i];
+        port0 = lpt->getDualPortId() * 2;
+        if (port == port0) {
+            m_stats.m_port[port0].m_tx_per_flow[index] +=
+                lpt->m_node_gen.m_v_if->m_stats[0].m_tx_per_flow[index];
+        } else if (port == port0 + 1) {
+            m_stats.m_port[port0 + 1].m_tx_per_flow[index] +=
+                lpt->m_node_gen.m_v_if->m_stats[1].m_tx_per_flow[index];
+        }
+    }
+
+    ret = m_stats.m_port[port].m_tx_per_flow[index] - m_stats.m_port[port].m_prev_tx_per_flow[index];
+
+    // Since we return diff from prev, following "clears" the stats.
     m_stats.m_port[port].m_prev_tx_per_flow[index] = m_stats.m_port[port].m_tx_per_flow[index];
+
+    return ret;
 }
 
 void CGlobalTRex::get_stats(CGlobalStats & stats){
@@ -3568,7 +3592,7 @@ CGlobalTRex::publish_async_data(bool sync_now) {
     m_zmq_publisher.publish_json(json);
 
     if (get_is_stateless()) {
-        if (m_trex_stateless->m_rx_flow_stat.dump_json(json))
+        if (m_trex_stateless->m_rx_flow_stat.dump_json(json, sync_now))
             m_zmq_publisher.publish_json(json);
     }
 }
@@ -3930,10 +3954,9 @@ int CPhyEthIF::get_flow_stats(uint64_t *rx_stats, tx_per_flow_t *tx_stats, int m
                 rx_stats[i - min] = m_stats.m_rx_per_flow[i] + diff_stats[i];
             }
             if (tx_stats != NULL) {
-                tx_stats[i - min] = g_trex.get_flow_tx_stats(m_port_id, i);
+                tx_stats[i - min] = g_trex.clear_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) {
@@ -4911,6 +4934,10 @@ 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);
+// get rx stats on _if, between min and max
+// prev_stats should be the previous values read from the hardware.
+//            Getting changed to be equal to current HW values.
+// stats return the diff between prev_stats and current hw values
 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();