refactor RX core

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

7 files changed, 98 insertions, 51 deletions

diff --git a/src/stateless/cp/trex_stateless_port.h b/src/stateless/cp/trex_stateless_port.h
index f2829b8a..74ab17f1 100644
--- a/src/stateless/cp/trex_stateless_port.h
+++ b/src/stateless/cp/trex_stateless_port.h
@@ -377,21 +377,21 @@ public:
      */
     void stop_rx_capture();
 
-	/**
-	 * start RX queueing of packets
-	 * 
-	 * @author imarom (11/7/2016)
-	 * 
-	 * @param limit 
-	 */
-	void start_rx_queue(uint64_t limit);
-
-	/**
-	 * stop RX queueing
-	 * 
-	 * @author imarom (11/7/2016)
-	 */
-	void stop_rx_queue();
+    /**
+     * start RX queueing of packets
+     * 
+     * @author imarom (11/7/2016)
+     * 
+     * @param limit 
+     */
+    void start_rx_queue(uint64_t limit);
+
+    /**
+     * stop RX queueing
+     * 
+     * @author imarom (11/7/2016)
+     */
+    void stop_rx_queue();
 
     /**
      * fetch the RX queue packets from the queue
diff --git a/src/stateless/messaging/trex_stateless_messaging.cpp b/src/stateless/messaging/trex_stateless_messaging.cpp
index 53d5a87e..17acb21e 100644
--- a/src/stateless/messaging/trex_stateless_messaging.cpp
+++ b/src/stateless/messaging/trex_stateless_messaging.cpp
@@ -243,6 +243,8 @@ TrexDpPortEventMsg::handle() {
 /************************* messages from CP to RX **********************/
 bool TrexStatelessRxEnableLatency::handle (CRxCoreStateless *rx_core) {
     rx_core->enable_latency();
+    m_reply.set_reply(true);
+    
     return true;
 }
 
diff --git a/src/stateless/messaging/trex_stateless_messaging.h b/src/stateless/messaging/trex_stateless_messaging.h
index 303548aa..79a6bf08 100644
--- a/src/stateless/messaging/trex_stateless_messaging.h
+++ b/src/stateless/messaging/trex_stateless_messaging.h
@@ -63,7 +63,7 @@ public:
         m_pending = false;
     }
 
