Enabled flow stats for all interface types + needed corrections

13 files changed, 293 insertions, 230 deletions

diff --git a/scripts/automation/regression/setups/trex09/config.yaml b/scripts/automation/regression/setups/trex09/config.yaml
index e0a2227e..585ca17a 100644
--- a/scripts/automation/regression/setups/trex09/config.yaml
+++ b/scripts/automation/regression/setups/trex09/config.yaml
@@ -33,6 +33,6 @@
 # expected_bw    - the "golden" bandwidth (in Gbps) results planned on receiving from the test
 
 trex:
-  hostname       : csi-trex-08
+  hostname       : csi-trex-09
   cores          : 2
   modes          : ['loopback']
diff --git a/scripts/automation/regression/stateless_tests/stl_rx_test.py b/scripts/automation/regression/stateless_tests/stl_rx_test.py
index 49c53855..cf346718 100644
--- a/scripts/automation/regression/stateless_tests/stl_rx_test.py
+++ b/scripts/automation/regression/stateless_tests/stl_rx_test.py
@@ -7,6 +7,9 @@ class STLRX_Test(CStlGeneral_Test):
     """Tests for RX feature"""
 
     def setUp(self):
+        per_driver_params = {"rte_vmxnet3_pmd": [1, 50], "rte_ixgbe_pmd": [30, 5000], "rte_i40e_pmd": [80, 5000],
+                        "rte_igb_pmd": [80, 500]}
+
         CStlGeneral_Test.setUp(self)
         assert 'bi' in CTRexScenario.stl_ports_map
 
@@ -19,12 +22,8 @@ class STLRX_Test(CStlGeneral_Test):
         if cap != 1:
             self.skip('port {0} does not support RX'.format(self.rx_port))
 
-        if port_info['speed'] == 40:
-            self.rate_percent = 80
-            self.total_pkts = 50000
-        else:
-            self.rate_percent = 1
-            self.total_pkts = 10
+        self.rate_percent = per_driver_params[port_info['driver']][0]
+        self.total_pkts = per_driver_params[port_info['driver']][1]
         self.c.reset(ports = [self.tx_port, self.rx_port])
 
         self.pkt = STLPktBuilder(pkt = Ether()/IP(src="16.0.0.1",dst="48.0.0.1")/UDP(dport=12,sport=1025)/IP()/'a_payload_example')
@@ -100,21 +99,26 @@ class STLRX_Test(CStlGeneral_Test):
 
     # one simple stream on TX --> RX
     def test_multiple_streams(self):
-        total_pkts = self.total_pkts * 10
+        num_streams = 10
+        total_pkts = self.total_pkts / num_streams
+        if total_pkts == 0:
+            total_pkts = 1
+        percent = self.rate_percent / num_streams
+        if percent == 0:
+            percent = 1
 
         try:
             streams = []
             exp = []
             # 10 identical streams
-            for pg_id in range(1, 10):
+            for pg_id in range(1, num_streams):
 
                 streams.append(STLStream(name = 'rx {0}'.format(pg_id),
                                          packet = self.pkt,
                                          flow_stats = STLFlowStats(pg_id = pg_id),
-                                         mode = STLTXSingleBurst(total_pkts = total_pkts * pg_id,
-                                                                 pps = total_pkts * pg_id)))
+                                         mode = STLTXSingleBurst(total_pkts = total_pkts+pg_id, percentage = percent)))
 
-                exp.append({'pg_id': pg_id, 'total_pkts': total_pkts * pg_id, 'pkt_len': self.pkt.get_pkt_len()})
+                exp.append({'pg_id': pg_id, 'total_pkts': total_pkts+pg_id, 'pkt_len': self.pkt.get_pkt_len()})
 
             # add both streams to ports
             self.c.add_streams(streams, ports = [self.tx_port])
diff --git a/src/bp_sim.h b/src/bp_sim.h
index 4b1a88e3..cd85e82b 100755
--- a/src/bp_sim.h
+++ b/src/bp_sim.h
@@ -41,7 +41,7 @@ limitations under the License.
 #include <common/Network/Packet/IPv6Header.h>
 #include <common/Network/Packet/EthernetHeader.h>
 #include <math.h>
-#include <common/bitMan.h> 
+#include <common/bitMan.h>
 #include <yaml-cpp/yaml.h>
 #include "trex_defs.h"
 #include "os_time.h"
@@ -97,7 +97,7 @@ public:
         MIN_VM_V6=1    // IPv6 addressing
     };
     uint8_t   m_cmd;
-    uint8_t   m_flags; 
+    uint8_t   m_flags;
     uint16_t  m_start_0;
     uint16_t  m_stop_1;
     uint16_t  m_add_pkt_len; /* request more length for mbuf packet the size */
@@ -116,16 +116,16 @@ public:
     uint16_t  m_server_port;
 };
 
-/* this command replace IP in 2 diffrent location and port 
+/* this command replace IP in 2 diffrent location and port
 
-c =  10.1.1.2 
-o =  10.1.1.2 
+c =  10.1.1.2
+o =  10.1.1.2
 m = audio 102000
 
 ==>
 
-c =  xx.xx.xx.xx 
-o =  xx.xx.xx.xx 
+c =  xx.xx.xx.xx
+o =  xx.xx.xx.xx
 m = audio yyyy
 
 */
@@ -248,7 +248,7 @@ class CFlowGenListPerThread ;
 
 /* callback */
 void on_node_first(uint8_t plugin_id,CGenNode *     node,
-                   CFlowYamlInfo *  template_info, 
+                   CFlowYamlInfo *  template_info,
                    CTupleTemplateGeneratorSmart * tuple_gen,
                    CFlowGenListPerThread  * flow_gen
                    );
