Regression tests working. Still missing python API to parse latency json

13 files changed, 251 insertions, 99 deletions

diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_stats.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_stats.py
index eace5cf2..e30da00e 100644
--- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_stats.py
+++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_stats.py
@@ -942,7 +942,7 @@ class CRxStats(CTRexStats):
 
     def calculate_bw_for_pg (self, pg_current, pg_prev = None, diff_sec = 0.0):
         # no previous values
-        if (pg_prev == None) or not (diff_sec > 0):
+        if (not pg_prev) or not (diff_sec > 0):
             pg_current['tx_pps']        = {}
             pg_current['tx_bps']        = {}
             pg_current['tx_bps_L1']     = {}
diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py
index a7fd3026..6264c17f 100755
--- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py
+++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py
@@ -210,25 +210,11 @@ STLStreamDstMAC_CFG_FILE=0
 STLStreamDstMAC_PKT     =1
 STLStreamDstMAC_ARP     =2
 
-# RX stats class
-class STLFlowStats(object):
-    """ Define per stream stats  
-
-        .. code-block:: python
-            :caption: STLFlowStats Example
-
-            flow_stats = STLFlowStats(pg_id = 7)
-
-    """
-
+class STLFlowStatsInterface(object):
     def __init__ (self, pg_id):
         self.fields = {}
-
         self.fields['enabled']         = True
         self.fields['stream_id']       = pg_id
-        self.fields['seq_enabled']     = False
-        self.fields['latency_enabled'] = False
-
 
     def to_json (self):
         """ Dump as json"""
@@ -238,6 +224,37 @@ class STLFlowStats(object):
     def defaults ():
         return {'enabled' : False}
 
