1 files changed, 59 insertions, 10 deletions

diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp
index ee408c63..1f415958 100644
--- a/src/main_dpdk.cpp
+++ b/src/main_dpdk.cpp
@@ -148,6 +148,7 @@ public:
     virtual int dump_fdir_global_stats(CPhyEthIF * _if, FILE *fd) { return -1;}
     virtual int get_stat_counters_num() {return 0;}
     virtual int get_rx_stat_capabilities() {return 0;}
+    virtual CFlowStatParser *get_flow_stat_parser();
 };
 
 
@@ -281,6 +282,7 @@ public:
     virtual int wait_for_stable_link();
     virtual int get_stat_counters_num() {return MAX_FLOW_STATS;}
     virtual int get_rx_stat_capabilities() {return TrexPlatformApi::IF_STAT_IPV4_ID;}
+    virtual CFlowStatParser *get_flow_stat_parser();
 };
 
 class CTRexExtendedDriverBase40G : public CTRexExtendedDriverBase10G {
@@ -332,9 +334,12 @@ public:
     // disabling flow control on 40G using DPDK API causes the interface to malfunction
     virtual bool flow_control_disable_supported(){return false;}
     virtual bool hw_rx_stat_supported(){return true;}
+    virtual CFlowStatParser *get_flow_stat_parser();
+
 private:
     virtual void add_del_rules(enum rte_filter_op op, uint8_t port_id, uint16_t type, uint8_t ttl, uint16_t ip_id, int queue, uint16_t stat_idx);
     virtual int configure_rx_filter_rules_statfull(CPhyEthIF * _if);
+
 private:
     uint8_t m_if_per_card;
 };
@@ -1231,6 +1236,10 @@ void CPhyEthIFStats::Dump(FILE *fd){
     DP_A(rx_nombuf);
 }
 
