RX software features

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

4 files changed, 102 insertions, 51 deletions

diff --git a/src/stateless/rx/trex_stateless_rx_core.cpp b/src/stateless/rx/trex_stateless_rx_core.cpp
index cedf2953..ca39b8a0 100644
--- a/src/stateless/rx/trex_stateless_rx_core.cpp
+++ b/src/stateless/rx/trex_stateless_rx_core.cpp
@@ -80,13 +80,15 @@ void CRxCoreStateless::create(const CRxSlCfg &cfg) {
         m_rfc2544[i].create();
     }
 
+    m_cpu_cp_u.Create(&m_cpu_dp_u);
+
     /* init per port manager */
     for (int i = 0; i < m_max_ports; i++) {
-        m_rx_port_mngr[i].create(cfg.m_ports[i], m_rfc2544, &m_err_cntrs);
+        m_rx_port_mngr[i].create(cfg.m_ports[i],
+                                 m_rfc2544,
+                                 &m_err_cntrs,
+                                 &m_cpu_dp_u);
     }
-
-    m_cpu_cp_u.Create(&m_cpu_dp_u);
-
 }
 
 void CRxCoreStateless::handle_cp_msg(TrexStatelessCpToRxMsgBase *msg) {
@@ -137,7 +139,7 @@ void CRxCoreStateless::idle_state_loop() {
             counter = 0;
             continue;
         } else {
-            flush_rx();
+            flush_all_pending_pkts();
         }
 
         /* enter deep sleep only if enough time had passed */
@@ -159,7 +161,7 @@ void CRxCoreStateless::handle_work_stage(bool do_try_rx_queue) {
             try_rx_queues();
         }
 
-        try_rx();
+        process_all_pending_pkts();
 
         i++;
         if (i == 100000) { // approx 10msec
@@ -256,35 +258,18 @@ void CRxCoreStateless::try_rx_queues() {
     }
 }
 
-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++) {
-        RXPortManager &lp = m_rx_port_mngr[i];
-
-        /* try to read 64 packets clean up the queue */
-        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;
-        m_cpu_dp_u.start_work1();
+int CRxCoreStateless::process_all_pending_pkts(bool flush_rx) {
 
-        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();
+	int total_pkts = 0;
+	for (int i = 0; i < m_max_ports; i++) {
+		total_pkts += m_rx_port_mngr[i].process_all_pending_pkts(flush_rx);
+	}
 
-    }
+	return total_pkts;
 
-    return total_pkts;
 }
 
+
 void CRxCoreStateless::reset_rx_stats(uint8_t port_id) {
     m_rx_port_mngr[port_id].clear_stats();
 }
@@ -351,6 +336,14 @@ double CRxCoreStateless::get_cpu_util() {
 }
 
 