+
+class STLFlowStats(STLFlowStatsInterface):
+    """ Define per stream basic stats
+
+    .. code-block:: python
+    :caption: STLFlowStats Example
+
+    flow_stats = STLFlowStats(pg_id = 7)
+
+    """
+
+    def __init__(self, pg_id):
+        super(STLFlowStats, self).__init__(pg_id)
+        self.fields['rule_type'] = 'stats'
+
+
+class STLFlowLatencyStats(STLFlowStatsInterface):
+    """ Define per stream basic stats + latency, jitter, packet reorder/loss
+
+    .. code-block:: python
+    :caption: STLFlowLatencyStats Example
+
+    flow_stats = STLFlowLatencyStats(pg_id = 7)
+
+    """
+
+    def __init__(self, pg_id):
+        super(STLFlowLatencyStats, self).__init__(pg_id)
+        self.fields['rule_type'] = 'latency'
+
+
 class STLStream(object):
     """ One stream object. Includes mode, Field Engine mode packet template and Rx stats  
 
@@ -320,6 +337,7 @@ class STLStream(object):
         # type checking
         validate_type('mode', mode, STLTXMode)
         validate_type('packet', packet, (type(None), CTrexPktBuilderInterface))
+        validate_type('flow_stats', flow_stats, (type(None), STLFlowStatsInterface))
         validate_type('enabled', enabled, bool)
         validate_type('self_start', self_start, bool)
         validate_type('isg', isg, (int, float))
diff --git a/src/bp_sim.h b/src/bp_sim.h
index 97e14158..bb7dd928 100755
--- a/src/bp_sim.h
+++ b/src/bp_sim.h
@@ -274,7 +274,7 @@ public:
     uint64_t   m_tx_queue_full;
     uint64_t   m_tx_alloc_error;
     tx_per_flow_t m_tx_per_flow[MAX_FLOW_STATS + MAX_FLOW_STATS_PAYLOAD];
-    uint64_t   m_seq_num[MAX_FLOW_STATS_PAYLOAD]; // seq num to put in packet for payload rules
+    uint32_t   m_seq_num[MAX_FLOW_STATS_PAYLOAD]; // seq num to put in packet for payload rules
     CPerTxthreadTemplateInfo m_template;
 
 public:
@@ -297,6 +297,9 @@ public:
        m_tx_alloc_error=0;
        m_tx_queue_full=0;
        m_template.Clear();
+       for (int i = 0; i < MAX_FLOW_STATS_PAYLOAD; i++) {
+           m_seq_num[i] = UINT32_MAX - 1; // catch wrap around issues early
+       }
     }
 
     inline void Dump(FILE *fd);
diff --git a/src/flow_stat.cpp b/src/flow_stat.cpp
index 52d8129c..d63dc778 100644
--- a/src/flow_stat.cpp
+++ b/src/flow_stat.cpp
@@ -164,11 +164,18 @@ void CFlowStatUserIdInfo::reset_hw_id() {
 
 /************** class CFlowStatUserIdInfoPayload ***************/
 void CFlowStatUserIdInfoPayload::add_stream(uint8_t proto) {
-    //??? add unit test
     throw TrexFStatEx("Can't have two streams with same packet group id for payload rules"
                       , TrexException::T_FLOW_STAT_DUP_PG_ID);
 }
 
+void CFlowStatUserIdInfoPayload::reset_hw_id() {
+    CFlowStatUserIdInfo::reset_hw_id();
+
+    m_seq_error_base += m_rfc2544_info.m_seq_error;
+    m_out_of_order_base += m_rfc2544_info.m_out_of_order;
+    m_rfc2544_info.m_seq_error = 0;
+    m_rfc2544_info.m_out_of_order = 0;
+}
 
 /************** class CFlowStatUserIdMap ***************/
 CFlowStatUserIdMap::CFlowStatUserIdMap() {
@@ -370,7 +377,7 @@ uint16_t CFlowStatUserIdMap::unmap(uint32_t user_id) {
         return UINT16_MAX;
     }
     uint16_t old_hw_id = c_user_id->get_hw_id();
-    c_user_id->reset_hw_id();//??? need to call reset of CFlowStatUserIdInfoPayload if needed
+    c_user_id->reset_hw_id();
 
     return old_hw_id;
 }
@@ -535,8 +542,6 @@ int CFlowStatRuleMgr::add_stream(TrexStream * stream) {
         create();
     }
 
-    //??? put back    assert(stream->m_rx_check.m_hw_id == HW_ID_INIT);
-
     TrexPlatformApi::driver_stat_cap_e rule_type = (TrexPlatformApi::driver_stat_cap_e)stream->m_rx_check.m_rule_type;
 
     if ((m_cap & rule_type) == 0) {
@@ -893,28 +898,36 @@ void CFlowStatRuleMgr::send_start_stop_msg_to_rx(bool is_start) {
 }
 
 // return false if no counters changed since last run. true otherwise
-bool CFlowStatRuleMgr::dump_json(std::string & json, bool baseline) {
+// s_json - flow statistics json
+// l_json - latency data json
+// baseline - If true, send flow statistics fields even if they were not changed since last run
+bool CFlowStatRuleMgr::dump_json(std::string & s_json, std::string & l_json, bool baseline) {
     rx_per_flow_t rx_stats[MAX_FLOW_STATS];
     rx_per_flow_t rx_stats_payload[MAX_FLOW_STATS];
     tx_per_flow_t tx_stats[MAX_FLOW_STATS];
     tx_per_flow_t tx_stats_payload[MAX_FLOW_STATS_PAYLOAD];
     rfc2544_info_t rfc2544_info[MAX_FLOW_STATS_PAYLOAD];
     Json::FastWriter writer;
-    Json::Value root;
+    Json::Value s_root;
+    Json::Value l_root;
 
-    root["name"] = "flow_stats";
-    root["type"] = 0;
+    s_root["name"] = "flow_stats";
+    s_root["type"] = 0;
+    l_root["name"] = "latency_stats";
+    l_root["type"] = 0;
 
     if (baseline) {
-        root["baseline"] = true;
+        s_root["baseline"] = true;
     }
 
-    Json::Value &data_section = root["data"];
-    data_section["ts"]["value"] = Json::Value::UInt64(os_get_hr_tick_64());
-    data_section["ts"]["freq"] = Json::Value::UInt64(os_get_hr_freq());
+    Json::Value &s_data_section = s_root["data"];
+    Json::Value &l_data_section = l_root["data"];
+    s_data_section["ts"]["value"] = Json::Value::UInt64(os_get_hr_tick_64());
+    s_data_section["ts"]["freq"] = Json::Value::UInt64(os_get_hr_freq());
 
     if (m_user_id_map.is_empty()) {
-        json = writer.write(root);
+        s_json = writer.write(s_root);
+        l_json = writer.write(l_root);
         return true;
     }
 
@@ -992,7 +1005,7 @@ bool CFlowStatRuleMgr::dump_json(std::string & json, bool baseline) {
         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;
+            s_data_section[str_user_id]["first_time"] = true;
             user_id_info->set_was_sent(true);
             send_empty = false;
         }
@@ -1000,18 +1013,22 @@ bool CFlowStatRuleMgr::dump_json(std::string & json, bool baseline) {
             std::string str_port = static_cast<std::ostringstream*>( &(std::ostringstream() << int(port) ) )->str();
             if (user_id_info->need_to_send_rx(port) || baseline) {
                 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_cntr(port).get_pkts());
+                s_data_section[str_user_id]["rx_pkts"][str_port] = Json::Value::UInt64(user_id_info->get_rx_cntr(port).get_pkts());
                 if (m_cap & TrexPlatformApi::IF_STAT_RX_BYTES_COUNT)
-                    //???put back                    data_section[str_user_id]["rx_bytes"][str_port] = Json::Value::UInt64(user_id_info->get_rx_cntr(port).get_bytes());
+                    s_data_section[str_user_id]["rx_bytes"][str_port] = Json::Value::UInt64(user_id_info->get_rx_cntr(port).get_bytes());
                 send_empty = false;
             }
             if (user_id_info->need_to_send_tx(port) || baseline) {
                 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_cntr(port).get_pkts());
-                //??? pub back                data_section[str_user_id]["tx_bytes"][str_port] = Json::Value::UInt64(user_id_info->get_tx_cntr(port).get_bytes());
+                s_data_section[str_user_id]["tx_pkts"][str_port] = Json::Value::UInt64(user_id_info->get_tx_cntr(port).get_pkts());
+                s_data_section[str_user_id]["tx_bytes"][str_port] = Json::Value::UInt64(user_id_info->get_tx_cntr(port).get_bytes());
                 send_empty = false;
             }
         }
+        if (send_empty) {
+            s_data_section[str_user_id] = Json::objectValue;
+        }
+
         if (user_id_info->rfc2544_support()) {
             CFlowStatUserIdInfoPayload *user_id_info_p = (CFlowStatUserIdInfoPayload *)user_id_info;
             // payload object. Send also latency, jitter...
@@ -1022,33 +1039,34 @@ bool CFlowStatRuleMgr::dump_json(std::string & json, bool baseline) {
                 rfc2544_info[hw_id].get_latency_json(json);
                 user_id_info_p->set_seq_err_cnt(rfc2544_info[hw_id].get_seq_err_cnt());
                 user_id_info_p->set_ooo_cnt(rfc2544_info[hw_id].get_ooo_cnt());
-                data_section[str_user_id]["rfc2544"][""] = json;
-                data_section[str_user_id]["rfc2544"]["jitter"] = rfc2544_info[hw_id].get_jitter();
+                l_data_section[str_user_id]["latency"] = json;
+                l_data_section[str_user_id]["jitter"] = rfc2544_info[hw_id].get_jitter();
             } else {
                 // Not mapped to hw_id. Get saved info.
                 user_id_info_p->get_latency_json(json);
-                data_section[str_user_id]["rfc2544"]["latency"] = json;
-                data_section[str_user_id]["rfc2544"]["jitter"] = user_id_info_p->get_jitter();
+                l_data_section[str_user_id]["latency"] = json;
+                l_data_section[str_user_id]["jitter"] = user_id_info_p->get_jitter();
             }
-            data_section[str_user_id]["rfc2544"]["err_cntrs"]["lost"]
+            ///????? add last 10 samples
+            l_data_section[str_user_id]["err_cntrs"]["lost"]
                 = Json::Value::UInt64(user_id_info_p->get_seq_err_cnt());
-            data_section[str_user_id]["rfc2544"]["err_cntrs"]["out_of_order"]
+            l_data_section[str_user_id]["err_cntrs"]["out_of_order"]
                 = Json::Value::UInt64(user_id_info_p->get_ooo_cnt());
 
             //??? temp - remove
-            data_section[str_user_id]["tx_bytes"]["0"]
+#if 0
+            s_data_section[str_user_id]["tx_bytes"]["0"]
                 = Json::Value::UInt64(user_id_info_p->get_seq_err_cnt());
-            data_section[str_user_id]["tx_bytes"]["1"] = 0;
-            data_section[str_user_id]["rx_bytes"]["0"]
+            s_data_section[str_user_id]["tx_bytes"]["1"] = 0;
+            s_data_section[str_user_id]["rx_bytes"]["0"]
                 = Json::Value::UInt64(user_id_info_p->get_ooo_cnt());
-            data_section[str_user_id]["rx_bytes"]["1"] = 0;
-        }
-        if (send_empty) {
-            data_section[str_user_id] = Json::objectValue;
+            s_data_section[str_user_id]["rx_bytes"]["1"] = 0;
+#endif
         }
     }
 
-    json = writer.write(root);
+    s_json = writer.write(s_root);
+    l_json = writer.write(l_root);
     // We always want to publish, even only the timestamp.
     return true;
 }
diff --git a/src/flow_stat.h b/src/flow_stat.h
index 36a4bad1..5bfab44a 100644
--- a/src/flow_stat.h
+++ b/src/flow_stat.h
@@ -43,10 +43,10 @@ typedef std::map<uint32_t, uint16_t>::iterator flow_stat_map_it_t;
 class CRxCoreStateless;
 
 struct flow_stat_payload_header {
-    uint64_t time_stamp;
-    uint16_t hw_id;
     uint16_t magic;
+    uint16_t hw_id;
     uint32_t seq;
+    uint64_t time_stamp;
 };
 
 
@@ -246,7 +246,8 @@ class CFlowStatUserIdInfo {
     bool rfc2544_support() {return m_rfc2544_support;}
 
  protected:
-        bool m_rfc2544_support;
+    bool m_rfc2544_support;
+    uint16_t m_hw_id;     // Associated hw id. UINT16_MAX if no associated hw id.
 
  private:
     bool m_rx_changed[TREX_MAX_PORTS]; // Which RX counters changed since we last published
@@ -257,7 +258,6 @@ class CFlowStatUserIdInfo {
     tx_per_flow_t m_tx_cntr[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.
     tx_per_flow_t m_tx_cntr_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 user id exists
     uint8_t m_trans_ref_count;  // How many streams with this user id currently transmit
@@ -309,13 +309,7 @@ class CFlowStatUserIdInfoPayload : public CFlowStatUserIdInfo {
         return m_rfc2544_info.m_out_of_order + m_out_of_order_base;
     }
 
-    inline void reset_hw_id() {
-        m_seq_error_base += m_rfc2544_info.m_seq_error;
-        m_out_of_order_base += m_rfc2544_info.m_out_of_order;
-        m_rfc2544_info.m_seq_error = 0;
-        m_rfc2544_info.m_out_of_order = 0;
-    }
-
+    inline void reset_hw_id();    
  private:
     rfc2544_info_t m_rfc2544_info;
     uint64_t m_seq_error_base;
@@ -377,7 +371,7 @@ class CFlowStatRuleMgr {
     int start_stream(TrexStream * stream);
     int stop_stream(TrexStream * stream);
     int get_active_pgids(flow_stat_active_t &result);
-    bool dump_json(std::string & json, bool baseline);
+    bool dump_json(std::string & s_json, std::string & l_json, bool baseline);
 
  private:
     void create();
diff --git a/src/gtest/trex_stateless_gtest.cpp b/src/gtest/trex_stateless_gtest.cpp
index 18a09a2c..d5e36944 100644
--- a/src/gtest/trex_stateless_gtest.cpp
+++ b/src/gtest/trex_stateless_gtest.cpp
@@ -3729,6 +3729,7 @@ TEST_F(flow_stat, add_del_stream) {
 
     TrexStream stream(TrexStream::stSINGLE_BURST, 0, 0);
     TrexStream stream2(TrexStream::stSINGLE_BURST, 0, 0);
+    TrexStream stream3(TrexStream::stSINGLE_BURST, 0, 0);
     
     stream.m_rx_check.m_enabled = true;
 
@@ -3762,17 +3763,29 @@ TEST_F(flow_stat, add_del_stream) {
     test_pkt[27] = IPPROTO_UDP;
     int ret = rule_mgr.add_stream(&stream);
     assert (ret == 0);
+
+    stream3.m_rx_check.m_enabled = true;
+    stream3.m_rx_check.m_rule_type = TrexPlatformApi::IF_STAT_PAYLOAD;
+    stream3.m_rx_check.m_pg_id = 5; // same as first stream
+    stream3.m_pkt.binary = (uint8_t *)test_pkt;
+    stream3.m_pkt.len = sizeof(test_pkt);
+    try {
+        ret = rule_mgr.add_stream(&stream3);
+    } catch (TrexFStatEx e) {
+        assert(e.type() == TrexException::T_FLOW_STAT_DUP_PG_ID);
+    }
+
     ret = rule_mgr.del_stream(&stream);
     assert (ret == 0);
 
+    stream2.m_rx_check.m_enabled = true;
     stream2.m_rx_check.m_rule_type = TrexPlatformApi::IF_STAT_IPV4_ID;
-    stream2.m_rx_check.m_pg_id = 5; // ??? same as first stream
+    stream2.m_rx_check.m_pg_id = 5; // same as first stream
     stream2.m_pkt.binary = (uint8_t *)test_pkt;
     stream2.m_pkt.len = sizeof(test_pkt);
     ret = rule_mgr.add_stream(&stream2);
     assert (ret == 0);
 
-
     ret = rule_mgr.del_stream(&stream2);
     assert (ret == 0);
     try {
@@ -3781,9 +3794,10 @@ TEST_F(flow_stat, add_del_stream) {
         assert(e.type() == TrexException::T_FLOW_STAT_DEL_NON_EXIST);
     } 
 
-    // do not want the constructor to try to free it
+    // do not want the destructor to try to free it
     stream.m_pkt.binary = NULL; 
     stream2.m_pkt.binary = NULL;
+    stream3.m_pkt.binary = NULL;
 }
 
 TEST_F(flow_stat, start_stop_stream) {
diff --git a/src/internal_api/trex_platform_api.h b/src/internal_api/trex_platform_api.h
index f60d1591..a52f9e60 100644
--- a/src/internal_api/trex_platform_api.h
+++ b/src/internal_api/trex_platform_api.h
@@ -234,6 +234,8 @@ public:
     virtual void publish_async_data_now(uint32_t key, bool baseline) const {
 
     }
+    int get_flow_stats(uint8_t port_id, void *stats, void *tx_stats, int min, int max, bool reset
+                       , TrexPlatformApi::driver_stat_cap_e type) const {return 0;};
     virtual int get_rfc2544_info(void *rfc2544_info, int min, int max, bool reset) const {return 0;};
     virtual int reset_hw_flow_stats(uint8_t port_id) const {return 0;};
     virtual void get_port_num(uint8_t &port_num) const {port_num = 2;};
diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp
index 8b33ead8..ccd1aa8d 100644
--- a/src/main_dpdk.cpp
+++ b/src/main_dpdk.cpp
@@ -1784,7 +1784,8 @@ protected:
 
 class CCoreEthIFStateless : public CCoreEthIF {
 public:
-    virtual int send_node_flow_stat(CGenNodeStateless * node_sl);
+    virtual int send_node_flow_stat(rte_mbuf *m, CGenNodeStateless * node_sl, CCorePerPort *  lp_port
+                                    , CVirtualIFPerSideStats  * lp_stats);
     virtual int send_node(CGenNode * node);
 protected:
     int handle_slow_path_node(CGenNode *node);
@@ -2005,9 +2006,89 @@ void CCoreEthIF::update_mac_addr(CGenNode * node,uint8_t *p){
     }
 }
 
-/// ??? need better implementation. Maybe implement as template of send_node.
+int CCoreEthIFStateless::send_node_flow_stat(rte_mbuf *m, CGenNodeStateless * node_sl, CCorePerPort *  lp_port
+                                             , CVirtualIFPerSideStats  * lp_stats) {
+    //??? remove
+# if 0
+    static int temp=1;
+    temp++;
+#endif
+
+    uint16_t hw_id = node_sl->get_stat_hw_id();
+    rte_mbuf *m_lat, *mi;
+
+    if (hw_id >= MAX_FLOW_STATS) {
+        // payload rule hw_ids are in the range right above ip id rules
+        uint16_t hw_id_payload = hw_id - MAX_FLOW_STATS;
+        if (hw_id_payload > max_stat_hw_id_seen_payload) {
+            max_stat_hw_id_seen_payload = hw_id_payload;
+        }
+        // alloc mbuf just for the latency header
+        m_lat = CGlobalInfo::pktmbuf_alloc( get_socket_id(), sizeof(struct flow_stat_payload_header));
+        if ( unlikely(m_lat == 0)) {
+            return -1;
+        }
+        char *p = rte_pktmbuf_append(m_lat, sizeof(struct flow_stat_payload_header));
+        struct flow_stat_payload_header *fsp_head = (struct flow_stat_payload_header *)p;
+        fsp_head->seq = lp_stats->m_seq_num[hw_id_payload];
+        fsp_head->time_stamp = os_get_hr_tick_64();
+        // ??? maybe following two lines can be done offline
+        fsp_head->hw_id = hw_id_payload;
+        fsp_head->magic = FLOW_STAT_PAYLOAD_MAGIC;
+
+        lp_stats->m_seq_num[hw_id_payload]++;
+
+        //??? remove
+#if 0
+        if (temp % 10 == 0) {
+            fsp_head->seq = lp_stats->m_seq_num[hw_id_payload]++;
+        }
+
+        if ((temp - 1) % 100 == 0) {
+            fsp_head->seq = lp_stats->m_seq_num[hw_id_payload] - 4;
+            //            lp_stats->m_seq_num[hw_id_payload]--;
+        }
+#endif
+
+
+
+        if (rte_pktmbuf_is_contiguous(m)) {
+            // We have only the const mbuf
+            mi = CGlobalInfo::pktmbuf_alloc_small(get_socket_id());
+            assert(mi);
+            rte_pktmbuf_attach(mi, m);
+            rte_pktmbuf_trim(mi, sizeof(struct flow_stat_payload_header));
+            utl_rte_pktmbuf_add_after2(mi, m_lat);
+        } else {
+            // Field engine (vm) case.
+            rte_pktmbuf_trim(m, sizeof(struct flow_stat_payload_header));
+            utl_rte_pktmbuf_add_last(m, m_lat);
+            mi = m;
+        }
+
+    } else {
+        // ip id rule
+        if (hw_id > max_stat_hw_id_seen) {
+            max_stat_hw_id_seen = hw_id;
+        }
+        mi = m;
+    }
+    tx_per_flow_t *lp_s = &lp_stats->m_tx_per_flow[hw_id];
+    lp_s->add_pkts(1);
+    lp_s->add_bytes(mi->pkt_len);
+
+    send_pkt(lp_port, mi, lp_stats);
+    return 0;
+}
+
+#if 0
+//??? remove
 // Maybe make it part of send_node somehow
 int CCoreEthIFStateless::send_node_flow_stat(CGenNodeStateless * node_sl) {
+    //??? remove
+    static int temp=1;
+    temp++;
+
     uint16_t hw_id = node_sl->get_stat_hw_id();
     tx_per_flow_t *lp_s;
     /* check that we have mbuf  */
@@ -2041,14 +2122,14 @@ int CCoreEthIFStateless::send_node_flow_stat(CGenNodeStateless * node_sl) {
         fsp_head->time_stamp = os_get_hr_tick_64();
         lp_stats->m_seq_num[hw_id_payload]++;
         // remove ???
-#if 0
-        if (temp % 11 == 0) {
+
+        if (temp % 10 == 0) {
             fsp_head->seq = lp_stats->m_seq_num[hw_id_payload]++;
         }
-
-        if ((temp -1) % 100 == 0) {
-            fsp_head->seq = lp_stats->m_seq_num[hw_id_payload] - 3;
-            lp_stats->m_seq_num[hw_id_payload]--;
+#if 1
+        if ((temp - 1) % 100 == 0) {
+            fsp_head->seq = lp_stats->m_seq_num[hw_id_payload] - 4;
+            //            lp_stats->m_seq_num[hw_id_payload]--;
         }
 #endif
     } else {
@@ -2064,9 +2145,9 @@ int CCoreEthIFStateless::send_node_flow_stat(CGenNodeStateless * node_sl) {
     send_pkt(lp_port,m,lp_stats);
     return 0;
 }
+#endif
 
 int CCoreEthIFStateless::send_node(CGenNode * no) {
-
     /* if a node is marked as slow path - single IF to redirect it to slow path */
     if (no->get_is_slow_path()) {
         return handle_slow_path_node(no);
@@ -2074,10 +2155,6 @@ int CCoreEthIFStateless::send_node(CGenNode * no) {
 
     CGenNodeStateless * node_sl=(CGenNodeStateless *) no;
 
-    if (unlikely(node_sl->is_stat_needed())) {
-        return send_node_flow_stat(node_sl);
-    }
-
     /* check that we have mbuf  */
     rte_mbuf_t *    m;
 
@@ -2098,7 +2175,11 @@ int CCoreEthIFStateless::send_node(CGenNode * no) {
         }
     }
 
-    send_pkt(lp_port,m,lp_stats);
+    if (unlikely(node_sl->is_stat_needed())) {
+        return send_node_flow_stat(m, node_sl, lp_port, lp_stats);
+    } else {
+        send_pkt(lp_port,m,lp_stats);
+    }
 
     return (0);
 };
@@ -3802,8 +3883,12 @@ CGlobalTRex::publish_async_data(bool sync_now, bool baseline) {
     m_zmq_publisher.publish_json(json);
 
     if (get_is_stateless()) {
-        if (m_trex_stateless->m_rx_flow_stat.dump_json(json, baseline))
-            m_zmq_publisher.publish_json(json);
+        std::string stat_json;
+        std::string latency_json;
+        if (m_trex_stateless->m_rx_flow_stat.dump_json(stat_json, latency_json, baseline)) {
+            m_zmq_publisher.publish_json(stat_json);
+            m_zmq_publisher.publish_json(latency_json);
+        }
     }
 }
 
diff --git a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
index d69b7d7f..736f3d02 100644
--- a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
+++ b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
@@ -115,8 +115,12 @@ TrexRpcCmdAddStream::_run(const Json::Value &params, Json::Value &result) {
         }
 
         stream->m_rx_check.m_pg_id      = parse_int(rx, "stream_id", result);
-        stream->m_rx_check.m_seq_enabled  = parse_bool(rx, "seq_enabled", result);
-        stream->m_rx_check.m_latency      = parse_bool(rx, "latency_enabled", result);
+        std::string type = parse_string(rx, "rule_type", result);
+        if (type == "latency") {
+            stream->m_rx_check.m_rule_type = TrexPlatformApi::IF_STAT_PAYLOAD;
+        } else {
+            stream->m_rx_check.m_rule_type = TrexPlatformApi::IF_STAT_IPV4_ID;
+        }
     }
 
     /* make sure this is a valid stream to add */
diff --git a/src/stateless/cp/trex_stream.cpp b/src/stateless/cp/trex_stream.cpp
index 9021ebf4..6959476c 100644
--- a/src/stateless/cp/trex_stream.cpp
+++ b/src/stateless/cp/trex_stream.cpp
@@ -144,7 +144,6 @@ TrexStream::TrexStream(uint8_t type,
     m_expected_pkt_len = 0;
 
     m_rx_check.m_enabled = false;
-    m_rx_check.m_rule_type = TrexPlatformApi::IF_STAT_PAYLOAD; // default for now. Should come from user???
 
     m_burst_total_pkts=0; 
     m_num_bursts=1; 
diff --git a/src/stateless/dp/trex_stateless_dp_core.cpp b/src/stateless/dp/trex_stateless_dp_core.cpp
index 9c05c16b..8bb89ee9 100644
--- a/src/stateless/dp/trex_stateless_dp_core.cpp
+++ b/src/stateless/dp/trex_stateless_dp_core.cpp
@@ -212,8 +212,7 @@ std::string CGenNodeStateless::get_stream_state_str(stream_state_t stream_state)
 }
 
 rte_mbuf_t * CGenNodeStateless::alloc_flow_stat_mbuf(rte_mbuf_t *m) {
-    //?????????
-    // temp implementation. Just copy the entire mbuf
+    //????????? temp implementation. Just copy the entire mbuf
     rte_mbuf_t *m_new = CGlobalInfo::pktmbuf_alloc( get_socket_id(), m->data_len );
     /* TBD remove this, should handle cases of error */
     assert(m_new);
diff --git a/src/stateless/rx/trex_stateless_rx_core.cpp b/src/stateless/rx/trex_stateless_rx_core.cpp
index 1a5d9a7e..d624f455 100644
--- a/src/stateless/rx/trex_stateless_rx_core.cpp
+++ b/src/stateless/rx/trex_stateless_rx_core.cpp
@@ -30,7 +30,9 @@ void CRxCoreStateless::create(const CRxSlCfg &cfg) {
     m_cpu_cp_u.Create(&m_cpu_dp_u);
 
     for (int i = 0; i < MAX_FLOW_STATS_PAYLOAD; i++) {
-        m_per_flow_seq[i] = 0;
+        // This is the seq num value we expect next packet to have.
+        // Init value should match m_seq_num in CVirtualIFPerSideStats
+        m_per_flow_seq[i] = UINT32_MAX - 1;  // catch wrap around issues early
         m_per_flow_hist[i].Reset();
         m_per_flow_jitter[i].reset();
         m_per_flow_seq_error[i] = 0;
@@ -133,7 +135,7 @@ void CRxCoreStateless::handle_rx_pkt(CLatencyManagerPerPortStl *lp, rte_mbuf_t *
             if (is_flow_stat_id(ip_id)) {
                 uint16_t hw_id;
                 if (is_flow_stat_payload_id(ip_id)) {
-                    uint32_t seq; //??? handle seq wrap around
+                    uint32_t seq;
                     uint8_t *p = rte_pktmbuf_mtod(m, uint8_t*);
                     struct flow_stat_payload_header *fsp_head = (struct flow_stat_payload_header *)
                         (p + m->pkt_len - sizeof(struct flow_stat_payload_header));
@@ -142,20 +144,34 @@ void CRxCoreStateless::handle_rx_pkt(CLatencyManagerPerPortStl *lp, rte_mbuf_t *
                         seq = fsp_head->seq;
                         if (unlikely(seq != m_per_flow_seq[hw_id])) {
                             if (seq < m_per_flow_seq[hw_id]) {
-                                if (seq ==  (m_per_flow_seq[hw_id] - 1)) {
-                                    m_per_flow_dup[hw_id] += 1;
-                                    printf("dup packets seq:%d %ld\n", seq, m_per_flow_seq[hw_id]);
+                                if (m_per_flow_seq[hw_id] - seq > 100000) {
+                                    // packet loss while we had wrap around
+                                    m_per_flow_seq_error[hw_id] += seq - m_per_flow_seq[hw_id];
+                                    m_per_flow_seq[hw_id] = seq + 1;
                                 } else {
-                                    m_per_flow_out_of_order[hw_id] += 1;
-                                    // We thought it was lost, but it was just out of order
-                                    m_per_flow_seq_error[hw_id] -= 1;
-                                    printf("ooo packets seq:%d %ld\n", seq, m_per_flow_seq[hw_id]);
+                                    if (seq == (m_per_flow_seq[hw_id] - 1)) {
+                                        m_per_flow_dup[hw_id] += 1;
+                                    } else {
+                                        m_per_flow_out_of_order[hw_id] += 1;
+                                        // We thought it was lost, but it was just out of order
+                                        m_per_flow_seq_error[hw_id] -= 1;
+                                    }
                                 }
                             } else {
-                                // seq > m_per_flow_seq[hw_id]
-                                printf("lost packets seq:%d %ld\n", seq, m_per_flow_seq[hw_id]);
-                                m_per_flow_seq_error[hw_id] += seq - m_per_flow_seq[hw_id];
-                                m_per_flow_seq[hw_id] = seq + 1;
+                                if (unlikely (m_per_flow_seq[hw_id] - seq > 100000)) {
+                                    // packet reorder while we had wrap around
+                                    if (seq == (m_per_flow_seq[hw_id] - 1)) {
+                                        m_per_flow_dup[hw_id] += 1;
+                                    } else {
+                                        m_per_flow_out_of_order[hw_id] += 1;
+                                        // We thought it was lost, but it was just out of order
+                                        m_per_flow_seq_error[hw_id] -= 1;
+                                    }
+                                } else {
+                                // seq > m_per_flow_seq[hw_id]. Assuming lost packets
+                                    m_per_flow_seq_error[hw_id] += seq - m_per_flow_seq[hw_id];
+                                    m_per_flow_seq[hw_id] = seq + 1;
+                                }
                             }
                         } else {
                             m_per_flow_seq[hw_id] = seq + 1;
diff --git a/src/stateless/rx/trex_stateless_rx_core.h b/src/stateless/rx/trex_stateless_rx_core.h
index d946f920..2ebd209a 100644
--- a/src/stateless/rx/trex_stateless_rx_core.h
+++ b/src/stateless/rx/trex_stateless_rx_core.h
@@ -101,7 +101,7 @@ class CRxCoreStateless {
     CCpuUtlCp m_cpu_cp_u;
     // Used for acking "work" (go out of idle) messages from cp
     volatile bool m_ack_start_work_msg __rte_cache_aligned;
-    uint64_t m_per_flow_seq[MAX_FLOW_STATS_PAYLOAD]; // expected next seq num
+    uint32_t m_per_flow_seq[MAX_FLOW_STATS_PAYLOAD]; // expected next seq num
     CTimeHistogram  m_per_flow_hist[MAX_FLOW_STATS_PAYLOAD]; /* latency info */
     CJitter         m_per_flow_jitter[MAX_FLOW_STATS_PAYLOAD];
     uint64_t m_per_flow_seq_error[MAX_FLOW_STATS_PAYLOAD]; // How many packet seq num gaps we saw (packets lost or out of order)