Make RX core in VM read packets directly instead of getting message from TX core

Signed-off-by: Ido Barnea <[email protected]>

diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp
index a003334..095052e 100644 (file)
--- a/src/main_dpdk.cpp
+++ b/src/main_dpdk.cpp
@@ -1095,7 +1095,7 @@ static int parse_options(int argc, char *argv[], CParserOption* po, bool first_t
     }
 
     if (po->preview.get_is_rx_check_enable() ||  po->is_latency_enabled() || CGlobalInfo::is_learn_mode()
-        || (CGlobalInfo::m_options.m_arp_ref_per != 0)) {
+        || (CGlobalInfo::m_options.m_arp_ref_per != 0) || get_vm_one_queue_enable()) {
         po->set_rx_enabled();
     }
 
@@ -1986,7 +1986,6 @@ public:
     virtual int send_node(CGenNode * node);
     virtual void send_one_pkt(pkt_dir_t dir, rte_mbuf_t *m);
     virtual int flush_tx_queue(void);
-    __attribute__ ((noinline)) void handle_rx_queue();
     __attribute__ ((noinline)) void handle_slowpath_features(CGenNode *node, rte_mbuf_t *m, uint8_t *p, pkt_dir_t dir);
 
     void apply_client_cfg(const ClientCfgBase *cfg, rte_mbuf_t *m, pkt_dir_t dir, uint8_t *p);
@@ -2062,46 +2061,6 @@ bool CCoreEthIF::Create(uint8_t             core_id,
     return (true);
 }
 
-// This function is only relevant if we are in VM. In this case, we only have one rx queue. Can't have
-// rules to drop queue 0 packets, and pass queue 1 packets to RX core, like in other cases.
-// We receive all packets in the same core that transmitted, and handle them to RX core.
-void CCoreEthIF::handle_rx_queue(void) {
-    if ( likely( ! get_vm_one_queue_enable() ) ) {
-        return;
-    }
-
-    pkt_dir_t dir;
-    bool is_rx = get_is_rx_thread_enabled();
-    for (dir=CLIENT_SIDE; dir<CS_NUM; dir++) {
-        CCorePerPort * lp_port=&m_ports[dir];
-        CPhyEthIF * lp=lp_port->m_port;
-
-        rte_mbuf_t * rx_pkts[32];
-        int j=0;
-
-        while (true) {
-            j++;
-            uint16_t cnt =lp->rx_burst(0,rx_pkts,32);
-            if ( cnt ) {
-                int i;
-                for (i=0; i<(int)cnt;i++) {
-                    rte_mbuf_t * m=rx_pkts[i];
-                    if ( is_rx ){
-                        if (!process_rx_pkt(dir,m)){
-                            rte_pktmbuf_free(m);
-                        }
-                    }else{
-                        rte_pktmbuf_free(m);
-                    }
-                }
-            }
-            if ((cnt<5) || j>10 ) {
-                break;
-            }
-        }
-    }
-}
-
 int CCoreEthIF::flush_tx_queue(void){
     /* flush both sides */
     pkt_dir_t dir;
@@ -2114,8 +2073,6 @@ int CCoreEthIF::flush_tx_queue(void){
         }
     }
 
-    handle_rx_queue();
-
     return 0;
 }
 