+/**
+ * configure RX filtering mode (HW or software)
+ * 
+ * @author imarom (11/1/2016)
+ * 
+ * @param port_id 
+ * @param filter_mode 
+ */
 void
 CRxCoreStateless::set_rx_filter_mode(uint8_t port_id, rx_filter_mode_e filter_mode) {
     const TrexPlatformApi *api = get_stateless_obj()->get_platform_api();
diff --git a/src/stateless/rx/trex_stateless_rx_core.h b/src/stateless/rx/trex_stateless_rx_core.h
index 4e6c0c46..2a59862b 100644
--- a/src/stateless/rx/trex_stateless_rx_core.h
+++ b/src/stateless/rx/trex_stateless_rx_core.h
@@ -136,10 +136,10 @@ class CRxCoreStateless {
     void handle_rx_queue_msgs(uint8_t thread_id, CNodeRing * r);
     void handle_work_stage(bool do_try_rx_queue);
 
-    int try_rx(bool flush_rx = false);
+    int process_all_pending_pkts(bool flush_rx = false);
 
-    void flush_rx() {
-        try_rx(true);
+    void flush_all_pending_pkts() {
+        process_all_pending_pkts(true);
     }
 
     void try_rx_queues();
diff --git a/src/stateless/rx/trex_stateless_rx_port_mngr.cpp b/src/stateless/rx/trex_stateless_rx_port_mngr.cpp
index fdf41471..5b10eebf 100644
--- a/src/stateless/rx/trex_stateless_rx_port_mngr.cpp
+++ b/src/stateless/rx/trex_stateless_rx_port_mngr.cpp
@@ -196,3 +196,53 @@ RXPacketRecorder::handle_pkt(const rte_mbuf_t *m) {
         stop();
     }
 }
+
+
+void RXPortManager::handle_pkt(const rte_mbuf_t *m) {
+
+    /* handle features */
+
+    if (is_feature_set(LATENCY)) {
+        m_latency.handle_pkt(m);
+    }
+
+    if (is_feature_set(RECORD)) {
+        m_recorder.handle_pkt(m);
+    }
+
+    if (is_feature_set(QUEUE)) {
+        m_pkt_buffer->push(new RxPacket(m));
+    }
+}
+
+
+int RXPortManager::process_all_pending_pkts(bool flush_rx) {
+
+    rte_mbuf_t *rx_pkts[64];
+
+    /* try to read 64 packets clean up the queue */
+    uint16_t cnt_p = m_io->rx_burst(rx_pkts, 64);
+    if (cnt_p == 0) {
+        return cnt_p;
+    }
+
+
+    m_cpu_dp_u->start_work1();
+
+    for (int j = 0; j < cnt_p; j++) {
+        rte_mbuf_t *m = rx_pkts[j];
+
+        if (!flush_rx) {
+            handle_pkt(m);
+        }
+
+        rte_pktmbuf_free(m);
+    }
+
+    /* commit only if there was work to do ! */
+    m_cpu_dp_u->commit1();
+
+
+    return cnt_p;
+}
+
diff --git a/src/stateless/rx/trex_stateless_rx_port_mngr.h b/src/stateless/rx/trex_stateless_rx_port_mngr.h
index 8737e267..d197c423 100644
--- a/src/stateless/rx/trex_stateless_rx_port_mngr.h
+++ b/src/stateless/rx/trex_stateless_rx_port_mngr.h
@@ -242,12 +242,17 @@ public:
     RXPortManager() {
         m_features = 0;
         m_pkt_buffer = NULL;
-        m_io = NULL;
+        m_io         = NULL;
+        m_cpu_dp_u   = NULL;
         set_feature(LATENCY);
     }
 
-    void create(CPortLatencyHWBase *io, CRFC2544Info *rfc2544, CRxCoreErrCntrs *err_cntrs) {
+    void create(CPortLatencyHWBase *io,
+                CRFC2544Info *rfc2544,
+                CRxCoreErrCntrs *err_cntrs,
+                CCpuUtlDp *cpu_util) {
         m_io = io;
+        m_cpu_dp_u = cpu_util;
         m_latency.create(rfc2544, err_cntrs);
     }
 
@@ -301,28 +306,31 @@ public:
         return old_buffer;
     }
 
-    void handle_pkt(const rte_mbuf_t *m) {
 
-        /* handle features */
-
-        if (is_feature_set(LATENCY)) {
-            m_latency.handle_pkt(m);
-        }
+    /**
+     * fetch and process all packets
+     * 
+     */
+    int process_all_pending_pkts(bool flush_rx = false);
 
-        if (is_feature_set(RECORD)) {
-            m_recorder.handle_pkt(m);
-        }
 
-        if (is_feature_set(QUEUE)) {
-            m_pkt_buffer->push(new RxPacket(m));
-        }
+    /**
+     * flush all pending packets without processing them
+     * 
+     */
+    void flush_all_pending_pkts() {
+        process_all_pending_pkts(true);
     }
 
-    CPortLatencyHWBase *get_io() {
-        return m_io;
-    }
+    
+    /**
+     * handle a single packet
+     * 
+     */
+    void handle_pkt(const rte_mbuf_t *m);
 
 private:
+    
 
     void set_feature(features_t feature) {
         m_features |= feature;
@@ -344,7 +352,7 @@ private:
     RXPacketRecorder             m_recorder;
     RXLatency                    m_latency;
     RxPacketBuffer              *m_pkt_buffer;
-
+    CCpuUtlDp                   *m_cpu_dp_u;
     CPortLatencyHWBase          *m_io;
 };