some refactor to RX core to allow multiple features

Signed-off-by: imarom <imarom@cisco.com>

5 files changed, 310 insertions, 244 deletions

diff --git a/src/stateless/rx/trex_stateless_rx_core.cpp b/src/stateless/rx/trex_stateless_rx_core.cpp
index e8d5857e..cedf2953 100644
--- a/src/stateless/rx/trex_stateless_rx_core.cpp
+++ b/src/stateless/rx/trex_stateless_rx_core.cpp
@@ -66,15 +66,7 @@ void CRFC2544Info::export_data(rfc2544_info_t_ &obj) {
     obj.set_latency_json(json);
 };
 
-void CCPortLatencyStl::reset() {
-    for (int i = 0; i < MAX_FLOW_STATS; i++) {
-        m_rx_pg_stat[i].clear();
-        m_rx_pg_stat_payload[i].clear();
-    }
-}
-
 void CRxCoreStateless::create(const CRxSlCfg &cfg) {
-    m_rcv_all = false;
     m_capture = false;
     m_max_ports = cfg.m_max_ports;
 
@@ -84,16 +76,17 @@ void CRxCoreStateless::create(const CRxSlCfg &cfg) {
     m_ring_to_cp   = cp_rx->getRingDpToCp(0);
     m_state = STATE_IDLE;
 
+    for (int i = 0; i < MAX_FLOW_STATS_PAYLOAD; i++) {
+        m_rfc2544[i].create();
+    }
+
+    /* init per port manager */
     for (int i = 0; i < m_max_ports; i++) {
-        CLatencyManagerPerPortStl * lp = &m_ports[i];
-        lp->m_io = cfg.m_ports[i];
-        lp->m_port.reset();
+        m_rx_port_mngr[i].create(cfg.m_ports[i], m_rfc2544, &m_err_cntrs);
     }
+
     m_cpu_cp_u.Create(&m_cpu_dp_u);
 
-    for (int i = 0; i < MAX_FLOW_STATS_PAYLOAD; i++) {
-        m_rfc2544[i].create();
-    }
 }
 
 void CRxCoreStateless::handle_cp_msg(TrexStatelessCpToRxMsgBase *msg) {
@@ -157,143 +150,55 @@ void CRxCoreStateless::idle_state_loop() {
     }
 }
 
-void CRxCoreStateless::start() {
-    int count = 0;
+void CRxCoreStateless::handle_work_stage(bool do_try_rx_queue) {
     int i = 0;
-    bool do_try_rx_queue =CGlobalInfo::m_options.preview.get_vm_one_queue_enable() ? true : false;
+
+    while (m_state == STATE_WORKING) {
+
+        if (do_try_rx_queue) {
+            try_rx_queues();
+        }
+
+        try_rx();
+
+        i++;
+        if (i == 100000) { // approx 10msec
+            i = 0;
+            periodic_check_for_cp_messages(); // m_state might change in here
+        }
+
+        rte_pause();
+    }
+}
+
+void CRxCoreStateless::start() {
+    bool do_try_rx_queue = CGlobalInfo::m_options.preview.get_vm_one_queue_enable() ? true : false;
 
     /* register a watchdog handle on current core */
     m_monitor.create("STL RX CORE", 1);
     TrexWatchDog::getInstance().register_monitor(&m_monitor);
 
-    while (true) {
-        if (m_state == STATE_WORKING) {
-            i++;
-            if (i == 100000) { // approx 10msec
-                i = 0;
-                periodic_check_for_cp_messages(); // m_state might change in here
-            }
-        } else {
-            if (m_state == STATE_QUIT)
-                break;
-            count = 0;
-            i = 0;
+    while (m_state != STATE_QUIT) {
+
+        switch (m_state) {
+        case STATE_IDLE:
             set_working_msg_ack(false);
             idle_state_loop();
             set_working_msg_ack(true);
-        }
-        if (do_try_rx_queue) {
-            try_rx_queues();
-        }
-        count += try_rx();
-    }
-    rte_pause();
+            break;
 
-    m_monitor.disable();
-}
+        case STATE_WORKING:
+            handle_work_stage(do_try_rx_queue);
+            break;
 
-void CRxCoreStateless::handle_rx_pkt(CLatencyManagerPerPortStl *lp, rte_mbuf_t *m) {
-    CFlowStatParser parser;
-
-    if (m_rcv_all || parser.parse(rte_pktmbuf_mtod(m, uint8_t *), m->pkt_len) == 0) {
-        uint32_t ip_id;
-        if (m_rcv_all || (parser.get_ip_id(ip_id) == 0)) {
-            if (m_rcv_all || is_flow_stat_id(ip_id)) {
-                uint16_t hw_id;
-                if (m_rcv_all || is_flow_stat_payload_id(ip_id)) {
-                    bool good_packet = true;
-                    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));
-                    hw_id = fsp_head->hw_id;
-                    CRFC2544Info *curr_rfc2544;
-
-                    if (unlikely(fsp_head->magic != FLOW_STAT_PAYLOAD_MAGIC) || hw_id >= MAX_FLOW_STATS_PAYLOAD) {
-                        good_packet = false;
-                        if (!m_rcv_all)
-                            m_err_cntrs.m_bad_header++;
-                    } else {
-                        curr_rfc2544 = &m_rfc2544[hw_id];
-
-                        if (fsp_head->flow_seq != curr_rfc2544->get_exp_flow_seq()) {
-                            // bad flow seq num
-                            // Might be the first packet of a new flow, packet from an old flow, or garbage.
-
-                            if (fsp_head->flow_seq == curr_rfc2544->get_prev_flow_seq()) {
-                                // packet from previous flow using this hw_id that arrived late
-                                good_packet = false;
-                                m_err_cntrs.m_old_flow++;
-                            } else {
-                                if (curr_rfc2544->no_flow_seq()) {
-                                    // first packet we see from this flow
-                                    good_packet = true;
-                                    curr_rfc2544->set_exp_flow_seq(fsp_head->flow_seq);
-                                } else {
-                                    // garbage packet
-                                    good_packet = false;
-                                    m_err_cntrs.m_bad_header++;
-                                }
-                            }
-                        }
-                    }
-
-                    if (good_packet) {
-                        uint32_t pkt_seq = fsp_head->seq;
-                        uint32_t exp_seq = curr_rfc2544->get_seq();
-                        if (unlikely(pkt_seq != exp_seq)) {
-                            if (pkt_seq < exp_seq) {
-                                if (exp_seq - pkt_seq > 100000) {
-                                    // packet loss while we had wrap around
-                                    curr_rfc2544->inc_seq_err(pkt_seq - exp_seq);
-                                    curr_rfc2544->inc_seq_err_too_big();
-                                    curr_rfc2544->set_seq(pkt_seq + 1);
-                                } else {
-                                    if (pkt_seq == (exp_seq - 1)) {
-                                        curr_rfc2544->inc_dup();
-                                    } else {
-                                        curr_rfc2544->inc_ooo();
-                                        // We thought it was lost, but it was just out of order
-                                        curr_rfc2544->dec_seq_err();
-                                    }
-                                    curr_rfc2544->inc_seq_err_too_low();
-                                }
-                            } else {
-                                if (unlikely (pkt_seq - exp_seq > 100000)) {
-                                    // packet reorder while we had wrap around
-                                    if (pkt_seq == (exp_seq - 1)) {
-                                        curr_rfc2544->inc_dup();
-                                    } else {
-                                        curr_rfc2544->inc_ooo();
-                                        // We thought it was lost, but it was just out of order
-                                        curr_rfc2544->dec_seq_err();
-                                    }
-                                    curr_rfc2544->inc_seq_err_too_low();
-                                } else {
-                                // seq > curr_rfc2544->seq. Assuming lost packets
-                                    curr_rfc2544->inc_seq_err(pkt_seq - exp_seq);
-                                    curr_rfc2544->inc_seq_err_too_big();
-                                    curr_rfc2544->set_seq(pkt_seq + 1);
-                                }
-                            }
-                        } else {
-                            curr_rfc2544->set_seq(pkt_seq + 1);
-                        }
-                        lp->m_port.m_rx_pg_stat_payload[hw_id].add_pkts(1);
-                        lp->m_port.m_rx_pg_stat_payload[hw_id].add_bytes(m->pkt_len + 4); // +4 for ethernet CRC
-                        uint64_t d = (os_get_hr_tick_64() - fsp_head->time_stamp );
-                        dsec_t ctime = ptime_convert_hr_dsec(d);
-                        curr_rfc2544->add_sample(ctime);
-                    }
-                } else {
-                    hw_id = get_hw_id(ip_id);
-                    if (hw_id < MAX_FLOW_STATS) {
-                        lp->m_port.m_rx_pg_stat[hw_id].add_pkts(1);
-                        lp->m_port.m_rx_pg_stat[hw_id].add_bytes(m->pkt_len + 4); // +4 for ethernet CRC
-                    }
-                }
-            }
+        default:
+            assert(0);
+            break;
         }
+
     }
+
+    m_monitor.disable();  
 }
 
 void CRxCoreStateless::capture_pkt(rte_mbuf_t *m) {
@@ -313,7 +218,7 @@ void CRxCoreStateless::handle_rx_queue_msgs(uint8_t thread_id, CNodeRing * r) {
         CGenNodeLatencyPktInfo * l_msg;
         uint8_t msg_type =  msg->m_msg_type;
         uint8_t rx_port_index;
-        CLatencyManagerPerPortStl * lp;
+
 
         switch (msg_type) {
         case CGenNodeMsgBase::LATENCY_PKT:
@@ -321,8 +226,9 @@ void CRxCoreStateless::handle_rx_queue_msgs(uint8_t thread_id, CNodeRing * r) {
             assert(l_msg->m_latency_offset == 0xdead);
             rx_port_index = (thread_id << 1) + (l_msg->m_dir & 1);
             assert( rx_port_index < m_max_ports );
-            lp = &m_ports[rx_port_index];
-            handle_rx_pkt(lp, (rte_mbuf_t *)l_msg->m_pkt);
+
+            m_rx_port_mngr[rx_port_index].handle_pkt((rte_mbuf_t *)l_msg->m_pkt);
+
             if (m_capture)
                 capture_pkt((rte_mbuf_t *)l_msg->m_pkt);
             rte_pktmbuf_free((rte_mbuf_t *)l_msg->m_pkt);
@@ -350,89 +256,55 @@ void CRxCoreStateless::try_rx_queues() {
     }
 }
 
-// exactly the same as try_rx, without the handle_rx_pkt
-// purpose is to flush rx queues when core is in idle state
-void CRxCoreStateless::flush_rx() {
+int CRxCoreStateless::try_rx(bool flush_rx) {
     rte_mbuf_t * rx_pkts[64];
     int i, total_pkts = 0;
     for (i = 0; i < m_max_ports; i++) {
-        CLatencyManagerPerPortStl * lp = &m_ports[i];
-        rte_mbuf_t * m;
-        /* try to read 64 packets clean up the queue */
-        uint16_t cnt_p = lp->m_io->rx_burst(rx_pkts, 64);
-        total_pkts += cnt_p;
-        if (cnt_p) {
-            m_cpu_dp_u.start_work1();
-            int j;
-            for (j = 0; j < cnt_p; j++) {
-                m = rx_pkts[j];
-                rte_pktmbuf_free(m);
-            }
-            /* commit only if there was work to do ! */
-            m_cpu_dp_u.commit1();
-        }/* if work */
-    }// all ports
-}
-
+        RXPortManager &lp = m_rx_port_mngr[i];
 
-int CRxCoreStateless::try_rx() {
-    rte_mbuf_t * rx_pkts[64];
-    int i, total_pkts = 0;
-    for (i = 0; i < m_max_ports; i++) {
-        CLatencyManagerPerPortStl * lp = &m_ports[i];
-        rte_mbuf_t * m;
         /* try to read 64 packets clean up the queue */
-        uint16_t cnt_p = lp->m_io->rx_burst(rx_pkts, 64);
+        uint16_t cnt_p = lp.get_io()->rx_burst(rx_pkts, 64);
+        /* if no packets or its a flush - ignore */
+        if ( (cnt_p == 0) || flush_rx ) {
+            continue;
+        }
+
         total_pkts += cnt_p;
-        if (cnt_p) {
-            m_cpu_dp_u.start_work1();
-            int j;
-            for (j = 0; j < cnt_p; j++) {
-                m = rx_pkts[j];
-                //handle_rx_pkt(lp, m);
-                m_rx_port_mngr[i].handle_pkt(m);
-                rte_pktmbuf_free(m);
-            }
-            /* commit only if there was work to do ! */
-            m_cpu_dp_u.commit1();
-        }/* if work */
-    }// all ports
-    return total_pkts;
-}
+        m_cpu_dp_u.start_work1();
 
-bool CRxCoreStateless::is_flow_stat_id(uint32_t id) {
-    if ((id & 0x000fff00) == IP_ID_RESERVE_BASE) return true;
-    return false;
-}
+        for (int j = 0; j < cnt_p; j++) {
+            rte_mbuf_t *m = rx_pkts[j];
+            lp.handle_pkt(m);
+            rte_pktmbuf_free(m);
+        }
+        /* commit only if there was work to do ! */
+        m_cpu_dp_u.commit1();
 
-bool CRxCoreStateless::is_flow_stat_payload_id(uint32_t id) {
-    if (id == FLOW_STAT_PAYLOAD_IP_ID) return true;
-    return false;
-}
+    }
 
-uint16_t CRxCoreStateless::get_hw_id(uint16_t id) {
-    return (0x00ff & id);
+    return total_pkts;
 }
 
 void CRxCoreStateless::reset_rx_stats(uint8_t port_id) {
-    for (int hw_id = 0; hw_id < MAX_FLOW_STATS; hw_id++) {
-        m_ports[port_id].m_port.m_rx_pg_stat[hw_id].clear();
-    }
+    m_rx_port_mngr[port_id].clear_stats();
 }
 
 int CRxCoreStateless::get_rx_stats(uint8_t port_id, rx_per_flow_t *rx_stats, int min, int max
                                    , bool reset, TrexPlatformApi::driver_stat_cap_e type) {
+
+    RXLatency &latency = m_rx_port_mngr[port_id].get_latency();
+
     for (int hw_id = min; hw_id <= max; hw_id++) {
         if (type == TrexPlatformApi::IF_STAT_PAYLOAD) {
-            rx_stats[hw_id - min] = m_ports[port_id].m_port.m_rx_pg_stat_payload[hw_id];
+            rx_stats[hw_id - min] = latency.m_rx_pg_stat_payload[hw_id];
         } else {
-            rx_stats[hw_id - min] = m_ports[port_id].m_port.m_rx_pg_stat[hw_id];
+            rx_stats[hw_id - min] = latency.m_rx_pg_stat[hw_id];
         }
         if (reset) {
             if (type == TrexPlatformApi::IF_STAT_PAYLOAD) {
-                m_ports[port_id].m_port.m_rx_pg_stat_payload[hw_id].clear();
+                latency.m_rx_pg_stat_payload[hw_id].clear();
             } else {
-                m_ports[port_id].m_port.m_rx_pg_stat[hw_id].clear();
+                latency.m_rx_pg_stat[hw_id].clear();
             }
         }
     }
diff --git a/src/stateless/rx/trex_stateless_rx_core.h b/src/stateless/rx/trex_stateless_rx_core.h
index 294c7527..4e6c0c46 100644
--- a/src/stateless/rx/trex_stateless_rx_core.h
+++ b/src/stateless/rx/trex_stateless_rx_core.h
@@ -29,33 +29,6 @@
 
 class TrexStatelessCpToRxMsgBase;
 
-class CCPortLatencyStl {
- public:
-    void reset();
-
- public:
-    rx_per_flow_t m_rx_pg_stat[MAX_FLOW_STATS];
-    rx_per_flow_t m_rx_pg_stat_payload[MAX_FLOW_STATS_PAYLOAD];
-};
-
-class CLatencyManagerPerPortStl {
-public:
-     CCPortLatencyStl     m_port;
-     CPortLatencyHWBase * m_io;
-};
-
-class CRxSlCfg {
- public:
-    CRxSlCfg (){
-        m_max_ports = 0;
-        m_cps = 0.0;
-    }
-
- public:
-    uint32_t             m_max_ports;
-    double               m_cps;
-    CPortLatencyHWBase * m_ports[TREX_MAX_PORTS];
-};
 
 class CRFC2544Info {
  public:
@@ -110,7 +83,7 @@ class CRxCoreErrCntrs {
         m_old_flow = 0;
     }
 
- private:
+ public:
     uint64_t m_bad_header;
     uint64_t m_old_flow;
 };
@@ -159,30 +132,31 @@ class CRxCoreStateless {
     bool periodic_check_for_cp_messages();
     void tickle();
     void idle_state_loop();
-    void handle_rx_pkt(CLatencyManagerPerPortStl * lp, rte_mbuf_t * m);
-    void handle_rx_pkt_2(int port_id, rte_mbuf_t *m);
     void capture_pkt(rte_mbuf_t *m);
     void handle_rx_queue_msgs(uint8_t thread_id, CNodeRing * r);
-    void flush_rx();
-    int try_rx();
+    void handle_work_stage(bool do_try_rx_queue);
+
+    int try_rx(bool flush_rx = false);
+
+    void flush_rx() {
+        try_rx(true);
+    }
+
     void try_rx_queues();
-    bool is_flow_stat_id(uint32_t id);
-    bool is_flow_stat_payload_id(uint32_t id);
-    uint16_t get_hw_id(uint16_t id);
 
  private:
     TrexMonitor     m_monitor;
     uint32_t m_max_ports;
     bool m_capture;
-    bool m_rcv_all;
-    CLatencyManagerPerPortStl m_ports[TREX_MAX_PORTS];
     state_e   m_state;
     CNodeRing *m_ring_from_cp;
     CNodeRing *m_ring_to_cp;
     CCpuUtlDp m_cpu_dp_u;
     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;
+
     CRxCoreErrCntrs m_err_cntrs;
     CRFC2544Info m_rfc2544[MAX_FLOW_STATS_PAYLOAD];
 
diff --git a/src/stateless/rx/trex_stateless_rx_defs.h b/src/stateless/rx/trex_stateless_rx_defs.h
index 9fd11c14..9df6af67 100644
--- a/src/stateless/rx/trex_stateless_rx_defs.h
+++ b/src/stateless/rx/trex_stateless_rx_defs.h
@@ -22,6 +22,27 @@
 #ifndef __TREX_STATELESS_RX_DEFS_H__
 #define __TREX_STATELESS_RX_DEFS_H__
 
+#include "trex_defs.h"
+
+class CPortLatencyHWBase;
+
+/**
+ * general SL cfg
+ * 
+ */
+class CRxSlCfg {
+ public:
+    CRxSlCfg (){
+        m_max_ports = 0;
+        m_cps = 0.0;
+    }
+
+ public:
+    uint32_t             m_max_ports;
+    double               m_cps;
+    CPortLatencyHWBase * m_ports[TREX_MAX_PORTS];
+};
+
 /**
  * describes the filter type applied to the RX 
  *  RX_FILTER_MODE_HW - only hardware filtered traffic will
diff --git a/src/stateless/rx/trex_stateless_rx_port_mngr.cpp b/src/stateless/rx/trex_stateless_rx_port_mngr.cpp
index 8a0fd613..fdf41471 100644
--- a/src/stateless/rx/trex_stateless_rx_port_mngr.cpp
+++ b/src/stateless/rx/trex_stateless_rx_port_mngr.cpp
@@ -21,6 +21,121 @@
 #include "bp_sim.h"
 #include "trex_stateless_rx_port_mngr.h"
 #include "common/captureFile.h"
+#include "trex_stateless_rx_core.h"
+
+/************************** latency feature ************/
+void RXLatency::handle_pkt(const rte_mbuf_t *m) {
+    CFlowStatParser parser;
+
+    if (m_rcv_all || parser.parse(rte_pktmbuf_mtod(m, uint8_t *), m->pkt_len) == 0) {
+        uint32_t ip_id;
+        if (m_rcv_all || (parser.get_ip_id(ip_id) == 0)) {
+            if (m_rcv_all || is_flow_stat_id(ip_id)) {
+                uint16_t hw_id;
+                if (m_rcv_all || is_flow_stat_payload_id(ip_id)) {
+                    bool good_packet = true;
+                    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));
+                    hw_id = fsp_head->hw_id;
+                    CRFC2544Info *curr_rfc2544;
+
+                    if (unlikely(fsp_head->magic != FLOW_STAT_PAYLOAD_MAGIC) || hw_id >= MAX_FLOW_STATS_PAYLOAD) {
+                        good_packet = false;
+                        if (!m_rcv_all)
+                            m_err_cntrs->m_bad_header++;
+                    } else {
+                        curr_rfc2544 = &m_rfc2544[hw_id];
+
+                        if (fsp_head->flow_seq != curr_rfc2544->get_exp_flow_seq()) {
+                            // bad flow seq num
+                            // Might be the first packet of a new flow, packet from an old flow, or garbage.
+
+                            if (fsp_head->flow_seq == curr_rfc2544->get_prev_flow_seq()) {
+                                // packet from previous flow using this hw_id that arrived late
+                                good_packet = false;
+                                m_err_cntrs->m_old_flow++;
+                            } else {
+                                if (curr_rfc2544->no_flow_seq()) {
+                                    // first packet we see from this flow
+                                    good_packet = true;
+                                    curr_rfc2544->set_exp_flow_seq(fsp_head->flow_seq);
+                                } else {
+                                    // garbage packet
+                                    good_packet = false;
+                                    m_err_cntrs->m_bad_header++;
+                                }
+                            }
+                        }
+                    }
+
+                    if (good_packet) {
+                        uint32_t pkt_seq = fsp_head->seq;
+                        uint32_t exp_seq = curr_rfc2544->get_seq();
+                        if (unlikely(pkt_seq != exp_seq)) {
+                            if (pkt_seq < exp_seq) {
+                                if (exp_seq - pkt_seq > 100000) {
+                                    // packet loss while we had wrap around
+                                    curr_rfc2544->inc_seq_err(pkt_seq - exp_seq);
+                                    curr_rfc2544->inc_seq_err_too_big();
+                                    curr_rfc2544->set_seq(pkt_seq + 1);
+                                } else {
+                                    if (pkt_seq == (exp_seq - 1)) {
+                                        curr_rfc2544->inc_dup();
+                                    } else {
+                                        curr_rfc2544->inc_ooo();
+                                        // We thought it was lost, but it was just out of order
+                                        curr_rfc2544->dec_seq_err();
+                                    }
+                                    curr_rfc2544->inc_seq_err_too_low();
+                                }
+                            } else {
+                                if (unlikely (pkt_seq - exp_seq > 100000)) {
+                                    // packet reorder while we had wrap around
+                                    if (pkt_seq == (exp_seq - 1)) {
+                                        curr_rfc2544->inc_dup();
+                                    } else {
+                                        curr_rfc2544->inc_ooo();
+                                        // We thought it was lost, but it was just out of order
+                                        curr_rfc2544->dec_seq_err();
+                                    }
+                                    curr_rfc2544->inc_seq_err_too_low();
+                                } else {
+                                // seq > curr_rfc2544->seq. Assuming lost packets
+                                    curr_rfc2544->inc_seq_err(pkt_seq - exp_seq);
+                                    curr_rfc2544->inc_seq_err_too_big();
+                                    curr_rfc2544->set_seq(pkt_seq + 1);
+                                }
+                            }
+                        } else {
+                            curr_rfc2544->set_seq(pkt_seq + 1);
+                        }
+                        m_rx_pg_stat_payload[hw_id].add_pkts(1);
+                        m_rx_pg_stat_payload[hw_id].add_bytes(m->pkt_len + 4); // +4 for ethernet CRC
+                        uint64_t d = (os_get_hr_tick_64() - fsp_head->time_stamp );
+                        dsec_t ctime = ptime_convert_hr_dsec(d);
+                        curr_rfc2544->add_sample(ctime);
+                    }
+                } else {
+                    hw_id = get_hw_id(ip_id);
+                    if (hw_id < MAX_FLOW_STATS) {
+                        m_rx_pg_stat[hw_id].add_pkts(1);
+                        m_rx_pg_stat[hw_id].add_bytes(m->pkt_len + 4); // +4 for ethernet CRC
+                    }
+                }
+            }
+        }
+    }
+}
+
+void
+RXLatency::reset_stats() {
+    for (int hw_id = 0; hw_id < MAX_FLOW_STATS; hw_id++) {
+        m_rx_pg_stat[hw_id].clear();
+    }
+}
+
+/****************************** packet recorder ****************************/
 
 RXPacketRecorder::RXPacketRecorder() {
     m_writer = NULL;
diff --git a/src/stateless/rx/trex_stateless_rx_port_mngr.h b/src/stateless/rx/trex_stateless_rx_port_mngr.h
index 622dffda..8737e267 100644
--- a/src/stateless/rx/trex_stateless_rx_port_mngr.h
+++ b/src/stateless/rx/trex_stateless_rx_port_mngr.h
@@ -27,6 +27,63 @@
 
 #include "common/captureFile.h"
 
+/************************* latency ***********************/
+
+class CPortLatencyHWBase;
+class CRFC2544Info;
+class CRxCoreErrCntrs;
+
+
+class RXLatency {
+public:
+
+    RXLatency() {
+        m_rcv_all    = false;
+        m_rfc2544    = NULL;
+        m_err_cntrs  = NULL;
+
+        for (int i = 0; i < MAX_FLOW_STATS; i++) {
+            m_rx_pg_stat[i].clear();
+            m_rx_pg_stat_payload[i].clear();
+        }
+    }
+
+    void create(CRFC2544Info *rfc2544, CRxCoreErrCntrs *err_cntrs) {
+        m_rfc2544 = rfc2544;
+        m_err_cntrs = err_cntrs;
+    }
+
+    void reset_stats();
+
+    void handle_pkt(const rte_mbuf_t *m);
+
+private:
+    bool is_flow_stat_id(uint32_t id) {
+        if ((id & 0x000fff00) == IP_ID_RESERVE_BASE) return true;
+        return false;
+    }
+
+    bool is_flow_stat_payload_id(uint32_t id) {
+        if (id == FLOW_STAT_PAYLOAD_IP_ID) return true;
+        return false;
+    }
+
+    uint16_t get_hw_id(uint16_t id) {
+    return (0x00ff & id);
+}
+
+public:
+
+    rx_per_flow_t        m_rx_pg_stat[MAX_FLOW_STATS];
+    rx_per_flow_t        m_rx_pg_stat_payload[MAX_FLOW_STATS_PAYLOAD];
+
+    bool                 m_rcv_all;
+    CRFC2544Info         *m_rfc2544;
+    CRxCoreErrCntrs      *m_err_cntrs;
+};
+
+/************************ queue ***************************/
+
 /**                
  * describes a single saved RX packet
  * 
@@ -145,6 +202,8 @@ private:
     RxPacket **m_buffer;
 };
 
+/************************ recoder ***************************/
+
 /**
  * RX packet recorder to PCAP file
  * 
@@ -165,6 +224,8 @@ private:
 };
 
 
+/************************ manager ***************************/
+
 /**
  * per port RX features manager
  * 
@@ -181,9 +242,23 @@ public:
     RXPortManager() {
         m_features = 0;
         m_pkt_buffer = NULL;
+        m_io = NULL;
         set_feature(LATENCY);
     }
 
+    void create(CPortLatencyHWBase *io, CRFC2544Info *rfc2544, CRxCoreErrCntrs *err_cntrs) {
+        m_io = io;
+        m_latency.create(rfc2544, err_cntrs);
+    }
+
+    void clear_stats() {
+        m_latency.reset_stats();
+    }
+
+    RXLatency & get_latency() {
+        return m_latency;
+    }
+
     void start_recorder(const std::string &pcap, uint32_t limit_pkts) {
         m_recorder.start(pcap, limit_pkts);
         set_feature(RECORD);
@@ -227,9 +302,11 @@ public:
     }
 
     void handle_pkt(const rte_mbuf_t *m) {
-        /* fast path */
-        if (no_features_set()) {
-            return;
+
+        /* handle features */
+
+        if (is_feature_set(LATENCY)) {
+            m_latency.handle_pkt(m);
         }
 
         if (is_feature_set(RECORD)) {
@@ -241,6 +318,10 @@ public:
         }
     }
 
+    CPortLatencyHWBase *get_io() {
+        return m_io;
+    }
+
 private:
 
     void set_feature(features_t feature) {
@@ -259,9 +340,12 @@ private:
         return (m_features == 0);
     }
 
-    uint32_t            m_features;
-    RXPacketRecorder    m_recorder;
-    RxPacketBuffer     *m_pkt_buffer;
+    uint32_t                     m_features;
+    RXPacketRecorder             m_recorder;
+    RXLatency                    m_latency;
+    RxPacketBuffer              *m_pkt_buffer;
+
+    CPortLatencyHWBase          *m_io;
 };