@@ -2605,10 +2562,11 @@ private:
 class CLatencyVmPort : public CPortLatencyHWBase {
 public:
     void Create(uint8_t port_index,CNodeRing * ring,
-                CLatencyManager * mgr){
+                CLatencyManager * mgr, CPhyEthIF  * p) {
         m_dir        = (port_index%2);
         m_ring_to_dp = ring;
         m_mgr        = mgr;
+        m_port = p;
     }
 
     virtual int tx(rte_mbuf_t * m){
@@ -2635,17 +2593,23 @@ public:
         return (-1);
     }
 
-    virtual rte_mbuf_t * rx(){
-        return (0);
+    virtual rte_mbuf_t * rx() {
+        rte_mbuf_t * rx_pkts[1];
+        uint16_t cnt = m_port->rx_burst(0, rx_pkts, 1);
+        if (cnt) {
+            return (rx_pkts[0]);
+        } else {
+            return (0);
+        }
     }
 
-    virtual uint16_t rx_burst(struct rte_mbuf **rx_pkts,
-                              uint16_t nb_pkts){
-        return (0);
+    virtual uint16_t rx_burst(struct rte_mbuf **rx_pkts, uint16_t nb_pkts) {
+        uint16_t cnt = m_port->rx_burst(0, rx_pkts, nb_pkts);
+        return (cnt);
     }
 
-
 private:
+    CPhyEthIF  * m_port;
     uint8_t                          m_dir;
     CNodeRing *                      m_ring_to_dp;   /* ring dp -> latency thread */
     CLatencyManager *                m_mgr;
@@ -3530,13 +3494,13 @@ void CGlobalTRex::ixgbe_configure_mg(void) {
     if ( get_vm_one_queue_enable() ) {
         /* vm mode, indirect queues  */
         for (i=0; i<m_max_ports; i++) {
-
+            CPhyEthIF * _if = &m_ports[i];
             CMessagingManager * rx_dp=CMsgIns::Ins()->getRxDp();
 
             uint8_t thread_id = (i>>1);
 
             CNodeRing * r = rx_dp->getRingCpToDp(thread_id);
-            m_latency_vm_vports[i].Create((uint8_t)i,r,&m_mg);
+            m_latency_vm_vports[i].Create((uint8_t)i, r, &m_mg, _if);
 
             mg_cfg.m_ports[i] =&m_latency_vm_vports[i];
         }
@@ -3566,10 +3530,11 @@ void CGlobalTRex::rx_sl_configure(void) {
     if ( get_vm_one_queue_enable() ) {
         /* vm mode, indirect queues  */
         for (i=0; i < m_max_ports; i++) {
+            CPhyEthIF * _if = &m_ports[i];
             CMessagingManager * rx_dp = CMsgIns::Ins()->getRxDp();
             uint8_t thread_id = (i >> 1);
             CNodeRing * r = rx_dp->getRingCpToDp(thread_id);
-            m_latency_vm_vports[i].Create((uint8_t)i, r, &m_mg);
+            m_latency_vm_vports[i].Create(i, r, &m_mg, _if);
             rx_sl_cfg.m_ports[i] = &m_latency_vm_vports[i];
         }
     } else {

diff --git a/src/stateful_rx_core.cpp b/src/stateful_rx_core.cpp
index 7ee802d..65d0a17 100644 (file)
--- a/src/stateful_rx_core.cpp
+++ b/src/stateful_rx_core.cpp
@@ -677,58 +677,6 @@ void CLatencyManager::handle_rx_pkt(CLatencyManagerPerPort * lp,
     rte_pktmbuf_free(m);
 }
 
-// In VM, we receive the RX packets in DP core, and send message to RX core with the packet
-void CLatencyManager::handle_latency_pkt_msg(uint8_t thread_id, CGenNodeLatencyPktInfo * msg) {
-
-    assert(msg->m_latency_offset==0xdead);
-
-    uint8_t rx_port_index=(thread_id<<1)+(msg->m_dir&1);
-    assert( rx_port_index <m_max_ports ) ;
-    CLatencyManagerPerPort * lp=&m_ports[rx_port_index];
-    handle_rx_pkt(lp,(rte_mbuf_t *)msg->m_pkt);
-}
-
-
-void  CLatencyManager::run_rx_queue_msgs(uint8_t thread_id,
-                                         CNodeRing * r){
-
-    while ( true ) {
-        CGenNode * node;
-        if ( r->Dequeue(node)!=0 ){
-            break;
-        }
-        assert(node);
-
-        CGenNodeMsgBase * msg=(CGenNodeMsgBase *)node;
-
-        uint8_t   msg_type =  msg->m_msg_type;
-        switch (msg_type ) {
-        case CGenNodeMsgBase::LATENCY_PKT:
-            handle_latency_pkt_msg(thread_id,(CGenNodeLatencyPktInfo *) msg);
-            break;
-        default:
-            printf("ERROR latency-thread message type is not valid %d \n",msg_type);
-            assert(0);
-        }
-
-        CGlobalInfo::free_node(node);
-    }
-}
-
-// VM mode function. Handle messages from DP
-void  CLatencyManager::try_rx_queues(){
-
-    CMessagingManager * rx_dp = CMsgIns::Ins()->getRxDp();
-    uint8_t threads=CMsgIns::Ins()->get_num_threads();
-    int ti;
-    for (ti=0; ti<(int)threads; ti++) {
-        CNodeRing * r = rx_dp->getRingDpToCp(ti);
-        if ( !r->isEmpty() ){
-            run_rx_queue_msgs((uint8_t)ti,r);
-        }
-    }
-}
-
 void  CLatencyManager::try_rx(){
     rte_mbuf_t * rx_pkts[64];
     int i;
@@ -790,8 +738,6 @@ void  CLatencyManager::start(int iter, bool activate_watchdog) {
         m_p_queue.push(node);
     }
 
-    bool do_try_rx_queue = CGlobalInfo::m_options.preview.get_vm_one_queue_enable() ? true : false;
-
     if (activate_watchdog) {
         m_monitor.create("STF RX CORE", 1);
         TrexWatchDog::getInstance().register_monitor(&m_monitor);
@@ -807,9 +753,6 @@ void  CLatencyManager::start(int iter, bool activate_watchdog) {
             if (dt> (0.0)) {
                 break;
             }
-            if (do_try_rx_queue){
-                try_rx_queues();
-            }
             try_rx();
             rte_pause();
         }

diff --git a/src/stateful_rx_core.h b/src/stateful_rx_core.h
index 48fbeb9..2df406d 100644 (file)
--- a/src/stateful_rx_core.h
+++ b/src/stateful_rx_core.h
@@ -355,12 +355,8 @@ private:
     double grat_arp_timeout();
     void  send_one_grat_arp();
     void  try_rx();
-    void  try_rx_queues();
-    void  run_rx_queue_msgs(uint8_t thread_id, CNodeRing * r);
     void  wait_for_rx_dump();
     void  handle_rx_pkt(CLatencyManagerPerPort * lp, rte_mbuf_t * m);
-    /* messages handlers */
-    void handle_latency_pkt_msg(uint8_t thread_id, CGenNodeLatencyPktInfo * msg);
 
  private:
      pqueue_t                m_p_queue; /* priorty queue */