@@ -259,7 +259,7 @@ rte_mbuf_t * on_node_generate_mbuf(uint8_t plugin_id,CGenNode *     node,CFlowPk
 
 class CPreviewMode ;
 struct CGenNode;
-/* represent the virtual interface 
+/* represent the virtual interface
 */
 
 /* counters per side */
@@ -276,7 +276,7 @@ public:
     uint64_t   m_tx_drop;
     uint64_t   m_tx_queue_full;
     uint64_t   m_tx_alloc_error;
-    tx_per_flow_t m_tx_per_flow[MAX_FLOW_STATS];    
+    tx_per_flow_t m_tx_per_flow[MAX_FLOW_STATS];
     CPerTxthreadTemplateInfo m_template;
 
 public:
@@ -309,10 +309,10 @@ public:
 void CVirtualIFPerSideStats::Dump(FILE *fd){
 
     #define DP_B(f) if (f) printf(" %-40s : %lu \n",#f,f)
-    DP_B(m_tx_pkt);  
+    DP_B(m_tx_pkt);
     DP_B(m_tx_rx_check_pkt);
-    DP_B(m_tx_bytes);  
-    DP_B(m_tx_drop);  
+    DP_B(m_tx_bytes);
+    DP_B(m_tx_drop);
     DP_B(m_tx_alloc_error);
     DP_B(m_tx_queue_full);
     m_template.Dump(fd);
@@ -342,17 +342,17 @@ public:
 
     /**
      * send one packet
-     * 
+     *
      * @param node
-     * 
-     * @return 
+     *
+     * @return
      */
     virtual int send_node(CGenNode * node) =0;
 
 
     /**
      * send one packet to a specific dir. flush all packets
-     * 
+     *
      * @param dir
      * @param m
      */
@@ -361,26 +361,29 @@ public:
 
 
     /**
-     * flush all pending packets into the stream 
-     * 
-     * @return 
+     * flush all pending packets into the stream
+     *
+     * @return
      */
     virtual int flush_tx_queue(void)=0;
-
+    // read all packets from rx_queue on dp core
+    virtual void flush_dp_rx_queue(void) {};
+    // read all packets from rx queue
+    virtual void flush_rx_queue(void) {};
 
     /**
      * update the source and destination mac-addr of a given mbuf by global database
-     * 
+     *
      * @param dir
      * @param m
-     * 
-     * @return 
+     *
+     * @return
      */
     virtual int update_mac_addr_from_global_cfg(pkt_dir_t       dir, uint8_t * p)=0;
 
     /**
      * translate a port_id to the correct dir on the core
-     * 
+     *
      */
     virtual pkt_dir_t port_id_to_dir(uint8_t port_id) {
         return (CS_INVALID);
@@ -602,13 +605,13 @@ public:
         }
     }
 
-	bool get_is_rx_check_enable(){
-		return (btGetMaskBit32(m_flags,31,31) ? true:false);
-	}
+    bool get_is_rx_check_enable(){
+        return (btGetMaskBit32(m_flags,31,31) ? true:false);
+    }
 
-	void set_rx_check_enable(bool enable){
-		btSetMaskBit32(m_flags,31,31,enable?1:0);
-	}
+    void set_rx_check_enable(bool enable){
+        btSetMaskBit32(m_flags,31,31,enable?1:0);
+    }
 
 
     bool get_mac_ip_features_enable(){
@@ -693,7 +696,7 @@ public:
         u.m_mac.dest[3]=1;
         u.m_mac.src[3]=1;
     }
-    union {  
+    union {
         mac_align_t m_mac;
         uint8_t     m_data[16];
     } u;
@@ -717,10 +720,10 @@ public:
     };
 
     enum trex_learn_mode_e {
-	LEARN_MODE_DISABLED=0,
-	LEARN_MODE_TCP_ACK=1,
-	LEARN_MODE_IP_OPTION=2,
-	LEARN_MODE_MAX=LEARN_MODE_IP_OPTION
+    LEARN_MODE_DISABLED=0,
+    LEARN_MODE_TCP_ACK=1,
+    LEARN_MODE_IP_OPTION=2,
+    LEARN_MODE_MAX=LEARN_MODE_IP_OPTION
     };
 
 public:
@@ -736,7 +739,7 @@ public:
         m_expected_portd = 4; /* should be at least the number of ports found in the system but could be less */
         m_vlan_port[0]=100;
         m_vlan_port[1]=100;
-		m_rx_check_sample=0;
+        m_rx_check_sample=0;
         m_rx_check_hops = 0;
         m_io_mode=1;
         m_run_flags=0;
@@ -759,12 +762,12 @@ public:
     uint32_t        m_latency_rate; /* pkt/sec for each thread/port zero disable */
     uint32_t        m_latency_mask;
     uint32_t        m_latency_prev;
-    uint16_t 		m_rx_check_sample; /* the sample rate of flows */
+    uint16_t        m_rx_check_sample; /* the sample rate of flows */
     uint16_t        m_rx_check_hops;
     uint16_t        m_zmq_port;
     uint16_t        m_telnet_port;
     uint16_t        m_expected_portd;
-    uint16_t        m_io_mode; //0,1,2 0 disable, 1- normal , 2 - short 
+    uint16_t        m_io_mode; //0,1,2 0 disable, 1- normal , 2 - short
     uint16_t        m_run_flags;
     uint8_t         m_mac_splitter;
     uint8_t         m_l_pkt_mode;
@@ -782,7 +785,7 @@ public:
     std::string        out_file;
     std::string        prefix;
 
-                                 
+
     CMacAddrCfg     m_mac_addr[TREX_MAX_PORTS];
 
     uint8_t *       get_src_mac_addr(int if_index){
@@ -861,7 +864,7 @@ public:
     void Dump(FILE *fd);
 
 public:
-    uint32_t         m_mbuf[MBUF_SIZE]; // relative to traffic norm to 2x10G ports 
+    uint32_t         m_mbuf[MBUF_SIZE]; // relative to traffic norm to 2x10G ports
     uint32_t         m_num_cores;
 
 };
@@ -869,28 +872,28 @@ public:
 typedef uint8_t socket_id_t;
 typedef uint8_t port_id_t;
 /* the real phsical thread id */
-typedef uint8_t physical_thread_id_t; 
+typedef uint8_t physical_thread_id_t;
 
 
-typedef uint8_t virtual_thread_id_t; 
-/* 
+typedef uint8_t virtual_thread_id_t;
+/*
+
+ virtual thread 0 (v0)- is always the master
 
- virtual thread 0 (v0)- is always the master 
-  
-for 2 dual ports ( 2x2 =4 ports) the virtual thread looks like that                      
+for 2 dual ports ( 2x2 =4 ports) the virtual thread looks like that
 -----------------
 DEFAULT:
 -----------------
   (0,1)       (2,3)
   dual-if0       dual-if-1
     v1        v2
-    v3        v4     
+    v3        v4
     v5        v6
-    v7        v8       
-       
-    rx is v9          
+    v7        v8
+
+    rx is v9
 
-  */  
+  */
 
 #define MAX_SOCKETS_SUPPORTED   (4)
 #define MAX_THREADS_SUPPORTED   (120)
@@ -904,12 +907,12 @@ public:
 
     /* is socket enabled */
     virtual bool is_sockets_enable(socket_id_t socket)=0;
-    
+
     /* number of main active sockets. socket #0 is always used  */
     virtual socket_id_t max_num_active_sockets()=0;
 
     virtual ~CPlatformSocketInfoBase() {}
-    
+
 public:
     /* which socket to allocate memory to each port */
     virtual socket_id_t port_to_socket(port_id_t port)=0;
@@ -949,7 +952,7 @@ public:
 
     /* is socket enabled */
     bool is_sockets_enable(socket_id_t socket);
-    
+
     /* number of main active sockets. socket #0 is always used  */
     socket_id_t max_num_active_sockets();
 
@@ -995,7 +998,7 @@ public:
 
         /* is socket enabled */
     bool is_sockets_enable(socket_id_t socket);
-    
+
     /* number of main active sockets. socket #0 is always used  */
     socket_id_t max_num_active_sockets();
 
@@ -1033,7 +1036,7 @@ private:
     bool                     m_sockets_enable[MAX_SOCKETS_SUPPORTED];
     uint32_t                 m_sockets_enabled;
     socket_id_t              m_socket_per_dual_if[(TREX_MAX_PORTS >> 1)];
-    
+
     uint32_t                 m_max_threads_per_dual_if;
 
     uint32_t                 m_num_dual_if;
@@ -1058,7 +1061,7 @@ public:
 
     /* is socket enabled */
     bool is_sockets_enable(socket_id_t socket);
-    
+
     /* number of main active sockets. socket #0 is always used  */
     socket_id_t max_num_active_sockets();
 
@@ -1141,15 +1144,15 @@ public:
 public:
     rte_mempool_t *   m_small_mbuf_pool; /* pool for start packets */
 
-    rte_mempool_t *   m_mbuf_pool_128;  
-    rte_mempool_t *   m_mbuf_pool_256;  
-    rte_mempool_t *   m_mbuf_pool_512;  
-    rte_mempool_t *   m_mbuf_pool_1024;  
-    rte_mempool_t *   m_mbuf_pool_2048;  
-    rte_mempool_t *   m_mbuf_pool_4096;  
-    rte_mempool_t *   m_mbuf_pool_9k;  
+    rte_mempool_t *   m_mbuf_pool_128;
+    rte_mempool_t *   m_mbuf_pool_256;
+    rte_mempool_t *   m_mbuf_pool_512;
+    rte_mempool_t *   m_mbuf_pool_1024;
+    rte_mempool_t *   m_mbuf_pool_2048;
+    rte_mempool_t *   m_mbuf_pool_4096;
+    rte_mempool_t *   m_mbuf_pool_9k;
 
-    rte_mempool_t *   m_mbuf_global_nodes;  
+    rte_mempool_t *   m_mbuf_global_nodes;
     uint32_t          m_pool_id;
 };
 