+// only on VM we have rx queues on DP cores
+void CPhyEthIF::flush_dp_rx_queue(void) {
+}
+
 // Clear the RX queue of an interface, dropping all packets
 void CPhyEthIF::flush_rx_queue(void){
 
@@ -1735,8 +1744,8 @@ public:
     virtual int send_node(CGenNode * node);
     virtual void send_one_pkt(pkt_dir_t       dir, rte_mbuf_t      *m);
 
+    virtual void flush_dp_rx_queue(void);
     virtual int flush_tx_queue(void);
-
     __attribute__ ((noinline)) void flush_rx_queue();
     __attribute__ ((noinline)) void update_mac_addr(CGenNode * node,uint8_t *p);
 
@@ -1804,6 +1813,11 @@ bool CCoreEthIF::Create(uint8_t             core_id,
     return (true);
 }
 
+// On VM, we get the packets in dp core, so just call general flush_rx_queue
+void CCoreEthIF::flush_dp_rx_queue(void) {
+    flush_rx_queue();
+}
+
 // 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, and pass queue 1 to RX core, like in other cases.
 // We receive all packets in the same core that transmitted, and handle them to RX core.
@@ -2699,7 +2713,7 @@ public:
     CFlowGenList  m_fl;
     bool          m_fl_was_init;
     volatile uint8_t       m_signal[BP_MAX_CORES] __rte_cache_aligned ; // Signal to main core when DP thread finished
-    volatile bool m_rx_running; // Signal main core when RX thread finished
+    volatile bool m_sl_rx_running; // Signal main core when RX thread finished
     CLatencyManager     m_mg; // statefull RX core
     CRxCoreStateless    m_rx_sl; // stateless RX core
     CTrexGlobalIoMode   m_io_modes;
@@ -2793,7 +2807,9 @@ void CGlobalTRex::try_stop_all_cores(){
     TrexStatelessDpQuit * dp_msg= new TrexStatelessDpQuit();
     TrexStatelessRxQuit * rx_msg= new TrexStatelessRxQuit();
     send_message_all_dp(dp_msg);
-    send_message_to_rx(rx_msg);
+    if (get_is_stateless()) {
+        send_message_to_rx(rx_msg);
+    }
     delete dp_msg;
     // no need to delete rx_msg. Deleted by receiver
     bool all_core_finished = false;
@@ -3804,16 +3820,16 @@ int CGlobalTRex::run_in_master() {
 
 int CGlobalTRex::run_in_rx_core(void){
     if (get_is_stateless()) {
-        m_rx_running = true;
+        m_sl_rx_running = true;
         m_rx_sl.start();
+        m_sl_rx_running = false;
     } else {
         if ( CGlobalInfo::m_options.is_rx_enabled() ){
-            m_rx_running = true;
+            m_sl_rx_running = false;
             m_mg.start(0);
         }
     }
 
-    m_rx_running = false;
     return (0);
 }
 
@@ -3905,7 +3921,7 @@ bool CGlobalTRex::is_all_cores_finished() {
             return false;
         }
     }
-    if (m_rx_running)
+    if (m_sl_rx_running)
         return false;
 
     return true;
@@ -4116,11 +4132,14 @@ bool CCoreEthIF::process_rx_pkt(pkt_dir_t   dir,
     return (send);
 }
 
-
 TrexStateless * get_stateless_obj() {
     return g_trex.m_trex_stateless;
 }
 
+CRxCoreStateless * get_rx_sl_core_obj() {
+    return &g_trex.m_rx_sl;
+}
+
 static int latency_one_lcore(__attribute__((unused)) void *dummy)
 {
     CPlatformSocketInfo * lpsock=&CGlobalInfo::m_socket;
@@ -4274,12 +4293,19 @@ int core_mask_sanity(uint32_t wanted_core_mask) {
     wanted_core_num = num_set_bits(wanted_core_mask);
     calc_core_num = num_set_bits(calc_core_mask);
 
+    if (calc_core_num == 1) {
+        printf ("Error: You have only 1 core available. Minimum configuration requires 2 cores\n");
+        printf("        If you are running on VM, consider adding more cores if possible\n");
+        return -1;
+    }
     if (wanted_core_num > calc_core_num) {
         printf("Error: You have %d threads available, but you asked for %d threads.\n", calc_core_num, wanted_core_num);
         printf("       Calculation is: -c <num>(%d) * dual ports (%d) + 1 master thread %s"
                , CGlobalInfo::m_options.preview.getCores(), CGlobalInfo::m_options.get_expected_dual_ports()
                , get_is_rx_thread_enabled() ? "+1 latency thread (because of -l flag)\n" : "\n");
-        printf("       Maybe try smaller -c <num>.\n");
+        if (CGlobalInfo::m_options.preview.getCores() > 1)
+            printf("       Maybe try smaller -c <num>.\n");
+        printf("       If you are running on VM, consider adding more cores if possible\n");
         return -1;
     }
 
@@ -4483,7 +4509,7 @@ int main_test(int argc , char * argv[]){
         g_trex.reset_counters();
     }
 
-    g_trex.m_rx_running = false;
+    g_trex.m_sl_rx_running = false;
     if ( get_is_stateless() ) {
         g_trex.start_master_stateless();
 
@@ -4537,6 +4563,12 @@ int CTRexExtendedDriverBase::configure_drop_queue(CPhyEthIF * _if) {
     return (rte_eth_dev_rx_queue_stop(port_id, 0));
 }
 
+CFlowStatParser *CTRexExtendedDriverBase::get_flow_stat_parser() {
+    CFlowStatParser *parser = new CFlowStatParser();
+    assert (parser);
+    return parser;
+}
+
 void wait_x_sec(int sec) {
     int i;
     printf(" wait %d sec ", sec);
@@ -4940,6 +4972,12 @@ int CTRexExtendedDriverBase10G::wait_for_stable_link(){
     return (0);
 }
 
+CFlowStatParser *CTRexExtendedDriverBase10G::get_flow_stat_parser() {
+    CFlowStatParser *parser = new C82599Parser();
+    assert (parser);
+    return parser;
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 void CTRexExtendedDriverBase40G::clear_extended_stats(CPhyEthIF * _if){
     rte_eth_stats_reset(_if->get_port_id());
@@ -5167,6 +5205,12 @@ int CTRexExtendedDriverBase40G::wait_for_stable_link(){
     return (0);
 }
 
+CFlowStatParser *CTRexExtendedDriverBase40G::get_flow_stat_parser() {
+    CFlowStatParser *parser = new CFlowStatParser();
+    assert (parser);
+    return parser;
+}
+
 /////////////////////////////////////////////////////////////////////
 
 
@@ -5407,3 +5451,8 @@ void TrexDpdkPlatformApi::flush_dp_messages() const {
 int TrexDpdkPlatformApi::get_active_pgids(flow_stat_active_t &result) const {
     return g_trex.m_trex_stateless->m_rx_flow_stat.get_active_pgids(result);
 }
+
+CFlowStatParser *TrexDpdkPlatformApi::get_flow_stat_parser() const {
+    return CTRexExtendedDriverDb::Ins()->get_drv()
+        ->get_flow_stat_parser();
+}