-    T wait_for_reply(int timeout_ms = 100, int backoff_ms = 1) {
+    T wait_for_reply(int timeout_ms = 500, int backoff_ms = 1) {
         int guard = timeout_ms;
 
         while (is_pending()) {
@@ -461,7 +461,14 @@ public:
 
 
 class TrexStatelessRxEnableLatency : public TrexStatelessCpToRxMsgBase {
+public:
+    TrexStatelessRxEnableLatency(MsgReply<bool> &reply) : m_reply(reply) {
+    }
+    
     bool handle (CRxCoreStateless *rx_core);
+    
+private:
+    MsgReply<bool>    &m_reply;
 };
 
 class TrexStatelessRxDisableLatency : public TrexStatelessCpToRxMsgBase {
@@ -505,7 +512,7 @@ public:
     virtual bool handle(CRxCoreStateless *rx_core);
 
 private:
-    uint8_t           m_port_id;
+    uint8_t m_port_id;
 };
 
 
@@ -515,8 +522,8 @@ public:
                               uint64_t size,
                               MsgReply<bool> &reply) : m_reply(reply) {
         
-        m_port_id           = port_id;
-        m_size              = size;
+        m_port_id = port_id;
+        m_size    = size;
     }
 
     virtual bool handle(CRxCoreStateless *rx_core);
@@ -537,7 +544,7 @@ public:
     virtual bool handle(CRxCoreStateless *rx_core);
 
 private:
-    uint8_t           m_port_id;
+    uint8_t m_port_id;
 };
 
 
diff --git a/src/stateless/rx/trex_stateless_rx_core.cpp b/src/stateless/rx/trex_stateless_rx_core.cpp
index f518fcd3..dc637163 100644
--- a/src/stateless/rx/trex_stateless_rx_core.cpp
+++ b/src/stateless/rx/trex_stateless_rx_core.cpp
@@ -121,6 +121,7 @@ bool CRxCoreStateless::periodic_check_for_cp_messages() {
         handle_cp_msg(msg);
     }
 
+    /* a message might result in a change of state */
     recalculate_next_state();
     return true;
 
@@ -218,9 +219,7 @@ void CRxCoreStateless::start() {
 
         switch (m_state) {
         case STATE_IDLE:
-            set_working_msg_ack(false);
             idle_state_loop();
-            set_working_msg_ack(true);
             break;
 
         case STATE_WORKING:
@@ -311,23 +310,11 @@ void CRxCoreStateless::reset_rx_stats(uint8_t port_id) {
 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] = latency.m_rx_pg_stat_payload[hw_id];
-        } else {
-            rx_stats[hw_id - min] = latency.m_rx_pg_stat[hw_id];
-        }
-        if (reset) {
-            if (type == TrexPlatformApi::IF_STAT_PAYLOAD) {
-                latency.m_rx_pg_stat_payload[hw_id].clear();
-            } else {
-                latency.m_rx_pg_stat[hw_id].clear();
-            }
-        }
-    }
-    return 0;
+    /* for now only latency stats */
+    m_rx_port_mngr[port_id].get_latency_stats(rx_stats, min, max, reset, type);
+    
+    return (0);
+    
 }
 
 int CRxCoreStateless::get_rfc2544_info(rfc2544_info_t *rfc2544_info, int min, int max, bool reset) {
@@ -354,13 +341,6 @@ int CRxCoreStateless::get_rx_err_cntrs(CRxCoreErrCntrs *rx_err) {
     return 0;
 }
 
-void CRxCoreStateless::set_working_msg_ack(bool val) {
-    sanb_smp_memory_barrier();
-    m_ack_start_work_msg = val;
-    sanb_smp_memory_barrier();
-}
-
-
 void CRxCoreStateless::update_cpu_util(){
     m_cpu_cp_u.Update();
 }
@@ -373,21 +353,25 @@ double CRxCoreStateless::get_cpu_util() {
 void
 CRxCoreStateless::start_recorder(uint8_t port_id, const std::string &pcap_filename, uint64_t limit) {
     m_rx_port_mngr[port_id].start_recorder(pcap_filename, limit);
+    recalculate_next_state();
 }
 
 void
 CRxCoreStateless::stop_recorder(uint8_t port_id) {
     m_rx_port_mngr[port_id].stop_recorder();
+    recalculate_next_state();
 }
 
 void
 CRxCoreStateless::start_queue(uint8_t port_id, uint64_t size) {
     m_rx_port_mngr[port_id].start_queue(size);
+    recalculate_next_state();
 }
 
 void
 CRxCoreStateless::stop_queue(uint8_t port_id) {
     m_rx_port_mngr[port_id].stop_queue();
+    recalculate_next_state();
 }
 
 void
@@ -395,6 +379,8 @@ CRxCoreStateless::enable_latency() {
     for (int i = 0; i < m_max_ports; i++) {
         m_rx_port_mngr[i].enable_latency();
     }
+    
+    recalculate_next_state();
 }
 
 void
@@ -402,6 +388,8 @@ CRxCoreStateless::disable_latency() {
     for (int i = 0; i < m_max_ports; i++) {
         m_rx_port_mngr[i].disable_latency();
     }
+    
+    recalculate_next_state();
 }
 
 const RXPortManager &
diff --git a/src/stateless/rx/trex_stateless_rx_core.h b/src/stateless/rx/trex_stateless_rx_core.h
index b27a7ca5..7481ae2f 100644
--- a/src/stateless/rx/trex_stateless_rx_core.h
+++ b/src/stateless/rx/trex_stateless_rx_core.h
@@ -29,7 +29,10 @@
 
 class TrexStatelessCpToRxMsgBase;
 
-
+/**
+ * RFC 2544 implementation
+ * 
+ */
 class CRFC2544Info {
  public:
     void create();
@@ -88,7 +91,15 @@ class CRxCoreErrCntrs {
     uint64_t m_old_flow;
 };
 
+/**
+ * stateless RX core
+ * 
+ */
 class CRxCoreStateless {
+    
+    /**
+     * core states
+     */
     enum state_e {
         STATE_IDLE,
         STATE_WORKING,
@@ -106,8 +117,7 @@ class CRxCoreStateless {
 
 
     void quit() {m_state = STATE_QUIT;}
-    bool is_working() const {return (m_ack_start_work_msg == true);}
-    void set_working_msg_ack(bool val);
+    bool is_working() const {return (m_state == STATE_WORKING);}
     double get_cpu_util();
     void update_cpu_util();
 
diff --git a/src/stateless/rx/trex_stateless_rx_port_mngr.cpp b/src/stateless/rx/trex_stateless_rx_port_mngr.cpp
index 00032e8b..bd8650a4 100644
--- a/src/stateless/rx/trex_stateless_rx_port_mngr.cpp
+++ b/src/stateless/rx/trex_stateless_rx_port_mngr.cpp
@@ -156,6 +156,31 @@ RXLatency::reset_stats() {
     }
 }
 
+
+void
+RXLatency::get_stats(rx_per_flow_t *rx_stats,
+                     int min,
+                     int max,
+                     bool reset,
+                     TrexPlatformApi::driver_stat_cap_e type) {
+    
+    for (int hw_id = min; hw_id <= max; hw_id++) {
+        if (type == TrexPlatformApi::IF_STAT_PAYLOAD) {
+            rx_stats[hw_id - min] = m_rx_pg_stat_payload[hw_id];
+        } else {
+            rx_stats[hw_id - min] = m_rx_pg_stat[hw_id];
+        }
+        if (reset) {
+            if (type == TrexPlatformApi::IF_STAT_PAYLOAD) {
+                m_rx_pg_stat_payload[hw_id].clear();
+            } else {
+                m_rx_pg_stat[hw_id].clear();
+            }
+        }
+    }
+}
+
+
 Json::Value
 RXLatency::to_json() const {
     return Json::objectValue;
diff --git a/src/stateless/rx/trex_stateless_rx_port_mngr.h b/src/stateless/rx/trex_stateless_rx_port_mngr.h
index bc34b5aa..6af90f8b 100644
--- a/src/stateless/rx/trex_stateless_rx_port_mngr.h
+++ b/src/stateless/rx/trex_stateless_rx_port_mngr.h
@@ -43,12 +43,18 @@ public:
 
     void create(CRFC2544Info *rfc2544, CRxCoreErrCntrs *err_cntrs);
 
-    void reset_stats();
-
     void handle_pkt(const rte_mbuf_t *m);
 
     Json::Value to_json() const;
     
+    void get_stats(rx_per_flow_t *rx_stats,
+                   int min,
+                   int max,
+                   bool reset,
+                   TrexPlatformApi::driver_stat_cap_e type);
+    
+    void reset_stats();
+    
 private:
     bool is_flow_stat_id(uint32_t id) {
         if ((id & 0x000fff00) == IP_ID_RESERVE_BASE) return true;
@@ -276,6 +282,15 @@ public:
         m_latency.reset_stats();
     }
 
+    void get_latency_stats(rx_per_flow_t *rx_stats,
+                           int min,
+                           int max,
+                           bool reset,
+                           TrexPlatformApi::driver_stat_cap_e type) {
+        
+        return m_latency.get_stats(rx_stats, min, max, reset, type);
+    }
+    
     RXLatency & get_latency() {
         return m_latency;
     }