@@ -1167,16 +1170,16 @@ public:
         return ( m_mem_pool[socket].pktmbuf_alloc_small() );
     }
 
-    
+
 
     /**
-     * try to allocate small buffers too 
-     * _alloc allocate big buffers only 
-     * 
+     * try to allocate small buffers too
+     * _alloc allocate big buffers only
+     *
      * @param socket
      * @param size
-     * 
-     * @return 
+     *
+     * @return
      */
     static inline rte_mbuf_t   * pktmbuf_alloc(socket_id_t socket,uint16_t size){
         if (size<FIRST_PKT_SIZE) {
@@ -1232,7 +1235,7 @@ public:
 public:
     static CRteMemPool       m_mem_pool[MAX_SOCKETS_SUPPORTED];
 
-    static uint32_t              m_nodes_pool_size;     
+    static uint32_t              m_nodes_pool_size;
     static CParserOption         m_options;
     static CGlobalMemory         m_memory_cfg;
     static CPlatformSocketInfo   m_socket;
@@ -1320,19 +1323,19 @@ struct CFlowYamlInfo {
         m_server_pool_idx = 0;
         m_cap_mode=false;
     }
-    
+
     std::string     m_name;
     std::string     m_client_pool_name;
     std::string     m_server_pool_name;
-    double          m_k_cps;    //k CPS 
-	double          m_restart_time; /* restart time of this template */
-    dsec_t          m_ipg_sec;   // ipg in sec 
+    double          m_k_cps;    //k CPS
+    double          m_restart_time; /* restart time of this template */
+    dsec_t          m_ipg_sec;   // ipg in sec
     dsec_t          m_rtt_sec;   // rtt in sec
-    uint32_t        m_w;    
+    uint32_t        m_w;
     uint32_t        m_wlength;
     uint32_t        m_limit;
     uint32_t        m_flowcnt;
-    uint8_t         m_plugin_id; /* 0 - default , 1 - RTSP160 , 2- RTSP250 */       
+    uint8_t         m_plugin_id; /* 0 - default , 1 - RTSP160 , 2- RTSP250 */
     uint8_t         m_client_pool_idx;
     uint8_t         m_server_pool_idx;
     bool            m_one_app_server;
@@ -1418,7 +1421,7 @@ public:
         NODE_FLAGS_LEARN_MSG_PROCESSED  =0x10,   /* got NAT msg */
 
         NODE_FLAGS_LATENCY              =0x20,   /* got NAT msg */
-        NODE_FLAGS_INIT_START_FROM_SERVER_SIDE = 0x40,  
+        NODE_FLAGS_INIT_START_FROM_SERVER_SIDE = 0x40,
         NODE_FLAGS_ALL_FLOW_SAME_PORT_SIDE     = 0x80,
         NODE_FLAGS_INIT_START_FROM_SERVER_SIDE_SERVER_ADDR = 0x100 /* init packet start from server side with server addr */
     };
@@ -1434,8 +1437,8 @@ public:
 
     uint16_t            m_src_port;
     uint16_t            m_flags; /* BIT 0 - DIR ,
-                                    BIT 1 - mbug_cache 
-									BIT 2 - SAMPLE DUPLICATE */
+                                    BIT 1 - mbug_cache
+                                    BIT 2 - SAMPLE DUPLICATE */
 
     double              m_time;    /* can't change this header - size 16 bytes*/
 
@@ -1512,7 +1515,7 @@ public:
     /* is it possible to cache MBUF */
 
     inline void update_next_pkt_in_flow(void);
-    inline void reset_pkt_in_flow(void);    
+    inline void reset_pkt_in_flow(void);
     inline uint8_t get_plugin_id(void){
         return ( m_template_info->m_plugin_id);
     }
@@ -1567,8 +1570,8 @@ public:
     /* direction for TCP/UDP port */
     inline  pkt_dir_t cur_pkt_port_addr_dir();
     /* from which interface dir to get out */
-    inline  pkt_dir_t cur_interface_dir(); 
-    
+    inline  pkt_dir_t cur_interface_dir();
+
 
     inline void set_mbuf_cache_dir(pkt_dir_t  dir){
         if (dir) {
@@ -1597,19 +1600,19 @@ public:
 
 public:
 
-	inline void set_rx_check(){
-		m_flags |= NODE_FLAGS_SAMPLE_RX_CHECK;
-	}
+    inline void set_rx_check(){
+        m_flags |= NODE_FLAGS_SAMPLE_RX_CHECK;
+    }
 
-	inline bool is_rx_check_enabled(){
-		return ((m_flags & NODE_FLAGS_SAMPLE_RX_CHECK)?true:false);
-	}
+    inline bool is_rx_check_enabled(){
+        return ((m_flags & NODE_FLAGS_SAMPLE_RX_CHECK)?true:false);
+    }
 
 public:
 
     inline void set_nat_first_state(){
         btSetMaskBit16(m_flags,4,3,1);
-        m_type=FLOW_PKT_NAT; 
+        m_type=FLOW_PKT_NAT;
     }
 
     inline bool is_nat_first_state(){
@@ -1665,8 +1668,8 @@ public:
 
     bool is_external_is_eq_to_internal_ip(){
         /* this API is used to check TRex itself */
-        if ( (get_nat_ipv4_addr() == m_src_ip ) && 
-             (get_nat_ipv4_port()==m_src_port) && 
+        if ( (get_nat_ipv4_addr() == m_src_ip ) &&
+             (get_nat_ipv4_port()==m_src_port) &&
              ( get_nat_ipv4_addr_server() == m_dest_ip) ) {
             return (true);
         }else{
@@ -1704,7 +1707,7 @@ struct CGenNodeDeferPort  {
     uint16_t            m_ports[DEFER_CLIENTS_NUM];
     uint8_t             m_pool_idx[DEFER_CLIENTS_NUM];
 public:
-    void init(void){ 
+    void init(void){
         m_type=CGenNode::FLOW_DEFER_PORT_RELEASE;
         m_cnt=0;
     }
@@ -1724,7 +1727,7 @@ public:
 
 } __rte_cache_aligned ;
 
-/* run time verification of objects size and offsets 
+/* run time verification of objects size and offsets
    need to clean this up and derive this objects from base object but require too much refactoring right now
    hhaim
 */
@@ -1817,19 +1820,19 @@ public:
 
     /**
      * send one packet
-     * 
+     *
      * @param node
-     * 
-     * @return 
+     *
+     * @return
      */
     virtual int send_node(CGenNode * node);
 
 
 
     /**
-     * flush all pending packets into the stream 
-     * 
-     * @return 
+     * flush all pending packets into the stream
+     *
+     * @return
      */
     virtual int flush_tx_queue(void);
 
@@ -1858,7 +1861,7 @@ public:
 
 /**
  * same as regular STL but no I/O (dry run)
- * 
+ *
  * @author imarom (07-Jan-16)
  */
 class CErfIFStlNull : public CErfIFStl {
@@ -1959,7 +1962,7 @@ public:
     int   open_file(std::string file_name,
                     CPreviewMode * preview);
     int   close_file(CFlowGenListPerThread * thread);
-    int   flush_file(dsec_t max_time, 
+    int   flush_file(dsec_t max_time,
                      dsec_t d_time,
                      bool always,
                      CFlowGenListPerThread * thread,
@@ -2012,7 +2015,7 @@ public:
     CPreviewMode              m_preview_mode;
     uint64_t                  m_cnt;
     uint64_t                  m_limit;
-    CTimeHistogram            m_realtime_his;  
+    CTimeHistogram            m_realtime_his;
 };
 
 
@@ -2128,7 +2131,7 @@ inline bool CFlowKey::operator ==(const CFlowKey& rhs) const{
 
 #define IS_PCAP_TIMING 7
 
-// 8-12 is used 
+// 8-12 is used
 #define FLOW_ID 8
 
 
@@ -2164,9 +2167,9 @@ public:
     }
 
 private:
-    // per direction info 
+    // per direction info
     uint16_t    m_dir_pkt_num;   // pkt id
-    uint16_t    m_max_dir_flow_pkts;  
+    uint16_t    m_max_dir_flow_pkts;
 };
 
 
@@ -2219,7 +2222,7 @@ public:
     }
     /**
      * start from zero 0,1,2,.. , it is on global flow if you have couple of flows it will count all of the flows
-     * 
+     *
      * flow  FlowPktNum
      * 0         0
      * 0         1
@@ -2227,8 +2230,8 @@ public:
      * 1         0
      * 1         1
      * 2         0
-     * 
-     * @return 
+     *
+     * @return
      */
     inline uint32_t getFlowPktNum(){
         return ( m_flow_pkt_num);
@@ -2252,7 +2255,7 @@ public:
     }
 
 
-    /* return true if this packet in diff direction from prev flow packet , 
+    /* return true if this packet in diff direction from prev flow packet ,
     if true need to choose RTT else IPG for inter packet gap */
     inline bool IsRtt(){
         return (btGetMaskBit32(m_flags,IS_RTT,IS_RTT) ? true:false);
@@ -2313,7 +2316,7 @@ public:
 
     inline void SetId(uint16_t _id){
         btSetMaskBit32(m_flags,31,16,_id);
-        
+
     }
     inline uint16_t getId(){
         return ( ( uint16_t)btGetMaskBit32(m_flags,31,16));
@@ -2328,7 +2331,7 @@ public:
         return (btGetMaskBit32(m_flags,IS_LAST_PKT_S,IS_LAST_PKT_E) ? true:false);
     }
 
-    // there could be couple of flows per template in case of plugin 
+    // there could be couple of flows per template in case of plugin
     inline void SetMaxPktsPerFlow(uint32_t pkts){
         assert(pkts<65000);
         m_max_flow_pkts=pkts;
@@ -2336,7 +2339,7 @@ public:
     inline uint16_t GetMaxPktsPerFlow(){
         return ( m_max_flow_pkts );
     }
-        // there could be couple of flows per template in case of plugin 
+        // there could be couple of flows per template in case of plugin
     inline void SetMaxFlowTimeout(double sec){
         //assert (sec<65000);
         sec = sec*2.0+5.0;
@@ -2369,12 +2372,12 @@ public:
 
 private:
     uint32_t    m_flags;
-    uint16_t    m_flow_pkt_num; // packet number inside the flow 
-    uint8_t     m_plugin_id; // packet number inside the flow 
+    uint16_t    m_flow_pkt_num; // packet number inside the flow
+    uint8_t     m_plugin_id; // packet number inside the flow
     uint8_t     m_pad;
     uint16_t    m_max_flow_pkts;  // how many packet per this flow getFlowId()
-    uint16_t    m_max_flow_aging; // maximum aging in sec 
-    CPacketDescriptorPerDir  m_per_dir[CS_NUM]; // per direction info 
+    uint16_t    m_max_flow_aging; // maximum aging in sec
+    CPacketDescriptorPerDir  m_per_dir[CS_NUM]; // per direction info
 };
 
 
@@ -2427,7 +2430,7 @@ public:
 class CPacketIndication {
 
 public:
-    dsec_t       m_cap_ipg; /* ipg from cap file */  
+    dsec_t       m_cap_ipg; /* ipg from cap file */
     CCapPktRaw * m_packet;
 
     CFlow *          m_flow;
@@ -2437,10 +2440,10 @@ public:
         IPv6Header     * m_ipv6;
     } l3;
     bool        m_is_ipv6;
-    union {                             
+    union {
         TCPHeader * m_tcp;
         UDPHeader * m_udp;
-        ICMPHeader * m_icmp;      
+        ICMPHeader * m_icmp;
     } l4;
     uint8_t *       m_payload;
     uint16_t        m_payload_len;
@@ -2489,10 +2492,10 @@ public:
 
 
     /**
-     * return the application ipv4/ipv6 option offset 
+     * return the application ipv4/ipv6 option offset
      * if learn bit is ON , it is always the first options ( IPV6/IPV4)
-     * 
-     * @return 
+     *
+     * @return
      */
     uint32_t getIpAppOptionOffset(){
         if ( is_ipv6() ) {
@@ -2585,7 +2588,7 @@ class CPacketParser {
 public:
     bool Create();
     void Delete();
-    bool ProcessPacket(CPacketIndication * pkt_indication, 
+    bool ProcessPacket(CPacketIndication * pkt_indication,
                        CCapPktRaw * raw_packet);
 public:
     CCPacketParserCounters m_counter;
@@ -2706,12 +2709,12 @@ public:
     inline rte_mbuf_t * do_generate_new_mbuf_ex(CGenNode * node,CFlowInfo * flow_info);
     inline rte_mbuf_t * do_generate_new_mbuf_ex_big(CGenNode * node,CFlowInfo * flow_info);
     inline rte_mbuf_t * do_generate_new_mbuf_ex_vm(CGenNode * node,
-    								CFlowInfo * flow_info, int16_t * s_size);
+                                    CFlowInfo * flow_info, int16_t * s_size);
 
 public:
-    /* push the number of bytes into the packets and make more room 
-      should be used by NAT feature that should have ipv4 option in the first packet 
-      this function should not be called in runtime, only when template is loaded due to it heavey cost of operation ( malloc/free memory ) 
+    /* push the number of bytes into the packets and make more room
+      should be used by NAT feature that should have ipv4 option in the first packet
+      this function should not be called in runtime, only when template is loaded due to it heavey cost of operation ( malloc/free memory )
     */
     char * push_ipv4_option_offline(uint8_t bytes);
     char * push_ipv6_option_offline(uint8_t bytes);
@@ -2720,10 +2723,10 @@ public:
 
     /**
      * mark this packet as learn packet
-     * should 
+     * should
      * 1. push ipv4 option ( 8 bytes)
-     * 2. mask the packet as learn 
-     * 3. update the option pointer 
+     * 2. mask the packet as learn
+     * 3. update the option pointer
      */
     void   mask_as_learn();
 
@@ -2750,7 +2753,7 @@ private:
 
 public:
     CPacketIndication   m_pkt_indication;
-    CCapPktRaw        * m_packet; 
+    CCapPktRaw        * m_packet;
     rte_mbuf_t        * m_big_mbuf[MAX_SOCKETS_SUPPORTED]; /* allocate big mbug per socket */
 };
 
@@ -2764,10 +2767,10 @@ inline void CFlowPktInfo::update_pkt_info2(char *p,
                                            int update_len ,
                                            CGenNode * node
                                            ){
-    IPHeader       * ipv4= 
+    IPHeader       * ipv4=
         (IPHeader       *)(p + m_pkt_indication.getFastIpOffsetFast());
 
-    EthernetHeader * et  = 
+    EthernetHeader * et  =
         (EthernetHeader * )(p + m_pkt_indication.getFastEtherOffset());
 
     (void)et;
@@ -2820,7 +2823,7 @@ inline void CFlowPktInfo::update_pkt_info2(char *p,
                 m_tcp->setSourcePort(flow_info->server_port);
             }
         }
-        
+
     }else {
         if ( m_pkt_indication.m_desc.IsUdp() ){
             UDPHeader * m_udp =(UDPHeader *)(p +m_pkt_indication.getFastTcpOffset() );
@@ -2848,10 +2851,10 @@ inline void CFlowPktInfo::update_pkt_info2(char *p,
 inline void CFlowPktInfo::update_pkt_info(char *p,
                                    CGenNode * node){
 
-    IPHeader       * ipv4= 
+    IPHeader       * ipv4=
         (IPHeader       *)(p + m_pkt_indication.getFastIpOffsetFast());
 
-    EthernetHeader * et  = 
+    EthernetHeader * et  =
         (EthernetHeader * )(p + m_pkt_indication.getFastEtherOffset());
 
     (void)et;
@@ -2863,7 +2866,7 @@ inline void CFlowPktInfo::update_pkt_info(char *p,
 
 
     if ( unlikely (m_pkt_indication.is_ipv6())) {
-    
+
         // Update the IPv6 address
         IPv6Header *ipv6= (IPv6Header *)ipv4;
 
@@ -2886,22 +2889,22 @@ inline void CFlowPktInfo::update_pkt_info(char *p,
                 ipv4->setTimeToLive(TTL_RESERVE_DUPLICATE);
 
                 /* first ipv4 option add the info in case of learn packet, usualy only the first packet */
-		if (CGlobalInfo::is_learn_mode(CParserOption::LEARN_MODE_IP_OPTION)) {
-		    CNatOption *lpNat =(CNatOption *)ipv4->getOption();
-		    lpNat->set_fid(node->get_short_fid());
-		    lpNat->set_thread_id(node->get_thread_id()); 
-		} else {
-		    // This method only work on first TCP SYN
-		    if (ipv4->getProtocol() == IPPROTO_TCP) {
-			TCPHeader *tcp = (TCPHeader *)(((uint8_t *)ipv4) + ipv4->getHeaderLength());
-			if (tcp->getSynFlag()) {
-			    tcp->setAckNumber(CNatRxManager::calc_tcp_ack_val(node->get_short_fid(), node->get_thread_id()));
-			}
+        if (CGlobalInfo::is_learn_mode(CParserOption::LEARN_MODE_IP_OPTION)) {
+            CNatOption *lpNat =(CNatOption *)ipv4->getOption();
+            lpNat->set_fid(node->get_short_fid());
+            lpNat->set_thread_id(node->get_thread_id());
+        } else {
+            // This method only work on first TCP SYN
+            if (ipv4->getProtocol() == IPPROTO_TCP) {
+            TCPHeader *tcp = (TCPHeader *)(((uint8_t *)ipv4) + ipv4->getHeaderLength());
+            if (tcp->getSynFlag()) {
+                tcp->setAckNumber(CNatRxManager::calc_tcp_ack_val(node->get_short_fid(), node->get_thread_id()));
+            }
 #ifdef NAT_TRACE_
-			printf(" %.3f : flow_id: %x thread_id %x TCP ack %x\n",now_sec(), node->get_short_fid(), node->get_thread_id(), tcp->getAckNumber());
+            printf(" %.3f : flow_id: %x thread_id %x TCP ack %x\n",now_sec(), node->get_short_fid(), node->get_thread_id(), tcp->getAckNumber());
 #endif
-		    }
-		}
+            }
+        }
             }
             /* in all cases update the ip using the outside ip */
 
@@ -3005,7 +3008,7 @@ inline rte_mbuf_t * CFlowPktInfo::do_generate_new_mbuf_ex(CGenNode * node,
 
     BP_ASSERT ( (((uintptr_t)m_packet->raw) & 0x7f )== 0) ;
 
-    memcpy(p,m_packet->raw,len); 
+    memcpy(p,m_packet->raw,len);
 
     update_pkt_info2(p,flow_info,0,node);
 
@@ -3014,7 +3017,7 @@ inline rte_mbuf_t * CFlowPktInfo::do_generate_new_mbuf_ex(CGenNode * node,
     return(m);
 }
 
-    
+
 inline rte_mbuf_t * CFlowPktInfo::do_generate_new_mbuf_ex_big(CGenNode * node,
                                                           CFlowInfo * flow_info){
     rte_mbuf_t        * m;
@@ -3029,7 +3032,7 @@ inline rte_mbuf_t * CFlowPktInfo::do_generate_new_mbuf_ex_big(CGenNode * node,
 
     BP_ASSERT ( (((uintptr_t)m_packet->raw) & 0x7f )== 0) ;
 
-    memcpy(p,m_packet->raw,len); 
+    memcpy(p,m_packet->raw,len);
 
     update_pkt_info2(p,flow_info,0,node);
 
@@ -3055,7 +3058,7 @@ inline rte_mbuf_t * CFlowPktInfo::do_generate_new_mbuf_ex_vm(CGenNode * node,
 
     /* alloc big buffer to update it*/
     m = CGlobalInfo::pktmbuf_alloc(node->get_socket_id(), len);
-	assert(m);
+    assert(m);
 
     /* append the additional bytes requested and update later */
     char *p=rte_pktmbuf_append(m, len);
@@ -3063,7 +3066,7 @@ inline rte_mbuf_t * CFlowPktInfo::do_generate_new_mbuf_ex_vm(CGenNode * node,
     BP_ASSERT ( (((uintptr_t)m_packet->raw) & 0x7f )== 0) ;
 
     /* copy the headers until the payload  */
-    memcpy(p, m_packet->raw, m_pkt_indication.getPayloadOffset() ); 
+    memcpy(p, m_packet->raw, m_pkt_indication.getPayloadOffset() );
     CMiniVM vm;
     vm.m_pkt_info = this;
     vm.m_pyload_mbuf_ptr = p+m_pkt_indication.getPayloadOffset();
@@ -3077,7 +3080,7 @@ inline rte_mbuf_t * CFlowPktInfo::do_generate_new_mbuf_ex_vm(CGenNode * node,
     /* update IP length , and TCP checksum , we can accelerate this using hardware ! */
     uint16_t pkt_adjust = vm.m_new_pkt_size - m_packet->pkt_len;
     update_pkt_info2(p,flow_info,pkt_adjust,node);
-    
+
     /* return change in packet size due to packet tranforms */
     *s_size = vm.m_new_pkt_size - m_packet->pkt_len;
 
@@ -3110,7 +3113,7 @@ inline rte_mbuf_t * CFlowPktInfo::do_generate_new_mbuf(CGenNode * node){
 
     BP_ASSERT ( (((uintptr_t)m_packet->raw) & 0x7f )== 0) ;
 
-    memcpy(p,m_packet->raw,len); 
+    memcpy(p,m_packet->raw,len);
 
     update_pkt_info(p,node);
 
@@ -3119,7 +3122,7 @@ inline rte_mbuf_t * CFlowPktInfo::do_generate_new_mbuf(CGenNode * node){
     return m;
 }
 
-    
+
 inline rte_mbuf_t * CFlowPktInfo::do_generate_new_mbuf_big(CGenNode * node){
     rte_mbuf_t        * m;
     uint16_t len =  m_packet->pkt_len;
@@ -3133,7 +3136,7 @@ inline rte_mbuf_t * CFlowPktInfo::do_generate_new_mbuf_big(CGenNode * node){
 
     BP_ASSERT ( (((uintptr_t)m_packet->raw) & 0x7f )== 0) ;
 
-    memcpy(p,m_packet->raw,len); 
+    memcpy(p,m_packet->raw,len);
 
     update_pkt_info(p,node);
 
@@ -3206,15 +3209,15 @@ public:
 class CCapFileFlowInfo {
 public:
     enum load_cap_file_err {
-	kOK = 0,
-	kFileNotExist,
-	kNegTimestamp,
-	kNoSyn,
-	kTCPOffsetTooBig,
-	kNoTCPFromServer,
-	kPktNotSupp,
-	kPktProcessFail,
-	kCapFileErr
+    kOK = 0,
+    kFileNotExist,
+    kNegTimestamp,
+    kNoSyn,
+    kTCPOffsetTooBig,
+    kNoTCPFromServer,
+    kPktNotSupp,
+    kPktProcessFail,
+    kCapFileErr
     };
 
     bool Create();
@@ -3253,7 +3256,7 @@ public:
         return (m_total_errors);
     }
 
-    // return the cap file length in sec 
+    // return the cap file length in sec
     double get_cap_file_length_sec();
 
     void get_total_memory(CCCapFileMemoryUsage & memory);
@@ -3287,8 +3290,8 @@ public:
 // IPv4 addressing
 
 // IPv6 addressing
-    std::vector     <uint16_t> m_src_ipv6;   
-    std::vector     <uint16_t> m_dst_ipv6;   
+    std::vector     <uint16_t> m_src_ipv6;
+    std::vector     <uint16_t> m_dst_ipv6;
     bool             m_ipv6_set;
 
 // new section
@@ -3342,7 +3345,7 @@ public:
     double duration_sec;
     double m_cps;
     double m_mb_sec;
-    double m_mB_sec; 
+    double m_mB_sec;
     double m_c_flows;
     double m_pps ;
     double m_total_Mbytes ;
@@ -3367,7 +3370,7 @@ public:
 class CFlowGeneratorRecPerThread {
 
 public:
-    bool Create(CTupleGeneratorSmart  * global_gen, 
+    bool Create(CTupleGeneratorSmart  * global_gen,
                 CFlowYamlInfo *         info,
                 CFlowsYamlInfo *        yaml_flow_info,
                 CCapFileFlowInfo *      flow_info,
@@ -3388,11 +3391,11 @@ public:
     CCapFileFlowInfo *      m_flow_info;
     CFlowYamlInfo *         m_info;
     CFlowsYamlInfo *        m_flows_info;
-    CPolicer                m_policer;   
+    CPolicer                m_policer;
     uint16_t                m_id ;
     uint32_t                m_thread_id;
     bool                    m_tuple_gen_was_set;
-} __rte_cache_aligned; 
+} __rte_cache_aligned;
 
 
 
@@ -3405,16 +3408,16 @@ public:
                 uint16_t _id);
     void Delete();
 public:
-    
+
     void Dump(FILE *fd);
     void getFlowStats(CFlowStats * stats);
 public:
     CCapFileFlowInfo m_flow_info;
     CFlowYamlInfo *  m_info;
     CFlowsYamlInfo * m_flows_info;
-    CPolicer         m_policer;   
+    CPolicer         m_policer;
     uint16_t         m_id;
-private: 
+private:
     void fixup_ipg_if_needed();
 };
 
@@ -3423,7 +3426,7 @@ public:
     CPPSMeasure(){
         reset();
     }
-    //reset 
+    //reset
     void reset(void){
         m_start=false;
         m_last_time_msec=0;
@@ -3453,7 +3456,7 @@ public:
 class CBwMeasure {
 public:
     CBwMeasure();
-    //reset 
+    //reset
     void reset(void);
     //add packet size
     double add(uint64_t size);
@@ -3498,7 +3501,7 @@ public:
     friend class CNodeGenerator;
     friend class CPluginCallbackSimple;
     friend class CCapFileFlowInfo;
-    
+
     typedef  CGenericMap<flow_id_t,CGenNode> flow_id_node_t;
 
     bool Create(uint32_t           thread_id,
@@ -3518,23 +3521,23 @@ public:
         m_node_gen.set_vif(v_if);
     }
 
-    /* return the dual port ID this thread is attached to in 4 ports configuration 
-       there are 2 dual-ports 
+    /* return the dual port ID this thread is attached to in 4 ports configuration
+       there are 2 dual-ports
 
       thread 0 - dual 0
       thread 1 - dual 1
 
       thread 2 - dual 0
       thread 3 - dual 1
-      
-     */         
+
+     */
     uint32_t getDualPortId();
 public :
     double get_total_kcps();
     double get_total_kcps(uint8_t pool_idx, bool is_client);
     double get_delta_flow_is_sec();
-    double get_longest_flow(); 
-    double get_longest_flow(uint8_t pool_idx, bool is_client); 
+    double get_longest_flow();
+    double get_longest_flow(uint8_t pool_idx, bool is_client);
     void inc_current_template(void);
     int generate_flows_roundrobin(bool *done);
     int reschedule_flow(CGenNode *node);
@@ -3627,9 +3630,9 @@ public:
     CFlowGenList                *    m_flow_list;
     rte_mempool_t *                  m_node_pool;
 
-    std::vector<CFlowGeneratorRecPerThread *> m_cap_gen;   
+    std::vector<CFlowGeneratorRecPerThread *> m_cap_gen;
 
-    CFlowsYamlInfo                   m_yaml_info; 
+    CFlowsYamlInfo                   m_yaml_info;
 
     CTupleGeneratorSmart             m_smart_gen;
 
@@ -3644,7 +3647,7 @@ public:
     double                           m_stop_time_sec;
 
     CPreviewMode                     m_preview_mode;
-public:    
+public:
     CFlowGenStats                    m_stats;
     CBwMeasure                       m_mb_sec;
     CCpuUtlDp                        m_cpu_dp_u;
@@ -3663,7 +3666,7 @@ private:
     bool                             m_terminated_by_master;
 
 private:
-    uint8_t                 m_cacheline_pad[RTE_CACHE_LINE_SIZE][19]; // improve prefech 
+    uint8_t                 m_cacheline_pad[RTE_CACHE_LINE_SIZE][19]; // improve prefech
 } __rte_cache_aligned ;
 
 inline CGenNode * CFlowGenListPerThread::create_node(void){
@@ -3726,7 +3729,7 @@ public:
 public:
     std::vector<CFlowGeneratorRec *> m_cap_gen;   /* global info */
     CFlowsYamlInfo                   m_yaml_info; /* global yaml*/
-    std::vector<CFlowGenListPerThread   *> m_threads_info;  
+    std::vector<CFlowGenListPerThread   *> m_threads_info;
     CFlowGenListMac                  m_mac_info;
 };
 
@@ -3761,19 +3764,19 @@ inline void CCapFileFlowInfo::generate_flow(CTupleTemplateGeneratorSmart   * tup
     node->m_flow_info = this;
     node->m_flags=0;
     node->m_template_info =template_info;
-    node->m_tuple_gen = tuple_gen->get_gen(); 
+    node->m_tuple_gen = tuple_gen->get_gen();
     node->m_src_ip= tuple.getClient();
     node->m_dest_ip = tuple.getServer();
     node->m_src_idx = tuple.getClientId();
     node->m_dest_idx = tuple.getServerId();
     node->m_src_port = tuple.getClientPort();
-    memcpy(&node->m_src_mac, 
-           tuple.getClientMac(), 
+    memcpy(&node->m_src_mac,
+           tuple.getClientMac(),
            sizeof(mac_addr_align_t));
     node->m_plugin_info =(void *)0;
 
     if ( unlikely( CGlobalInfo::is_learn_mode()  ) ){
-        // check if flow is two direction 
+        // check if flow is two direction
         if ( lp->m_pkt_indication.m_desc.IsBiDirectionalFlow() ) {
             /* we are in learn mode */
             CFlowGenListPerThread  * lpThread=gen->Parent();
@@ -3822,7 +3825,7 @@ inline void CFlowGeneratorRecPerThread::generate_flow(CNodeGenerator * gen,
                                                       uint64_t flow_id,
                                                       CGenNode * node){
 
-    m_flow_info->generate_flow(&tuple_gen, 
+    m_flow_info->generate_flow(&tuple_gen,
                                gen,
                                time,
                                flow_id,
@@ -3869,7 +3872,7 @@ inline void CGenNode::update_next_pkt_in_flow(void){
         m_pkt_info = m_flow_info->GetPacket((pkt_index-1));
 }
 
-inline void CGenNode::reset_pkt_in_flow(void){       
+inline void CGenNode::reset_pkt_in_flow(void){
         m_pkt_info = m_flow_info->GetPacket(0);
 }
 
@@ -3898,7 +3901,7 @@ public:
 class CPluginCallbackSimple : public CPluginCallback {
 public:
     virtual void on_node_first(uint8_t plugin_id,CGenNode *     node,
-                               CFlowYamlInfo *  template_info, 
+                               CFlowYamlInfo *  template_info,
                                CTupleTemplateGeneratorSmart * tuple_gen,
                                CFlowGenListPerThread  * flow_gen);
     virtual void on_node_last(uint8_t plugin_id,CGenNode *     node);
@@ -3950,7 +3953,7 @@ inline  pkt_dir_t CGenNode::cur_interface_dir(){
         return (is_eligible_from_server_side()?SERVER_SIDE:CLIENT_SIDE);
     }else{
         return ( is_init ?CLIENT_SIDE:SERVER_SIDE);
-    } 
+    }
 }
 
 
diff --git a/src/flow_stat.cpp b/src/flow_stat.cpp
index 20cab376..13f8eb16 100644
--- a/src/flow_stat.cpp
+++ b/src/flow_stat.cpp
@@ -27,11 +27,11 @@ CFlowStatHwIdMap - Mapping between hardware id and packet group id
 CFlowStatRuleMgr - API to users of the file
 
 General idea of operation:
-For each stream needing flow statistics, the user provides packet group id (pg_id). Few streams can have the same pg_id. 
+For each stream needing flow statistics, the user provides packet group id (pg_id). Few streams can have the same pg_id.
 We maintain reference count.
 When doing start_stream, for the first stream in pg_id, hw_id is associated with the pg_id, and relevant hardware rules are
 inserted (on supported hardware). When stopping all streams with the pg_id, the hw_id <--> pg_id mapping is removed, hw_id is
-returned to the free hw_id pool, and hardware rules are removed. Counters for the pg_id are kept. 
+returned to the free hw_id pool, and hardware rules are removed. Counters for the pg_id are kept.
 If starting streams again, new hw_id will be assigned, and counters will continue from where they stopped. Only When deleting
 all streams using certain pg_id, infromation about this pg_id will be freed.
 
@@ -52,6 +52,7 @@ stream_del: HW_ID_INIT
 #include "internal_api/trex_platform_api.h"
 #include "trex_stateless.h"
 #include "trex_stateless_messaging.h"
+#include "trex_stateless_rx_core.h"
 #include "trex_stream.h"
 #include "flow_stat_parser.h"
 #include "flow_stat.h"
@@ -421,6 +422,7 @@ CFlowStatRuleMgr::CFlowStatRuleMgr() {
     m_ring_to_rx = NULL;
     m_capabilities = 0;
     m_parser = NULL;
+    m_rx_core = NULL;
 }
 
 CFlowStatRuleMgr::~CFlowStatRuleMgr() {
@@ -442,6 +444,7 @@ void CFlowStatRuleMgr::create() {
     }
     m_ring_to_rx = CMsgIns::Ins()->getCpRx()->getRingCpToDp(0);
     assert(m_ring_to_rx);
+    m_rx_core = get_rx_sl_core_obj();
     m_parser = m_api->get_flow_stat_parser();
     assert(m_parser);
     m_capabilities = capabilities;
@@ -663,6 +666,22 @@ int CFlowStatRuleMgr::start_stream(TrexStream * stream) {
 
     if (m_num_started_streams == 0) {
         send_start_stop_msg_to_rx(true); // First transmitting stream. Rx core should start reading packets;
+
+        // wait to make sure that message is acknowledged. RX core might be in deep sleep mode, and we want to
+        // start transmitting packets only after it is working, otherwise, packets will get lost.
+        if (m_rx_core) { // in simulation, m_rx_core will be NULL
+            int count = 0;
+            while (!m_rx_core->is_working()) {
+                delay(1);
+                count++;
+                if (count == 100) {
+                    throw TrexException("Critical error!! - RX core failed to start");
+                }
+            }
+        }
+    } else {
+        // make sure rx core is working. If not, we got really confused somehow.
+        assert(m_rx_core->is_working());
     }
     m_num_started_streams++;
     return 0;
diff --git a/src/flow_stat.h b/src/flow_stat.h
index b20a8b31..06b54d70 100644
--- a/src/flow_stat.h
+++ b/src/flow_stat.h
@@ -37,6 +37,8 @@
 typedef std::map<uint32_t, uint16_t> flow_stat_map_t;
 typedef std::map<uint32_t, uint16_t>::iterator flow_stat_map_it_t;
 
+class CRxCoreStateless;
+
 class tx_per_flow_t_ {
  public:
     tx_per_flow_t_() {
@@ -220,6 +222,7 @@ class CFlowStatRuleMgr {
     CFlowStatUserIdMap m_user_id_map; // map user ids to hw ids
     uint8_t m_num_ports; // How many ports are being used
     const TrexPlatformApi *m_api;
+    const CRxCoreStateless *m_rx_core;
     int m_max_hw_id; // max hw id we ever used
     uint32_t m_num_started_streams; // How many started (transmitting) streams we have
     CNodeRing *m_ring_to_rx; // handle for sending messages to Rx core
diff --git a/src/latency.h b/src/latency.h
index 3dd1cc36..f5f90cf9 100644
--- a/src/latency.h
+++ b/src/latency.h
@@ -22,6 +22,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 #include <bp_sim.h>
+#include <flow_stat.h>
 
 #define L_PKT_SUBMODE_NO_REPLY 1
 #define L_PKT_SUBMODE_REPLY 2
diff --git a/src/main.cpp b/src/main.cpp
index 6a6b5721..3c68990c 100755
--- a/src/main.cpp
+++ b/src/main.cpp
@@ -253,6 +253,10 @@ TrexStateless * get_stateless_obj() {
     return m_sim_statelss_obj;
 }
 
+CRxCoreStateless * get_rx_sl_core_obj() {
+    return NULL;
+}
+
 void set_stateless_obj(TrexStateless *obj) {
     m_sim_statelss_obj = obj;
 }
diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp
index 363189d4..1f415958 100644
--- a/src/main_dpdk.cpp
+++ b/src/main_dpdk.cpp
@@ -191,8 +191,7 @@ public:
     virtual void clear_extended_stats(CPhyEthIF * _if);
     virtual int dump_fdir_global_stats(CPhyEthIF * _if, FILE *fd) {return 0;}
     virtual int get_stat_counters_num() {return MAX_FLOW_STATS;}
-    virtual int get_rx_stat_capabilities() {return 0;}
-    //virtual int get_rx_stat_capabilities() {return TrexPlatformApi::IF_STAT_IPV4_ID;}
+    virtual int get_rx_stat_capabilities() {return TrexPlatformApi::IF_STAT_IPV4_ID;}
     virtual int wait_for_stable_link();
     virtual void wait_after_link_up();
 };
@@ -245,8 +244,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 0;}
-    //    virtual int get_rx_stat_capabilities() {return TrexPlatformApi::IF_STAT_IPV4_ID;}
+    virtual int get_rx_stat_capabilities() {return TrexPlatformApi::IF_STAT_IPV4_ID;}
 };
 
 
@@ -283,8 +281,7 @@ public:
     virtual void clear_extended_stats(CPhyEthIF * _if);
     virtual int wait_for_stable_link();
     virtual int get_stat_counters_num() {return MAX_FLOW_STATS;}
-    virtual int get_rx_stat_capabilities() {return 0;}
-    //    virtual int get_rx_stat_capabilities() {return TrexPlatformApi::IF_STAT_IPV4_ID;}
+    virtual int get_rx_stat_capabilities() {return TrexPlatformApi::IF_STAT_IPV4_ID;}
     virtual CFlowStatParser *get_flow_stat_parser();
 };
 
@@ -1239,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){
 
@@ -1743,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);
 
@@ -1812,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.
@@ -4126,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;
diff --git a/src/main_dpdk.h b/src/main_dpdk.h
index ff1ea784..a9bfed39 100644
--- a/src/main_dpdk.h
+++ b/src/main_dpdk.h
@@ -122,6 +122,7 @@ class CPhyEthIF  {
     CPhyEthIFStats     & get_stats(){
         return ( m_stats );
     }
+    void flush_dp_rx_queue(void);
     void flush_rx_queue(void);
     int add_rx_flow_stat_rule(uint8_t type, uint16_t proto, uint16_t id);
     int del_rx_flow_stat_rule(uint8_t type, uint16_t proto, uint16_t id);
diff --git a/src/stateless/cp/trex_stateless.h b/src/stateless/cp/trex_stateless.h
index 6e5e0c44..7db86174 100644
--- a/src/stateless/cp/trex_stateless.h
+++ b/src/stateless/cp/trex_stateless.h
@@ -197,6 +197,7 @@ protected:
  * @return TrexStateless& 
  */
 TrexStateless * get_stateless_obj();
+CRxCoreStateless * get_rx_sl_core_obj();
 
 #endif /* __TREX_STATELESS_H__ */
 
diff --git a/src/stateless/dp/trex_stateless_dp_core.cpp b/src/stateless/dp/trex_stateless_dp_core.cpp
index ba25f61d..f125a46a 100644
--- a/src/stateless/dp/trex_stateless_dp_core.cpp
+++ b/src/stateless/dp/trex_stateless_dp_core.cpp
@@ -399,6 +399,7 @@ TrexStatelessDpCore::idle_state_loop() {
     int counter = 0;
 
     while (m_state == STATE_IDLE) {
+        m_core->m_node_gen.m_v_if->flush_dp_rx_queue();
         bool had_msg = periodic_check_for_cp_messages();
         if (had_msg) {
             counter = 0;
diff --git a/src/stateless/rx/trex_stateless_rx_core.cpp b/src/stateless/rx/trex_stateless_rx_core.cpp
index 42889f0a..26f537f8 100644
--- a/src/stateless/rx/trex_stateless_rx_core.cpp
+++ b/src/stateless/rx/trex_stateless_rx_core.cpp
@@ -2,6 +2,7 @@
 #include "bp_sim.h"
 #include "flow_stat_parser.h"
 #include "latency.h"
+#include "pal/linux/sanb_atomic.h"
 #include "trex_stateless_messaging.h"
 #include "trex_stateless_rx_core.h"
 
@@ -55,11 +56,12 @@ void CRxCoreStateless::idle_state_loop() {
     int counter = 0;
 
     while (m_state == STATE_IDLE) {
-        flush_rx();
         bool had_msg = periodic_check_for_cp_messages();
         if (had_msg) {
             counter = 0;
             continue;
+        } else {
+            flush_rx();
         }
 
         /* enter deep sleep only if enough time had passed */
@@ -73,8 +75,8 @@ void CRxCoreStateless::idle_state_loop() {
 }
 
 void CRxCoreStateless::start() {
-    static int count = 0;
-    static int i = 0;
+    int count = 0;
+    int i = 0;
     bool do_try_rx_queue =CGlobalInfo::m_options.preview.get_vm_one_queue_enable() ? true : false;
 
     while (true) {
@@ -92,7 +94,11 @@ void CRxCoreStateless::start() {
         } else {
             if (m_state == STATE_QUIT)
                 break;
+            count = 0;
+            i = 0;
+            set_working_msg_ack(false);
             idle_state_loop();
+            set_working_msg_ack(true);
         }
         if (do_try_rx_queue) {
             try_rx_queues();
@@ -236,6 +242,12 @@ int CRxCoreStateless::get_rx_stats(uint8_t port_id, rx_per_flow_t *rx_stats, int
     return 0;
 }
 
+void CRxCoreStateless::set_working_msg_ack(bool val) {
+    sanb_smp_memory_barrier();
+    m_ack_start_work_msg = val;
+    sanb_smp_memory_barrier();
+}
+
 double CRxCoreStateless::get_cpu_util() {
     m_cpu_cp_u.Update();
     return m_cpu_cp_u.GetVal();
diff --git a/src/stateless/rx/trex_stateless_rx_core.h b/src/stateless/rx/trex_stateless_rx_core.h
index 81eca38a..b78256c2 100644
--- a/src/stateless/rx/trex_stateless_rx_core.h
+++ b/src/stateless/rx/trex_stateless_rx_core.h
@@ -54,6 +54,8 @@ class CRxCoreStateless {
     void work() {m_state = STATE_WORKING;}
     void idle() {m_state = STATE_IDLE;}
     void quit() {m_state = STATE_QUIT;}
+    bool is_working() const {return (m_ack_start_work_msg == true);}
+    void set_working_msg_ack(bool val);
     double get_cpu_util();
 
  private:
@@ -72,10 +74,13 @@ class CRxCoreStateless {
     uint32_t m_max_ports;
     bool m_has_streams;
     CLatencyManagerPerPort m_ports[TREX_MAX_PORTS];
-    state_e   m_state; /* state of all 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;
+
 };
 #endif