Merge Console merge

40 files changed, 1777 insertions, 1634 deletions

diff --git a/scripts/automation/trex_control_plane/stl/console/trex_console.py b/scripts/automation/trex_control_plane/stl/console/trex_console.py
index e037f517..4b5e5f54 100755
--- a/scripts/automation/trex_control_plane/stl/console/trex_console.py
+++ b/scripts/automation/trex_control_plane/stl/console/trex_console.py
@@ -349,17 +349,25 @@ class TRexConsole(TRexGeneralCmd):
         with self.stateless_client.logger.supress():
             table = stl_map_ports(self.stateless_client, ports = ports)
 
-        tmp = list(ports)
+        
         print format_text('\nAcquired ports topology:\n', 'bold', 'underline')
-        while tmp:
-            a = tmp.pop(0)
-            b = table[a]
-            tmp.remove(b)
 
-            print "port {0} <--> port {1}".format(a, b)
+        # bi-dir ports
+        print format_text('Bi-directional ports:\n','underline')
+        for port_a, port_b in table['bi']:
+            print "port {0} <--> port {1}".format(port_a, port_b)
+
+        print ""
 
+        # unknown ports
+        print format_text('Mapping unknown:\n','underline')
+        for port in table['unknown']:
+            print "port {0}".format(port)
         print ""
 
+       
+      
+
     def do_history (self, line):
         '''Manage the command history\n'''
 
@@ -555,11 +563,15 @@ class TRexConsole(TRexGeneralCmd):
             return
 
         if opts.xterm:
+            if not os.path.exists('/usr/bin/xterm'):
+                print format_text("XTERM does not exists on this machine", 'bold')
+                return
 
             info = self.stateless_client.get_connection_info()
 
             exe = './trex-console --top -t -q -s {0} -p {1} --async_port {2}'.format(info['server'], info['sync_port'], info['async_port'])
-            cmd = ['xterm', '-geometry', '111x47', '-sl', '0', '-title', 'trex_tui', '-e', exe]
+            cmd = ['/usr/bin/xterm', '-geometry', '111x47', '-sl', '0', '-title', 'trex_tui', '-e', exe]
+
             self.terminal = subprocess.Popen(cmd)
 
             return
diff --git a/scripts/automation/trex_control_plane/stl/examples/stl_imix.py b/scripts/automation/trex_control_plane/stl/examples/stl_imix.py
index e3c01ca9..cc7691a3 100644
--- a/scripts/automation/trex_control_plane/stl/examples/stl_imix.py
+++ b/scripts/automation/trex_control_plane/stl/examples/stl_imix.py
@@ -30,9 +30,10 @@ def imix_test ():
         # map ports - identify the routes
         table = stl_map_ports(c)
 
-        print "Mapped ports to sides {0} <--> {1}".format(table['dir'][0], table['dir'][1])
-        dir_0 = table['dir'][0]
-        dir_1 = table['dir'][1]
+        dir_0 = [x[0] for x in table['bi']]
+        dir_1 = [x[1] for x in table['bi']]
+
+        print "Mapped ports to sides {0} <--> {1}".format(dir_0, dir_1)
 
         # load IMIX profile
         profile = STLProfile.load_py('../../../../stl/imix.py')
diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py
index 50566853..04dd77ec 100644
--- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py
+++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py
@@ -227,7 +227,7 @@ class AsyncEventHandler(object):
             ev = "Port {0} job done".format(port_id)
 
             # call the handler
-            self.__async_event_port_stopped(port_id)
+            self.__async_event_port_job_done(port_id)
             show_event = True
 
         # port was stolen...
@@ -264,6 +264,9 @@ class AsyncEventHandler(object):
 
     # private functions
 
+    def __async_event_port_job_done (self, port_id):
+        self.client.ports[port_id].async_event_port_job_done()
+
     def __async_event_port_stopped (self, port_id):
         self.client.ports[port_id].async_event_port_stopped()
 
@@ -1332,7 +1335,6 @@ class STLClient(object):
         :parameters:
             ports : list
                 ports to execute the command
-                
 
         :raises:
             + :exc:`STLError`
@@ -1354,7 +1356,6 @@ class STLClient(object):
         if not rc:
             raise STLError(rc)
 
-
         
     """
         update traffic on port(s)
@@ -1433,8 +1434,6 @@ class STLClient(object):
         if not rc:
             raise STLError(rc)
 
-              
-    
     """
         resume traffic on port(s)
 
diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_port.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_port.py
index 6aa18847..4529efa9 100644
--- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_port.py
+++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_port.py
@@ -334,7 +334,7 @@ class Port(object):
                   "mul":      mul,
                   "duration": duration,
                   "force":    force}
-
+        
         rc = self.transmit("start_traffic", params)
         if rc.bad():
             return self.err(rc.err())
@@ -363,7 +363,6 @@ class Port(object):
         if rc.bad():
             return self.err(rc.err())
 
-        # only valid state after stop
         self.state = self.STATE_STREAMS
 
         return self.ok()
@@ -383,7 +382,6 @@ class Port(object):
         if rc.bad():
             return self.err(rc.err())
 
-        # only valid state after stop
         self.state = self.STATE_PAUSE
 
         return self.ok()
@@ -400,11 +398,12 @@ class Port(object):
         params = {"handler": self.handler,
                   "port_id": self.port_id}
 
+        # only valid state after stop
+
         rc = self.transmit("resume_traffic", params)
         if rc.bad():
             return self.err(rc.err())
 
-        # only valid state after stop
         self.state = self.STATE_TX
 
         return self.ok()
@@ -591,21 +590,26 @@ class Port(object):
     
 
 
-    ################# events handler ######################
-    def async_event_port_stopped (self):
+  ################# events handler ######################
+    def async_event_port_job_done (self):
         self.state = self.STATE_STREAMS
 
-
-    def async_event_port_started (self):
-        self.state = self.STATE_TX
-
+    # rest of the events are used for TUI / read only sessions
+    def async_event_port_stopped (self):
+        if not self.is_acquired():
+            self.state = self.STATE_STREAMS
 
     def async_event_port_paused (self):
-        self.state = self.STATE_PAUSE
+        if not self.is_acquired():
+            self.state = self.STATE_PAUSE
 
+    def async_event_port_started (self):
+        if not self.is_acquired():
+            self.state = self.STATE_TX
 
     def async_event_port_resumed (self):
-        self.state = self.STATE_TX
+        if not self.is_acquired():
+            self.state = self.STATE_TX
 
     def async_event_forced_acquired (self):
         self.handler = None
diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_std.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_std.py
index 72a5ea52..e0b25b1d 100644
--- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_std.py
+++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_std.py
@@ -4,7 +4,6 @@ from trex_stl_packet_builder_scapy import *
 # map ports
 # will destroy all streams/data on the ports
 def stl_map_ports (client, ports = None):
-
     # by default use all ports
     if ports == None:
         ports = client.get_all_ports()
@@ -15,12 +14,15 @@ def stl_map_ports (client, ports = None):
     # generate streams
     base_pkt = CScapyTRexPktBuilder(pkt = Ether()/IP())
     
+    tx_pkts = {}
     pkts = 1
     for port in ports:
+        tx_pkts[pkts] = port
         stream = STLStream(packet = base_pkt,
                            mode = STLTXSingleBurst(pps = 100000, total_pkts = pkts))
 
         client.add_streams(stream, [port])
+        
         pkts = pkts * 2
 
     # inject
@@ -33,35 +35,33 @@ def stl_map_ports (client, ports = None):
     # cleanup
     client.reset(ports = ports)
 
-    table = {}
-    for port in ports:
-        table[port] = None
+    table = {'map': {}, 'bi' : [], 'unknown': []}
 
+    # actual mapping
     for port in ports:
+
         ipackets = stats[port]["ipackets"]
+        table['map'][port] = None
 
-        exp = 1
-        while ipackets >= exp:
-            if ((ipackets & exp) == (exp)):
-                source = int(math.log(exp, 2))
-                table[source] = port
+        for pkts in tx_pkts.keys():
+            if ( (pkts & ipackets) == pkts ):
+                tx_port = tx_pkts[pkts]
+                table['map'][port] = tx_port
 
-            exp *= 2
 
-    if not all(x != None for x in table.values()):
-        raise STLError('unable to map ports')
+    unmapped = list(ports)
+    while len(unmapped) > 0:
+        port_a = unmapped.pop(0)
+        port_b = table['map'][port_a]
 
-    dir_a = set()
-    dir_b = set()
-    for src, dst in table.iteritems():
-        # src is not in
-        if src not in (dir_a, dir_b):
-            if dst in dir_a:
-                dir_b.add(src)
-            else:
-                dir_a.add(src)
+        # if unknown - add to the unknown list
+        if port_b == None:
+            table['unknown'].append(port_a)
 
-    table['dir'] = [list(dir_a), list(dir_b)]
+        # bi-directional ports
+        elif (table['map'][port_b] == port_a):
+            unmapped.remove(port_b)
+            table['bi'].append( (port_a, port_b) )
 
     return table
 
diff --git a/src/bp_gtest.cpp b/src/bp_gtest.cpp
index e312abc7..3189e886 100755
--- a/src/bp_gtest.cpp
+++ b/src/bp_gtest.cpp
@@ -877,11 +877,11 @@ public:
 TEST_F(basic, latency3) {
     CLatencyManager mg;
     CLatencyManagerCfg  cfg;
-    CDummyLatencyHWBase dports[MAX_LATENCY_PORTS];
+    CDummyLatencyHWBase dports[TREX_MAX_PORTS];
     cfg.m_cps =10;
     cfg.m_max_ports=4;
     int i;
-    for (i=0; i<MAX_LATENCY_PORTS; i++) {
+    for (i = 0; i < TREX_MAX_PORTS; i++) {
         dports[i].m_port_id=i;
         cfg.m_ports[i] = &dports[i];
     }
@@ -2179,7 +2179,7 @@ TEST_F(rx_check_system, rx_system1) {
     m_rxcs.lpVf=&m_rx_check;
     CParserOption * po =&CGlobalInfo::m_options;
 
-    po->m_rx_check_sampe=2; /* sample rate */
+    po->m_rx_check_sample=2; /* sample rate */
     po->m_duration=100;
     po->cfg_file ="cap2/dns.yaml";
 
@@ -2197,7 +2197,7 @@ TEST_F(rx_check_system, rx_system1_dns) {
     m_rxcs.lpVf=&m_rx_check;
     CParserOption * po =&CGlobalInfo::m_options;
 
-    po->m_rx_check_sampe=1; /* sample rate */
+    po->m_rx_check_sample=1; /* sample rate */
     po->m_duration=1;
     po->cfg_file ="cap2/dns.yaml";
     m_rx_check.m_store_pcfg=true;
@@ -2221,7 +2221,7 @@ TEST_F(rx_check_system, rx_system1_ipv6) {
     CParserOption * po =&CGlobalInfo::m_options;
     po->preview.set_ipv6_mode_enable(true);
 
-    po->m_rx_check_sampe=4; /* sample rate */
+    po->m_rx_check_sample=4; /* sample rate */
     po->m_duration=100;
     po->cfg_file ="cap2/dns.yaml";
 
@@ -2242,7 +2242,7 @@ TEST_F(rx_check_system, rx_system1_dns_ipv6) {
     CParserOption * po =&CGlobalInfo::m_options;
 
     po->preview.set_ipv6_mode_enable(true);
-    po->m_rx_check_sampe=1; /* sample rate */
+    po->m_rx_check_sample=1; /* sample rate */
     po->m_duration=1;
     po->cfg_file ="cap2/dns.yaml";
     m_rx_check.m_store_pcfg=true;
@@ -2265,7 +2265,7 @@ TEST_F(rx_check_system, rx_system2_plugin_one_dir) {
     m_rxcs.lpVf=&m_rx_check;
     CParserOption * po =&CGlobalInfo::m_options;
 
-    po->m_rx_check_sampe=2; /* sample rate */
+    po->m_rx_check_sample=2; /* sample rate */
     po->m_duration=100;
     po->cfg_file ="cap2/rtsp_short1.yaml";
 
@@ -2283,7 +2283,7 @@ TEST_F(rx_check_system, rx_system2_plugin) {
     m_rxcs.lpVf=&m_rx_check;
     CParserOption * po =&CGlobalInfo::m_options;
 
-    po->m_rx_check_sampe=1; /* sample rate */
+    po->m_rx_check_sample=1; /* sample rate */
     po->m_duration=1;
     po->cfg_file ="cap2/rtsp_short1.yaml";
     m_rx_check.m_store_pcfg=true;
@@ -2308,7 +2308,7 @@ TEST_F(rx_check_system, rx_system2_plugin_ipv6) {
     CParserOption * po =&CGlobalInfo::m_options;
 
     po->preview.set_ipv6_mode_enable(true);
-    po->m_rx_check_sampe=1; /* sample rate */
+    po->m_rx_check_sample=1; /* sample rate */
     po->m_duration=1;
     po->cfg_file ="cap2/rtsp_short1.yaml";
     m_rx_check.m_store_pcfg=true;
@@ -2331,7 +2331,7 @@ TEST_F(rx_check_system, rx_system2_plugin_two_dir) {
     m_rxcs.lpVf=&m_rx_check;
     CParserOption * po =&CGlobalInfo::m_options;
 
-    po->m_rx_check_sampe=2; /* sample rate */
+    po->m_rx_check_sample=2; /* sample rate */
     po->m_duration=100;
     po->cfg_file ="cap2/rtsp_short1_slow.yaml";
     m_rx_check.m_one_dir=false;
@@ -2349,7 +2349,7 @@ TEST_F(rx_check_system, rx_system2_plugin_two_dir_2) {
     m_rxcs.lpVf=&m_rx_check;
     CParserOption * po =&CGlobalInfo::m_options;
 
-    po->m_rx_check_sampe=2; /* sample rate */
+    po->m_rx_check_sample=2; /* sample rate */
     po->m_duration=100;
     po->cfg_file ="cap2/rtsp_short1.yaml";
     m_rx_check.m_one_dir=false;
@@ -2367,7 +2367,7 @@ TEST_F(rx_check_system, rx_system_two_dir) {
     m_rxcs.lpVf=&m_rx_check;
     CParserOption * po =&CGlobalInfo::m_options;
 
-    po->m_rx_check_sampe=2; /* sample rate */
+    po->m_rx_check_sample=2; /* sample rate */
     po->m_duration=100;
     po->cfg_file ="cap2/dns.yaml";
     m_rx_check.m_one_dir=false;
@@ -2386,7 +2386,7 @@ TEST_F(rx_check_system, rx_json) {
     m_rxcs.lpVf=&m_rx_check;
     CParserOption * po =&CGlobalInfo::m_options;
 
-    po->m_rx_check_sampe=2; /* sample rate */
+    po->m_rx_check_sample=2; /* sample rate */
     po->m_duration=100;
     po->cfg_file ="cap2/dns.yaml";
 
diff --git a/src/bp_sim.cpp b/src/bp_sim.cpp
index 88e2c3ad..6ea40be2 100755
--- a/src/bp_sim.cpp
+++ b/src/bp_sim.cpp
@@ -120,8 +120,8 @@ socket_id_t CPlatformSocketInfoNoConfig::port_to_socket(port_id_t port){
 }
 
 
-void CPlatformSocketInfoNoConfig::set_latency_thread_is_enabled(bool enable){
-    m_latency_is_enabled = enable;
+void CPlatformSocketInfoNoConfig::set_rx_thread_is_enabled(bool enable) {
+    m_rx_is_enabled = enable;
 }
 
 void CPlatformSocketInfoNoConfig::set_number_of_dual_ports(uint8_t num_dual_ports){
@@ -141,7 +141,7 @@ bool CPlatformSocketInfoNoConfig::sanity_check(){
 uint64_t CPlatformSocketInfoNoConfig::get_cores_mask(){
 
     uint32_t cores_number = m_threads_per_dual_if*m_dual_if;
-    if ( m_latency_is_enabled ) {
+    if ( m_rx_is_enabled ) {
         cores_number +=   2;
     }else{
         cores_number += 1; /* only MASTER*/
@@ -170,7 +170,7 @@ bool CPlatformSocketInfoNoConfig::thread_phy_is_master(physical_thread_id_t  phy
     return (phy_id==0);
 }
 
-bool CPlatformSocketInfoNoConfig::thread_phy_is_latency(physical_thread_id_t  phy_id){
+bool CPlatformSocketInfoNoConfig::thread_phy_is_rx(physical_thread_id_t  phy_id){
     return (phy_id==(m_threads_per_dual_if*m_dual_if+1));
 }
 
@@ -257,8 +257,8 @@ bool CPlatformSocketInfoConfig::init(){
         exit(1);
     }
 
-    if ( m_thread_phy_to_virtual[m_platform->m_latency_thread] ){
-        printf("ERROR physical latency thread %d already defined \n",m_platform->m_latency_thread);
+    if ( m_thread_phy_to_virtual[m_platform->m_rx_thread] ){
+        printf("ERROR physical latency thread %d already defined \n",m_platform->m_rx_thread);
         exit(1);
     }
 
@@ -286,7 +286,7 @@ void CPlatformSocketInfoConfig::dump(FILE *fd){
 
     fprintf(fd," ports_sockets : %d \n",max_num_active_sockets());
 
-    for (i=0; i<(MAX_LATENCY_PORTS); i++) {
+    for (i = 0; i <  TREX_MAX_PORTS; i++) {
         fprintf(fd,"%d,",port_to_socket(i));
     }
     fprintf(fd,"\n");
@@ -314,14 +314,14 @@ void CPlatformSocketInfoConfig::reset(){
     for (i=0; i<MAX_THREADS_SUPPORTED; i++) {
         m_thread_phy_to_virtual[i]=0;
     }
-    for (i=0; i<(MAX_LATENCY_PORTS>>1); i++) {
+    for (i = 0; i < TREX_MAX_PORTS >> 1; i++) {
         m_socket_per_dual_if[i]=0;
     }
 
     m_num_dual_if=0;
 
     m_threads_per_dual_if=0;
-    m_latency_is_enabled=false;
+    m_rx_is_enabled=false;
     m_max_threads_per_dual_if=0;
 }
 
@@ -343,8 +343,8 @@ socket_id_t CPlatformSocketInfoConfig::port_to_socket(port_id_t port){
     return ( m_socket_per_dual_if[(port>>1)]);
 }
 
-void CPlatformSocketInfoConfig::set_latency_thread_is_enabled(bool enable){
-    m_latency_is_enabled =enable;
+void CPlatformSocketInfoConfig::set_rx_thread_is_enabled(bool enable){
+    m_rx_is_enabled =enable;
 }
 
 void CPlatformSocketInfoConfig::set_number_of_dual_ports(uint8_t num_dual_ports){
@@ -376,9 +376,9 @@ uint64_t CPlatformSocketInfoConfig::get_cores_mask(){
 
     mask |=(1<<m_platform->m_master_thread);
     assert(m_platform->m_master_thread<64);
-    if (m_latency_is_enabled) {
-        mask |=(1<<m_platform->m_latency_thread);
-        assert(m_platform->m_latency_thread<64);
+    if (m_rx_is_enabled) {
+        mask |=(1<<m_platform->m_rx_thread);
+        assert(m_platform->m_rx_thread<64);
     }
     return (mask);
 }
@@ -395,8 +395,8 @@ bool CPlatformSocketInfoConfig::thread_phy_is_master(physical_thread_id_t  phy_i
     return (m_platform->m_master_thread==phy_id?true:false);
 }
 
-bool CPlatformSocketInfoConfig::thread_phy_is_latency(physical_thread_id_t  phy_id){
-    return (m_platform->m_latency_thread == phy_id?true:false);
+bool CPlatformSocketInfoConfig::thread_phy_is_rx(physical_thread_id_t  phy_id){
+    return (m_platform->m_rx_thread == phy_id?true:false);
 }
 
 
@@ -437,8 +437,8 @@ socket_id_t CPlatformSocketInfo::port_to_socket(port_id_t port){
 }
 
 
-void CPlatformSocketInfo::set_latency_thread_is_enabled(bool enable){
-    m_obj->set_latency_thread_is_enabled(enable);
+void CPlatformSocketInfo::set_rx_thread_is_enabled(bool enable){
+    m_obj->set_rx_thread_is_enabled(enable);
 }
 
 void CPlatformSocketInfo::set_number_of_dual_ports(uint8_t num_dual_ports){
@@ -470,8 +470,8 @@ bool CPlatformSocketInfo::thread_phy_is_master(physical_thread_id_t  phy_id){
     return ( m_obj->thread_phy_is_master(phy_id));
 }
 
-bool CPlatformSocketInfo::thread_phy_is_latency(physical_thread_id_t  phy_id){
-    return ( m_obj->thread_phy_is_latency(phy_id));
+bool CPlatformSocketInfo::thread_phy_is_rx(physical_thread_id_t  phy_id) {
+    return ( m_obj->thread_phy_is_rx(phy_id));
 }
 
 void CPlatformSocketInfo::dump(FILE *fd){
@@ -3930,9 +3930,9 @@ void CFlowGenListPerThread::terminate_nat_flows(CGenNode *p){
 }
 
 
-void CFlowGenListPerThread::handel_latecy_pkt_msg(CGenNodeLatencyPktInfo * msg){
+void CFlowGenListPerThread::handle_latency_pkt_msg(CGenNodeLatencyPktInfo * msg){
     /* send the packet */
-    #ifdef LATENCY_QUEUE_TRACE_
+    #ifdef RX_QUEUE_TRACE_
     printf(" latency  msg dir %d\n",msg->m_dir);
     struct rte_mbuf * m;
     m=msg->m_pkt;
@@ -3950,7 +3950,7 @@ void CFlowGenListPerThread::handel_latecy_pkt_msg(CGenNodeLatencyPktInfo * msg){
 }
 
 
-void CFlowGenListPerThread::handel_nat_msg(CGenNodeNatInfo * msg){
+void CFlowGenListPerThread::handle_nat_msg(CGenNodeNatInfo * msg){
     int i;
     for (i=0; i<msg->m_cnt; i++) {
         CNatFlowInfo * nat_msg=&msg->m_data[i];
@@ -4010,11 +4010,11 @@ void CFlowGenListPerThread::check_msgs(void) {
         uint8_t   msg_type =  msg->m_msg_type;
         switch (msg_type ) {
         case CGenNodeMsgBase::NAT_FIRST:
-            handel_nat_msg((CGenNodeNatInfo * )msg);
+            handle_nat_msg((CGenNodeNatInfo * )msg);
             break;
 
         case CGenNodeMsgBase::LATENCY_PKT:
-            handel_latecy_pkt_msg((CGenNodeLatencyPktInfo *) msg);
+            handle_latency_pkt_msg((CGenNodeLatencyPktInfo *) msg);
             break;
 
         default:
@@ -4525,7 +4525,7 @@ void CParserOption::dump(FILE *fd){
 
 
     int i;
-    for (i=0; i<MAX_LATENCY_PORTS; i++) {
+    for (i = 0; i < TREX_MAX_PORTS; i++) {
         fprintf(fd," port : %d dst:",i);
         CMacAddrCfg * lp=&m_mac_addr[i];
         dump_mac_addr(fd,lp->u.m_mac.dest);
diff --git a/src/bp_sim.h b/src/bp_sim.h
index c9550dcf..37ed7854 100755
--- a/src/bp_sim.h
+++ b/src/bp_sim.h
@@ -57,6 +57,7 @@ limitations under the License.
 #include <common/cgen_map.h>
 #include <arpa/inet.h>
 #include "platform_cfg.h"
+#include "flow_stat.h"
 
 #include <trex_stateless_dp_core.h>
 
@@ -70,8 +71,6 @@ usec_to_sec(double usec) {
 
 #define FORCE_NO_INLINE __attribute__ ((noinline))
 
-#define MAX_LATENCY_PORTS TREX_MAX_PORTS
-
 /* IP address, last 32-bits of IPv6 remaps IPv4 */
 typedef struct {
     uint16_t v6[6];  /* First 96-bits of IPv6 */
@@ -258,9 +257,6 @@ void on_node_last(uint8_t plugin_id,CGenNode *     node);
 
 rte_mbuf_t * on_node_generate_mbuf(uint8_t plugin_id,CGenNode *     node,CFlowPktInfo * pkt_info);
 
-
-
-
 class CPreviewMode ;
 struct CGenNode;
 /* represent the virtual interface 
@@ -280,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];    
     CPerTxthreadTemplateInfo m_template;
 
 public:
@@ -740,7 +736,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_sampe=0;
+		m_rx_check_sample=0;
         m_rx_check_hops = 0;
         m_io_mode=1;
         m_run_flags=0;
@@ -748,6 +744,7 @@ public:
         m_mac_splitter=0;
         m_run_mode = RUN_MODE_INVALID;
         m_l_pkt_mode = 0;
+        m_rx_thread_enabled = false;
     }
 
 
@@ -762,7 +759,7 @@ 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_sampe; /* 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;
@@ -773,6 +770,7 @@ public:
     uint8_t         m_l_pkt_mode;
     uint8_t         m_learn_mode;
     uint16_t        m_debug_pkt_proto;
+    bool            m_rx_thread_enabled;
     trex_run_mode_e    m_run_mode;
 
 
@@ -785,7 +783,7 @@ public:
     std::string        prefix;
 
                                  
-    CMacAddrCfg     m_mac_addr[MAX_LATENCY_PORTS];
+    CMacAddrCfg     m_mac_addr[TREX_MAX_PORTS];
 
     uint8_t *       get_src_mac_addr(int if_index){
         return (m_mac_addr[if_index].u.m_mac.src);
@@ -807,17 +805,19 @@ public:
     uint32_t get_number_of_dp_cores_needed() {
         return ( (m_expected_portd>>1)   * preview.getCores());
     }
-    bool is_latency_disabled(){
-        return ( m_latency_rate == 0 ?true:false);
-    }
-
     bool is_stateless(){
         return (m_run_mode == RUN_MODE_INTERACTIVE ?true:false);
     }
-
-    bool is_latency_enabled(){
-        return ( !is_latency_disabled() );
+    bool is_latency_enabled() {
+        return ( (m_latency_rate == 0) ? false : true);
+    }
+    bool is_rx_enabled() {
+        return m_rx_thread_enabled;
     }
+    void set_rx_enabled() {
+        m_rx_thread_enabled = true;
+    }
+
     inline void set_rxcheck_const_ts(){
         m_run_flags |= RUN_FLAGS_RXCHECK_CONST_TS;
     }
@@ -888,7 +888,7 @@ DEFAULT:
     v5        v6
     v7        v8       
        
-    latency is v9          
+    rx is v9          
 
   */  
 
@@ -917,7 +917,7 @@ public:
 public:
     /* this is from CLI, number of thread per dual port */
     virtual void set_number_of_threads_per_ports(uint8_t num_threads)=0;
-    virtual void set_latency_thread_is_enabled(bool enable)=0;
+    virtual void set_rx_thread_is_enabled(bool enable)=0;
     virtual void set_number_of_dual_ports(uint8_t num_dual_ports)=0;
 
 
@@ -933,7 +933,7 @@ public:
     virtual physical_thread_id_t thread_virt_to_phy(virtual_thread_id_t virt_id)=0;
 
     virtual bool thread_phy_is_master(physical_thread_id_t  phy_id)=0;
-    virtual bool thread_phy_is_latency(physical_thread_id_t  phy_id)=0;
+    virtual bool thread_phy_is_rx(physical_thread_id_t  phy_id)=0;
 
     virtual void dump(FILE *fd)=0;
 };
@@ -944,7 +944,7 @@ public:
     CPlatformSocketInfoNoConfig(){
         m_dual_if=0;
         m_threads_per_dual_if=0;
-        m_latency_is_enabled=false;
+        m_rx_is_enabled=false;
     }
 
     /* is socket enabled */
@@ -960,7 +960,7 @@ public:
 public:
     /* this is from CLI, number of thread per dual port */
     void set_number_of_threads_per_ports(uint8_t num_threads);
-    void set_latency_thread_is_enabled(bool enable);
+    void set_rx_thread_is_enabled(bool enable);
     void set_number_of_dual_ports(uint8_t num_dual_ports);
 
     bool sanity_check();
@@ -975,14 +975,14 @@ public:
     physical_thread_id_t thread_virt_to_phy(virtual_thread_id_t virt_id);
 
     bool thread_phy_is_master(physical_thread_id_t  phy_id);
-    bool thread_phy_is_latency(physical_thread_id_t  phy_id);
+    bool thread_phy_is_rx(physical_thread_id_t  phy_id);
 
     virtual void dump(FILE *fd);
 
 private:
     uint32_t                 m_dual_if;
     uint32_t                 m_threads_per_dual_if;
-    bool                     m_latency_is_enabled;
+    bool                     m_rx_is_enabled;
 };
 
 
@@ -1006,7 +1006,7 @@ public:
 public:
     /* this is from CLI, number of thread per dual port */
     void set_number_of_threads_per_ports(uint8_t num_threads);
-    void set_latency_thread_is_enabled(bool enable);
+    void set_rx_thread_is_enabled(bool enable);
     void set_number_of_dual_ports(uint8_t num_dual_ports);
 
     bool sanity_check();
@@ -1021,7 +1021,7 @@ public:
     physical_thread_id_t thread_virt_to_phy(virtual_thread_id_t virt_id);
 
     bool thread_phy_is_master(physical_thread_id_t  phy_id);
-    bool thread_phy_is_latency(physical_thread_id_t  phy_id);
+    bool thread_phy_is_rx(physical_thread_id_t  phy_id);
 
 public:
     virtual void dump(FILE *fd);
@@ -1032,13 +1032,13 @@ private:
 private:
     bool                     m_sockets_enable[MAX_SOCKETS_SUPPORTED];
     uint32_t                 m_sockets_enabled;
-    socket_id_t              m_socket_per_dual_if[(MAX_LATENCY_PORTS>>1)];
+    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;
     uint32_t                 m_threads_per_dual_if;
-    bool                     m_latency_is_enabled;
+    bool                     m_rx_is_enabled;
     uint8_t                  m_thread_virt_to_phy[MAX_THREADS_SUPPORTED];
     uint8_t                  m_thread_phy_to_virtual[MAX_THREADS_SUPPORTED];
 
@@ -1069,7 +1069,7 @@ public:
 public:
     /* this is from CLI, number of thread per dual port */
     void set_number_of_threads_per_ports(uint8_t num_threads);
-    void set_latency_thread_is_enabled(bool enable);
+    void set_rx_thread_is_enabled(bool enable);
     void set_number_of_dual_ports(uint8_t num_dual_ports);
 
 
@@ -1085,7 +1085,7 @@ public:
     physical_thread_id_t thread_virt_to_phy(virtual_thread_id_t virt_id);
 
     bool thread_phy_is_master(physical_thread_id_t  phy_id);
-    bool thread_phy_is_latency(physical_thread_id_t  phy_id);
+    bool thread_phy_is_rx(physical_thread_id_t  phy_id);
 
     void dump(FILE *fd);
 
@@ -1246,7 +1246,7 @@ static inline int get_is_rx_check_mode(){
     return (CGlobalInfo::m_options.preview.get_is_rx_check_enable() ?1:0);
 }
 
-static inline bool get_is_rx_filter_enable(){
+static inline bool get_is_rx_filter_enable(){//???
     uint32_t latency_rate=CGlobalInfo::m_options.m_latency_rate;
     return ( ( get_is_rx_check_mode() || CGlobalInfo::is_learn_mode() || latency_rate != 0) ?true:false );
 }
@@ -2902,7 +2902,7 @@ inline void CFlowPktInfo::update_pkt_info(char *p,
 		    }
 		}
             }
-            /* in call cases update the ip using the outside ip */
+            /* in all cases update the ip using the outside ip */
 
             if ( m_pkt_indication.m_desc.IsInitSide()  ) {
 #ifdef NAT_TRACE_
@@ -3579,8 +3579,8 @@ public:
 private:
     void check_msgs(void);
 
-    void handel_nat_msg(CGenNodeNatInfo * msg);
-    void handel_latecy_pkt_msg(CGenNodeLatencyPktInfo * msg);
+    void handle_nat_msg(CGenNodeNatInfo * msg);
+    void handle_latency_pkt_msg(CGenNodeLatencyPktInfo * msg);
 
     void terminate_nat_flows(CGenNode *node);
 
@@ -3653,8 +3653,8 @@ private:
     CGenNodeDeferPort     *          m_tcp_dpc;
     CGenNodeDeferPort     *          m_udp_dpc;
 
-    CNodeRing *                      m_ring_from_rx; /* ring latency thread -> dp */
-    CNodeRing *                      m_ring_to_rx;   /* ring dp -> latency thread */
+    CNodeRing *                      m_ring_from_rx; /* ring rx thread -> dp */
+    CNodeRing *                      m_ring_to_rx;   /* ring dp -> rx thread */
 
     flow_id_node_t                   m_flow_id_to_node_lookup;
 
@@ -3782,8 +3782,8 @@ inline void CCapFileFlowInfo::generate_flow(CTupleTemplateGeneratorSmart   * tup
     }
 
     if ( unlikely(  get_is_rx_check_mode()) ) {
-        if  ( (CGlobalInfo::m_options.m_rx_check_sampe == 1 ) ||
-            ( ( rte_rand() % CGlobalInfo::m_options.m_rx_check_sampe ) == 1 )){
+        if  ( (CGlobalInfo::m_options.m_rx_check_sample == 1 ) ||
+            ( ( rte_rand() % CGlobalInfo::m_options.m_rx_check_sample ) == 1 )){
            if (unlikely(!node->is_repeat_flow() )) {
                node->set_rx_check();
            }
diff --git a/src/debug.cpp b/src/debug.cpp
index 0ca34545..3f8ad2fa 100644
--- a/src/debug.cpp
+++ b/src/debug.cpp
@@ -359,10 +359,9 @@ int CTrexDebug::test_send(uint pkt_type) {
         exit(-1);
     }
 
-    // read first time to zero statistics
     for (port_id = 0; port_id < m_max_ports; port_id++) {
         CPhyEthIF * lp=&m_ports[port_id];
-        lp->get_rx_stats(NULL, -1, true);
+        lp->reset_hw_flow_stats();
     }
 
     printf("Sending packet:\n");
@@ -410,10 +409,10 @@ int CTrexDebug::test_send(uint pkt_type) {
         lp->dump_stats_extended(stdout);
     }
     for (port_id = 0; port_id < m_max_ports; port_id++) {
-        uint64_t fdir_stat[TREX_FDIR_STAT_SIZE];
+        uint64_t fdir_stat[MAX_FLOW_STATS];
         CPhyEthIF *lp = &m_ports[port_id];
-        if (lp->get_rx_stats(fdir_stat, -1, false) == 0)
-            rte_stat_dump_array(fdir_stat, "FDIR stat", TREX_FDIR_STAT_SIZE);
+        if (lp->get_flow_stats(fdir_stat, NULL, 0, MAX_FLOW_STATS, false) == 0)
+            rte_stat_dump_array(fdir_stat, "FDIR stat", MAX_FLOW_STATS);
     }
 
     return (0);
diff --git a/src/flow_stat.cpp b/src/flow_stat.cpp
index e97586f7..7f9df8c7 100644
--- a/src/flow_stat.cpp
+++ b/src/flow_stat.cpp
@@ -33,8 +33,8 @@
 #define FLOW_STAT_ADD_ALL_PORTS 255
 
 static const uint16_t FREE_HW_ID = UINT16_MAX;
+static bool no_stat_supported = true;
 
-#ifdef __DEBUG_FUNC_ENTRY__
 inline std::string methodName(const std::string& prettyFunction)
 {
     size_t colons = prettyFunction.find("::");
@@ -45,6 +45,7 @@ inline std::string methodName(const std::string& prettyFunction)
 }
 
 #define __METHOD_NAME__ methodName(__PRETTY_FUNCTION__)
+#ifdef __DEBUG_FUNC_ENTRY__
 #define FUNC_ENTRY (std::cout << __METHOD_NAME__ << std::endl);
 #else
 #define FUNC_ENTRY
@@ -117,7 +118,7 @@ void CFlowStatUserIdInfo::reset_hw_id() {
         m_rx_counter_base[i] += m_rx_counter[i];
         m_rx_counter[i] = 0;
         m_tx_counter_base[i] += m_tx_counter[i];
-        m_tx_counter[i] = 0;
+        memset(&m_tx_counter[i], 0, sizeof(m_tx_counter[0]));
     }
 }
 /************** class CFlowStatUserIdMap ***************/
@@ -378,6 +379,7 @@ void CFlowStatHwIdMap::unmap(uint16_t hw_id) {
 /************** class CFlowStatRuleMgr ***************/
 CFlowStatRuleMgr::CFlowStatRuleMgr() {
     m_api = NULL;
+    m_max_hw_id = -1;
 }
 
 std::ostream& operator<<(std::ostream& os, const CFlowStatRuleMgr& cf) {
@@ -396,11 +398,11 @@ int CFlowStatRuleMgr::compile_stream(const TrexStream * stream, Cxl710Parser &pa
     // currently we support only IP ID rule types
     // all our ports are the same type, so testing port 0 is enough
     uint16_t num_counters, capabilities;
-    m_api->get_interface_stat_info(0, num_counters, capabilities);    
+    m_api->get_interface_stat_info(0, num_counters, capabilities);
     if ((capabilities & TrexPlatformApi::IF_STAT_IPV4_ID) == 0) {
         return -2;
     }
-        
+
     if (parser.parse(stream->m_pkt.binary, stream->m_pkt.len) != 0) {
         // if we could not parse the packet, but no stat count needed, it is probably OK.
         if (stream->m_rx_check.m_enabled) {
@@ -432,10 +434,24 @@ int CFlowStatRuleMgr::add_stream(const TrexStream * stream) {
     if (! m_api ) {
         TrexStateless *tstateless = get_stateless_obj();
         m_api = tstateless->get_platform_api();
-        //    m_api = get_stateless_obj()->get_platform_api();    
+        uint16_t num_counters, capabilities;
+        m_api->get_interface_stat_info(0, num_counters, capabilities);
+        if ((capabilities & TrexPlatformApi::IF_STAT_IPV4_ID) == 0) {
+            // All our interfaces are from the same type. If statistics not supported.
+            // no operation will work
+            return -1;
+        } else {
+            no_stat_supported = false;
+        }
         m_api->get_port_num(m_num_ports);
+        for (uint8_t port = 0; port < m_num_ports; port++) {
+            assert(m_api->reset_hw_flow_stats(port) == 0);
+        }
     }
-    
+
+    if (no_stat_supported)
+        return -ENOTSUP;
+
     Cxl710Parser parser;
     int ret;
 
@@ -460,6 +476,9 @@ int CFlowStatRuleMgr::del_stream(const TrexStream * stream) {
     std::cout << __METHOD_NAME__ << " user id:" << stream->m_rx_check.m_user_id << std::endl;
 #endif
 
+    if (no_stat_supported)
+        return -ENOTSUP;
+
     if (! stream->m_rx_check.m_enabled) {
         return 0;
     }
@@ -473,7 +492,7 @@ int CFlowStatRuleMgr::del_stream(const TrexStream * stream) {
 // If stream does not need flow stat counting, make sure it does not interfere with
 // other streams that do need stat counting.
 // Might change the IP ID of the stream packet
-int CFlowStatRuleMgr::start_stream(TrexStream * stream) {
+int CFlowStatRuleMgr::start_stream(TrexStream * stream, uint16_t &ret_hw_id) {
 #ifdef __DEBUG_FUNC_ENTRY__
     std::cout << __METHOD_NAME__ << " user id:" << stream->m_rx_check.m_user_id << std::endl;
 #endif
@@ -481,9 +500,8 @@ int CFlowStatRuleMgr::start_stream(TrexStream * stream) {
     Cxl710Parser parser;
     int ret;
 
-    if (! m_api ) {
-        return 0;
-    }
+    if (no_stat_supported)
+        return -ENOTSUP;
 
     if ((ret = compile_stream(stream, parser)) < 0)
         return ret;
@@ -509,6 +527,9 @@ int CFlowStatRuleMgr::start_stream(TrexStream * stream) {
         m_user_id_map.start_stream(stream->m_rx_check.m_user_id); // just increase ref count;
     } else {
         uint16_t hw_id = m_hw_id_map.find_free_hw_id();
+        if (hw_id > m_max_hw_id) {
+            m_max_hw_id = hw_id;
+        }
         if (hw_id == FREE_HW_ID) {
             printf("Error: %s failed finding free hw_id\n", __func__);
             return -1;
@@ -523,10 +544,16 @@ int CFlowStatRuleMgr::start_stream(TrexStream * stream) {
     uint16_t hw_id = m_user_id_map.get_hw_id(stream->m_rx_check.m_user_id); // can't fail if we got here
     parser.set_ip_id(IP_ID_RESERVE_BASE + hw_id);
 
+    ret_hw_id = hw_id;
+
+#ifdef __DEBUG_FUNC_ENTRY__
+    std::cout << "exit:" << __METHOD_NAME__ << " hw_id:" << ret_hw_id << std::endl;
+#endif
+
     return 0;
 }
 
-int CFlowStatRuleMgr::add_hw_rule(uint16_t hw_id, uint8_t proto) {    
+int CFlowStatRuleMgr::add_hw_rule(uint16_t hw_id, uint8_t proto) {
     for (int port = 0; port < m_num_ports; port++) {
         m_api->add_rx_flow_stat_rule(port, FLOW_STAT_RULE_TYPE_IPV4_ID, proto, hw_id);
     }
@@ -538,18 +565,18 @@ int CFlowStatRuleMgr::stop_stream(const TrexStream * stream) {
 #ifdef __DEBUG_FUNC_ENTRY__
     std::cout << __METHOD_NAME__ << " user id:" << stream->m_rx_check.m_user_id << std::endl;
 #endif
+    if (no_stat_supported)
+        return -ENOTSUP;
+
     if (! stream->m_rx_check.m_enabled) {
         return 0;
     }
-    if (! m_api ) {
-        return 0;
-    }
 
     if (m_user_id_map.stop_stream(stream->m_rx_check.m_user_id) == 0) {
         // last stream associated with the entry stopped transmittig.
         // remove user_id <--> hw_id mapping
         uint8_t proto = m_user_id_map.l4_proto(stream->m_rx_check.m_user_id);
-        uint16_t hw_id = m_user_id_map.unmap(stream->m_rx_check.m_user_id);
+        uint16_t hw_id = m_user_id_map.get_hw_id(stream->m_rx_check.m_user_id);
         if (hw_id >= MAX_FLOW_STATS) {
             fprintf(stderr, "Error: %s got wrong hw_id %d from unmap\n", __func__, hw_id);
             return -1;
@@ -557,13 +584,15 @@ int CFlowStatRuleMgr::stop_stream(const TrexStream * stream) {
             // update counters, and reset before unmapping
             class CFlowStatUserIdInfo *p_user_id = m_user_id_map.find_user_id(m_hw_id_map.get_user_id(hw_id));
             assert(p_user_id != NULL);
-            uint64_t counter;
+            uint64_t rx_counter;
+            tx_per_flow_t tx_counter;
             for (uint8_t port = 0; port < m_num_ports; port++) {
                 m_api->del_rx_flow_stat_rule(port, FLOW_STAT_RULE_TYPE_IPV4_ID, proto, hw_id);
-                m_api->get_rx_stats(port, &counter, hw_id, true);
-                p_user_id->set_rx_counter(port, counter);
-                p_user_id->set_tx_counter(port, counter); //??? until tx work, just set for same value
+                m_api->get_flow_stats(port, &rx_counter, (void *)&tx_counter, hw_id, hw_id, true);
+                p_user_id->set_rx_counter(port, rx_counter);
+                p_user_id->set_tx_counter(port, tx_counter);
             }
+            m_user_id_map.unmap(stream->m_rx_check.m_user_id);
             m_hw_id_map.unmap(hw_id);
         }
     }
@@ -572,12 +601,13 @@ int CFlowStatRuleMgr::stop_stream(const TrexStream * stream) {
 
 // return false if no counters changed since last run. true otherwise
 bool CFlowStatRuleMgr::dump_json(std::string & json) {
-    uint64_t stats[TREX_FDIR_STAT_SIZE];
+    uint64_t rx_stats[MAX_FLOW_STATS];
+    tx_per_flow_t tx_stats[MAX_FLOW_STATS];
     Json::FastWriter writer;
     Json::Value root;
     bool ret = false;
 
-    if (! m_api ) {
+    if (m_user_id_map.is_empty()) {
         return false;
     }
     root["name"] = "rx-stats";
@@ -587,15 +617,24 @@ bool CFlowStatRuleMgr::dump_json(std::string & json) {
     // read hw counters, and update
     data_section["timestamp"] = Json::Value::UInt64(os_get_hr_tick_64());
     for (uint8_t port = 0; port < m_num_ports; port++) {
-        int rc = m_api->get_rx_stats(port, stats, -1, false);
-        if (rc == -1) {
-            continue;
-        }
-
-        for (int i = 0; i < TREX_FDIR_STAT_SIZE; i++) {
-            if (stats[i] != 0) {
-                m_user_id_map.find_user_id(m_hw_id_map.get_user_id(i))->set_rx_counter(port, stats[i]);
-                m_user_id_map.find_user_id(m_hw_id_map.get_user_id(i))->set_tx_counter(port, stats[i]);  //??? until tx work, just set for same value
+        m_api->get_flow_stats(port, rx_stats, (void *)tx_stats, 0, m_max_hw_id, false);
+        for (int i = 0; i <= m_max_hw_id; i++) {
+            if (rx_stats[i] != 0) {
+                class CFlowStatUserIdInfo *p_user_id = m_user_id_map.find_user_id(m_hw_id_map.get_user_id(i));
+                if (likely(p_user_id != NULL)) {
+                    p_user_id->set_rx_counter(port, rx_stats[i]);
+                } else {
+                    std::cerr <<  __METHOD_NAME__ << i << ":Could not count " << rx_stats[i] << " rx packets, because no mapping was found" << std::endl;
+                }
+            }
+            if (tx_stats[i].get_pkts() != 0) {
+                tx_per_flow_t tx_pkts = tx_stats[i];
+                class CFlowStatUserIdInfo *p_user_id = m_user_id_map.find_user_id(m_hw_id_map.get_user_id(i));
+                if (likely(p_user_id != NULL)) {
+                    p_user_id->set_tx_counter(port, tx_pkts);
+                } else {
+                    std::cerr <<  __METHOD_NAME__ << i << ":Could not count tx " << tx_pkts << " because no mapping was found" << std::endl;
+                }
             }
         }
     }
@@ -608,11 +647,14 @@ bool CFlowStatRuleMgr::dump_json(std::string & json) {
         std::string str_user_id = static_cast<std::ostringstream*>( &(std::ostringstream()
                                                                       << user_id) )->str();
         for (uint8_t port = 0; port < m_num_ports; port++) {
-            if ((user_id_info->get_tx_counter(port) != 0) || (user_id_info->get_rx_counter(port) != 0)) {
-                std::string str_port = static_cast<std::ostringstream*>( &(std::ostringstream()
-                                                                           << port) )->str();
-                data_section[str_user_id]["rx"][str_port] = Json::Value::UInt64(user_id_info->get_rx_counter(port));
-                data_section[str_user_id]["tx"][str_port] = Json::Value::UInt64(user_id_info->get_tx_counter(port));
+            std::string str_port = static_cast<std::ostringstream*>( &(std::ostringstream() << port) )->str();
+            if (user_id_info->get_rx_counter(port) != 0) {
+                data_section[str_user_id]["rx-pkts"][str_port] = Json::Value::UInt64(user_id_info->get_rx_counter(port));
+                ret = true;
+            }
+            if (user_id_info->get_tx_counter(port).get_pkts() != 0) {
+                data_section[str_user_id]["tx-pkts"][str_port] = Json::Value::UInt64(user_id_info->get_tx_counter(port).get_pkts());
+                data_section[str_user_id]["tx-bytes"][str_port] = Json::Value::UInt64(user_id_info->get_tx_counter(port).get_bytes());
                 ret = true;
             }
         }
diff --git a/src/flow_stat.h b/src/flow_stat.h
index 444daab0..eed3b797 100644
--- a/src/flow_stat.h
+++ b/src/flow_stat.h
@@ -25,8 +25,9 @@
 #include <string>
 #include <map>
 #include "trex_defs.h"
+#include "trex_stream.h"
+#include <internal_api/trex_platform_api.h>
 
-#define MAX_FLOW_STATS 128
 // range reserved for rx stat measurement is from IP_ID_RESERVE_BASE to 0xffff
 // Do not change this value. In i350 cards, we filter according to first byte of IP ID
 // In other places, we identify packets by if (ip_id > IP_ID_RESERVE_BASE)
@@ -35,6 +36,65 @@
 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 tx_per_flow_t_ {
+ public:
+    tx_per_flow_t_() {
+        clear();
+    }
+    inline uint64_t get_bytes() {
+        return m_bytes;
+    }
+    inline uint64_t get_pkts() {
+        return m_pkts;
+    }
+    inline void set_bytes(uint64_t bytes) {
+        m_bytes = bytes;;
+    }
+    inline void get_pkts(uint64_t pkts) {
+        m_pkts = pkts;
+    }
+    inline void add_bytes(uint64_t bytes) {
+        m_bytes += bytes;;
+    }
+    inline void add_pkts(uint64_t pkts) {
+        m_pkts += pkts;
+    }
+    inline void clear() {
+        m_bytes = 0;
+        m_pkts = 0;
+    }
+    inline tx_per_flow_t_ operator+ (const tx_per_flow_t_ &t_in) {
+        tx_per_flow_t_ t_out;
+        t_out.m_bytes = this->m_bytes + t_in.m_bytes;
+        t_out.m_pkts = this->m_pkts + t_in.m_pkts;
+        return t_out;
+    }
+
+    inline tx_per_flow_t_ operator- (const tx_per_flow_t_ &t_in) {
+        tx_per_flow_t_ t_out;
+        t_out.m_bytes = this->m_bytes - t_in.m_bytes;
+        t_out.m_pkts = this->m_pkts - t_in.m_pkts;
+        return t_out;
+    }
+
+    inline tx_per_flow_t_ operator+= (const tx_per_flow_t_ &t_in) {
+        m_bytes += t_in.m_bytes;
+        m_pkts += t_in.m_pkts;
+        return *this;
+    }
+
+    friend std::ostream& operator<<(std::ostream& os, const class tx_per_flow_t_ &t) {
+        os  << "p:" << t.m_pkts << " b:" << t.m_bytes;
+        return os;
+    }
+
+ private:
+    uint64_t m_bytes;
+    uint64_t m_pkts;
+};
+
+typedef class tx_per_flow_t_ tx_per_flow_t;
+
 class CPhyEthIF;
 class Cxl710Parser;
 
@@ -44,8 +104,8 @@ class CFlowStatUserIdInfo {
     friend std::ostream& operator<<(std::ostream& os, const CFlowStatUserIdInfo& cf);
     void set_rx_counter(uint8_t port, uint64_t val) {m_rx_counter[port] = val;}
     uint64_t get_rx_counter(uint8_t port) {return m_rx_counter[port] + m_rx_counter_base[port];}
-    void set_tx_counter(uint8_t port, uint64_t val) {m_tx_counter[port] = val;}
-    uint64_t get_tx_counter(uint8_t port) {return m_tx_counter[port] + m_tx_counter_base[port];}
+    void set_tx_counter(uint8_t port, tx_per_flow_t val) {m_tx_counter[port] = val;}
+    tx_per_flow_t get_tx_counter(uint8_t port) {return m_tx_counter[port] + m_tx_counter_base[port];}
     void set_hw_id(uint16_t hw_id) {m_hw_id = hw_id;}
     uint64_t get_hw_id() {return m_hw_id;}
     void reset_hw_id();
@@ -62,9 +122,9 @@ class CFlowStatUserIdInfo {
     uint64_t m_rx_counter[TREX_MAX_PORTS]; // How many packets received with this user id since stream start
     // How many packets received with this user id, since stream creation, before stream start.
     uint64_t m_rx_counter_base[TREX_MAX_PORTS];
-    uint64_t m_tx_counter[TREX_MAX_PORTS]; // How many packets transmitted with this user id since stream start
+    tx_per_flow_t m_tx_counter[TREX_MAX_PORTS]; // How many packets transmitted with this user id since stream start
     // How many packets transmitted with this user id, since stream creation, before stream start.
-    uint64_t m_tx_counter_base[TREX_MAX_PORTS];
+    tx_per_flow_t m_tx_counter_base[TREX_MAX_PORTS];
     uint16_t m_hw_id;     // Associated hw id. UINT16_MAX if no associated hw id.
     uint8_t m_proto;      // protocol (UDP, TCP, other), associated with this user id.
     uint8_t m_ref_count;  // How many streams with this ref count exists
@@ -78,6 +138,7 @@ class CFlowStatUserIdMap {
  public:
     CFlowStatUserIdMap();
     friend std::ostream& operator<<(std::ostream& os, const CFlowStatUserIdMap& cf);
+    bool is_empty() {return (m_map.empty() == true);};
     uint16_t get_hw_id(uint32_t user_id);
     class CFlowStatUserIdInfo * find_user_id(uint32_t user_id);
     class CFlowStatUserIdInfo * add_user_id(uint32_t user_id, uint8_t proto);
@@ -121,7 +182,7 @@ class CFlowStatRuleMgr {
     friend std::ostream& operator<<(std::ostream& os, const CFlowStatRuleMgr& cf);
     int add_stream(const TrexStream * stream);
     int del_stream(const TrexStream * stream);
-    int start_stream(TrexStream * stream);
+    int start_stream(TrexStream * stream, uint16_t &ret_hw_id);
     int stop_stream(const TrexStream * stream);
     bool dump_json(std::string & json);
 
@@ -134,6 +195,7 @@ class CFlowStatRuleMgr {
     class 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;
+    int m_max_hw_id; // max hw id we ever used
 };
 
 #endif
diff --git a/src/gtest/trex_stateless_gtest.cpp b/src/gtest/trex_stateless_gtest.cpp
index 4cc40cdb..c3dfcb95 100644
--- a/src/gtest/trex_stateless_gtest.cpp
+++ b/src/gtest/trex_stateless_gtest.cpp
@@ -3206,7 +3206,6 @@ public:
         /* first the message must be an event */
         TrexDpPortEventMsg *event = dynamic_cast<TrexDpPortEventMsg *>(msg);
         EXPECT_TRUE(event != NULL);
-        EXPECT_TRUE(event->get_event_type() == TrexDpPortEvent::EVENT_STOP);
 
         EXPECT_TRUE(event->get_event_id() == m_event_id);
         EXPECT_TRUE(event->get_port_id() == 0);
diff --git a/src/internal_api/trex_platform_api.h b/src/internal_api/trex_platform_api.h
index 847611e4..f6d7278e 100644
--- a/src/internal_api/trex_platform_api.h
+++ b/src/internal_api/trex_platform_api.h
@@ -141,12 +141,14 @@ public:
     virtual void publish_async_data_now(uint32_t key) const = 0;
     virtual uint8_t get_dp_core_count() const = 0;
     virtual void get_interface_stat_info(uint8_t interface_id, uint16_t &num_counters, uint16_t &capabilities) const =0;
-    virtual int get_rx_stats(uint8_t port_id, uint64_t *stats, int index, bool reset) const = 0;
+    virtual int get_flow_stats(uint8_t port_id, uint64_t *stats, void *tx_stats, int min, int max, bool reset) const = 0;
+    virtual int reset_hw_flow_stats(uint8_t port_id) const = 0;
     virtual void get_port_num(uint8_t &port_num) const = 0;
     virtual int add_rx_flow_stat_rule(uint8_t port_id, uint8_t type, uint16_t proto, uint16_t id) const = 0;
     virtual int del_rx_flow_stat_rule(uint8_t port_id, uint8_t type, uint16_t proto, uint16_t id) const = 0;
     virtual void set_promiscuous(uint8_t port_id, bool enabled) const = 0;
     virtual bool get_promiscuous(uint8_t port_id) const = 0;
+    virtual void flush_dp_messages() const = 0;
 
     virtual ~TrexPlatformApi() {}
 };
@@ -168,12 +170,14 @@ public:
     void publish_async_data_now(uint32_t key) const;
     uint8_t get_dp_core_count() const;
     void get_interface_stat_info(uint8_t interface_id, uint16_t &num_counters, uint16_t &capabilities) const;
-    int get_rx_stats(uint8_t port_id, uint64_t *stats, int index, bool reset) const;
+    int get_flow_stats(uint8_t port_id, uint64_t *stats, void *tx_stats, int min, int max, bool reset) const;
+    int reset_hw_flow_stats(uint8_t port_id) const;
     void get_port_num(uint8_t &port_num) const;
     int add_rx_flow_stat_rule(uint8_t port_id, uint8_t type, uint16_t proto, uint16_t id) const;
     int del_rx_flow_stat_rule(uint8_t port_id, uint8_t type, uint16_t proto, uint16_t id) const;
     void set_promiscuous(uint8_t port_id, bool enabled) const;
     bool get_promiscuous(uint8_t port_id) const;
+    void flush_dp_messages() const;
 };
 
 
@@ -219,7 +223,8 @@ public:
     virtual void publish_async_data_now(uint32_t key) const {
 
     }
-    virtual int get_rx_stats(uint8_t port_id, uint64_t *stats, int index, bool reset) const {return 0;}
+    virtual int get_flow_stats(uint8_t port_id, uint64_t *stats, void *tx_stats, int min, int max, bool reset) const {return 0;};
+    virtual int reset_hw_flow_stats(uint8_t port_id) const {return 0;};
     virtual void get_port_num(uint8_t &port_num) const {port_num = 2;};
     virtual int add_rx_flow_stat_rule(uint8_t port_id, uint8_t type, uint16_t proto, uint16_t id) const {return 0;}
     virtual int del_rx_flow_stat_rule(uint8_t port_id, uint8_t type, uint16_t proto, uint16_t id) const {return 0;}
@@ -231,6 +236,9 @@ public:
         return false;
     }
 
+    void flush_dp_messages() const {
+    }
+
 private:
     int m_dp_core_count;
 };
diff --git a/src/latency.cpp b/src/latency.cpp
index fb3c6aac..d57e97c8 100644
--- a/src/latency.cpp
+++ b/src/latency.cpp
@@ -231,11 +231,9 @@ void CCPortLatency::update_packet(rte_mbuf_t * m, int port_id){
 }
 
 
-void CCPortLatency::DumpShortHeader(FILE *fd){
-    
-
-    fprintf(fd," if|   tx_ok , rx_ok  , rx   ,error,    average   ,   max         , Jitter ,  max window \n");
-	fprintf(fd,"   |         ,        , check,     , latency(usec),latency (usec) ,(usec)  ,             \n");
+void CCPortLatency::DumpShortHeader(FILE *fd){   
+    fprintf(fd," if|   tx_ok , rx_ok  , rx check ,error,       latency (usec) ,    Jitter          max window \n");
+	fprintf(fd,"   |         ,        ,          ,     ,   average   ,   max  ,    (usec)                     \n");
     fprintf(fd," ---------------------------------------------------------------------------------------------------------------- \n");
 }
 
@@ -269,7 +267,7 @@ void CCPortLatency::dump_json(std::string & json ){
 void CCPortLatency::DumpShort(FILE *fd){
 
 //	m_hist.update(); <- moved to CLatencyManager::update()
-    fprintf(fd,"%8lu,%8lu,%10lu,%4lu,",                          
+    fprintf(fd,"%8lu,%8lu,%10lu,%5lu,",                          
                     m_tx_pkt_ok,
                     m_pkt_ok,
                     m_rx_check,
@@ -530,7 +528,7 @@ bool CLatencyManager::Create(CLatencyManagerCfg * cfg){
     }
     
     m_max_ports=cfg->m_max_ports;
-    assert (m_max_ports<=MAX_LATENCY_PORTS);
+    assert (m_max_ports <= TREX_MAX_PORTS);
     assert ((m_max_ports%2)==0);
     m_port_mask =0xffffffff;
     m_do_stop =false;
diff --git a/src/latency.h b/src/latency.h
index cbb67dbd..1f8ef5c0 100644
--- a/src/latency.h
+++ b/src/latency.h
@@ -273,7 +273,7 @@ public:
     }
     uint32_t             m_max_ports;
     double               m_cps;// CPS
-    CPortLatencyHWBase * m_ports[MAX_LATENCY_PORTS];
+    CPortLatencyHWBase * m_ports[TREX_MAX_PORTS];
     ipaddr_t             m_client_ip;
     ipaddr_t             m_server_ip;
     uint32_t             m_dual_port_mask;
@@ -387,7 +387,7 @@ private:
      pqueue_t                m_p_queue; /* priorty queue */
      bool                    m_is_active;
      CLatencyPktInfo         m_pkt_gen;
-     CLatencyManagerPerPort  m_ports[MAX_LATENCY_PORTS];
+     CLatencyManagerPerPort  m_ports[TREX_MAX_PORTS];
      uint64_t                m_d_time; // calc tick betwen sending 
      double                  m_cps;
      double                  m_delta_sec;
diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp
index 5c7be19e..57b3719b 100644
--- a/src/main_dpdk.cpp
+++ b/src/main_dpdk.cpp
@@ -1,22 +1,22 @@
 /*
- Hanoh Haim
- Cisco Systems, Inc.
+  Hanoh Haim
+  Cisco Systems, Inc.
 */
 
 /*
-Copyright (c) 2015-2016 Cisco Systems, Inc.
+  Copyright (c) 2015-2016 Cisco Systems, Inc.
 
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
+  Licensed under the Apache License, Version 2.0 (the "License");
+  you may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
 
-    http://www.apache.org/licenses/LICENSE-2.0
+  http://www.apache.org/licenses/LICENSE-2.0
 
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
 */
 #include <pwd.h>
 #include <rte_common.h>
@@ -87,8 +87,6 @@ extern "C" {
 #define BP_MAX_PKT      32
 #define MAX_PKT_BURST   32
 
-
-#define BP_MAX_PORTS (MAX_LATENCY_PORTS)
 #define BP_MAX_CORES 32
 #define BP_MAX_TX_QUEUE 16
 #define BP_MASTER_AND_LATENCY 2
@@ -106,12 +104,14 @@ extern "C" void i40e_set_trex_mode(int mode);
 #define RTE_TEST_TX_DESC_DEFAULT 512
 #define RTE_TEST_RX_DESC_DROP    0
 
+static int max_stat_hw_id_seen = 0;
+
 static inline int get_vm_one_queue_enable(){
     return (CGlobalInfo::m_options.preview.get_vm_one_queue_enable() ?1:0);
 }
 
-static inline int get_is_latency_thread_enable(){
-    return (CGlobalInfo::m_options.is_latency_enabled() ?1:0);
+static inline int get_is_rx_thread_enabled() {
+    return (CGlobalInfo::m_options.is_rx_enabled() ?1:0);
 }
 
 struct port_cfg_t;
@@ -145,7 +145,7 @@ public:
     virtual int  wait_for_stable_link()=0;
     virtual void wait_after_link_up(){};
     virtual bool flow_control_disable_supported(){return true;}
-    virtual int get_rx_stats(CPhyEthIF * _if, uint32_t *stats, uint32_t *prev_stats, int index) {return -1;}
+    virtual int get_rx_stats(CPhyEthIF * _if, uint32_t *stats, uint32_t *prev_stats, int min, int max) {return -1;}
     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;}
@@ -181,7 +181,7 @@ public:
     virtual int configure_rx_filter_rules(CPhyEthIF * _if);
     int configure_rx_filter_rules_statefull(CPhyEthIF * _if);
     int configure_rx_filter_rules_stateless(CPhyEthIF * _if);
-    
+
     virtual bool is_hardware_support_drop_queue(){
         return(true);
     }
@@ -215,7 +215,7 @@ public:
     }
 
     virtual void update_global_config_fdir(port_cfg_t * cfg){
-        
+
     }
 
     virtual int get_min_sample_rate(void){
@@ -318,14 +318,14 @@ public:
     }
     virtual void get_extended_stats(CPhyEthIF * _if,CPhyEthIFStats *stats);
     virtual void clear_extended_stats(CPhyEthIF * _if);
-    int get_rx_stats(CPhyEthIF * _if, uint32_t *stats, uint32_t *prev_stats, int index);
+    int get_rx_stats(CPhyEthIF * _if, uint32_t *stats, uint32_t *prev_stats, int min, int max);
     int dump_fdir_global_stats(CPhyEthIF * _if, FILE *fd);
-    int get_stat_counters_num() {return TREX_FDIR_STAT_SIZE;}
+    int get_stat_counters_num() {return MAX_FLOW_STATS;}
     int get_rx_stat_capabilities() {return TrexPlatformApi::IF_STAT_IPV4_ID;}
     virtual int wait_for_stable_link();
     // disabling flow control on 40G using DPDK API causes the interface to malfunction
     bool flow_control_disable_supported(){return false;}
-private:    
+private:
     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:
@@ -364,31 +364,31 @@ public:
 class CTRexExtendedDriverDb {
 public:
 
-   const std::string & get_driver_name() {
-       return m_driver_name;
-   }
+    const std::string & get_driver_name() {
+        return m_driver_name;
+    }
 
-   bool is_driver_exists(std::string name);
+    bool is_driver_exists(std::string name);
 
 
 
-   void set_driver_name(std::string name){
-       m_driver_was_set=true;
-       m_driver_name=name;
-       printf(" set driver name %s \n",name.c_str());
-       m_drv=create_driver(m_driver_name);
-       assert(m_drv);
-   }
+    void set_driver_name(std::string name){
+        m_driver_was_set=true;
+        m_driver_name=name;
+        printf(" set driver name %s \n",name.c_str());
+        m_drv=create_driver(m_driver_name);
+        assert(m_drv);
+    }
 
-   CTRexExtendedDriverBase * get_drv(){
-       if (!m_driver_was_set) {
-           printf(" ERROR too early to use this object !\n");
-           printf(" need to set the right driver \n");
-           assert(0);
-       }
-       assert(m_drv);
-       return (m_drv);
-   }
+    CTRexExtendedDriverBase * get_drv(){
+        if (!m_driver_was_set) {
+            printf(" ERROR too early to use this object !\n");
+            printf(" need to set the right driver \n");
+            assert(0);
+        }
+        assert(m_drv);
+        return (m_drv);
+    }
 
 public:
 
@@ -407,7 +407,7 @@ private:
         register_driver(std::string("rte_vmxnet3_pmd"),CTRexExtendedDriverBase1GVm::create);
         register_driver(std::string("rte_virtio_pmd"),CTRexExtendedDriverBase1GVm::create);
         register_driver(std::string("rte_enic_pmd"),CTRexExtendedDriverBaseVIC::create);
-        
+
 
 
 
@@ -486,49 +486,49 @@ static char global_loglevel_str[20];
 
 // cores =0==1,1*2,2,3,4,5,6
 // An enum for all the option types
-enum { OPT_HELP, 
-    OPT_MODE_BATCH, 
-    OPT_MODE_INTERACTIVE,
-    OPT_NODE_DUMP,  
-    OPT_UT,
-    OPT_FILE_OUT,
-    OPT_REAL_TIME,
-    OPT_CORES,
-    OPT_SINGLE_CORE,
-    OPT_FLIP_CLIENT_SERVER,
-    OPT_FLOW_FLIP_CLIENT_SERVER,
-    OPT_FLOW_FLIP_CLIENT_SERVER_SIDE,
-    OPT_BW_FACTOR,
-    OPT_DURATION,
-    OPT_PLATFORM_FACTOR,
-    OPT_PUB_DISABLE,
-    OPT_LIMT_NUM_OF_PORTS,
-    OPT_PLAT_CFG_FILE,
-
-
-    OPT_LATENCY,
-    OPT_NO_CLEAN_FLOW_CLOSE,
-    OPT_LATENCY_MASK,
-    OPT_ONLY_LATENCY,
-    OPT_1G_MODE,
-    OPT_LATENCY_PREVIEW ,
-    OPT_PCAP,
-	OPT_RX_CHECK,
-    OPT_IO_MODE,
-    OPT_IPV6,
-    OPT_LEARN,
-    OPT_LEARN_MODE,
-    OPT_LEARN_VERIFY,
-    OPT_L_PKT_MODE,
-    OPT_NO_FLOW_CONTROL,
-    OPT_RX_CHECK_HOPS,
-	OPT_MAC_FILE,
-    OPT_NO_KEYBOARD_INPUT,
-	OPT_VLAN,
-    OPT_VIRT_ONE_TX_RX_QUEUE,
-    OPT_PREFIX,
-    OPT_MAC_SPLIT,
-    OPT_SEND_DEBUG_PKT
+enum { OPT_HELP,
+       OPT_MODE_BATCH,
+       OPT_MODE_INTERACTIVE,
+       OPT_NODE_DUMP,
+       OPT_UT,
+       OPT_FILE_OUT,
+       OPT_REAL_TIME,
+       OPT_CORES,
+       OPT_SINGLE_CORE,
+       OPT_FLIP_CLIENT_SERVER,
+       OPT_FLOW_FLIP_CLIENT_SERVER,
+       OPT_FLOW_FLIP_CLIENT_SERVER_SIDE,
+       OPT_BW_FACTOR,
+       OPT_DURATION,
+       OPT_PLATFORM_FACTOR,
+       OPT_PUB_DISABLE,
+       OPT_LIMT_NUM_OF_PORTS,
+       OPT_PLAT_CFG_FILE,
+
+
+       OPT_LATENCY,
+       OPT_NO_CLEAN_FLOW_CLOSE,
+       OPT_LATENCY_MASK,
+       OPT_ONLY_LATENCY,
+       OPT_1G_MODE,
+       OPT_LATENCY_PREVIEW ,
+       OPT_PCAP,
+       OPT_RX_CHECK,
+       OPT_IO_MODE,
+       OPT_IPV6,
+       OPT_LEARN,
+       OPT_LEARN_MODE,
+       OPT_LEARN_VERIFY,
+       OPT_L_PKT_MODE,
+       OPT_NO_FLOW_CONTROL,
+       OPT_RX_CHECK_HOPS,
+       OPT_MAC_FILE,
+       OPT_NO_KEYBOARD_INPUT,
+       OPT_VLAN,
+       OPT_VIRT_ONE_TX_RX_QUEUE,
+       OPT_PREFIX,
+       OPT_MAC_SPLIT,
+       OPT_SEND_DEBUG_PKT
 };
 
 
@@ -538,60 +538,60 @@ enum { OPT_HELP,
    SO_MULTI --   multiple arguments needed
 */
 static CSimpleOpt::SOption parser_options[] =
-{
-    { OPT_HELP,                   "-?",                SO_NONE   },
-    { OPT_HELP,                   "-h",                SO_NONE   },
-    { OPT_HELP,                   "--help",            SO_NONE   },
-    { OPT_UT,                     "--ut",              SO_NONE   },
-    { OPT_MODE_BATCH,             "-f",                SO_REQ_SEP},
-    { OPT_MODE_INTERACTIVE,       "-i",                SO_NONE   },
-    { OPT_PLAT_CFG_FILE,          "--cfg",             SO_REQ_SEP},
-    { OPT_REAL_TIME ,             "-r",                SO_NONE  },
-    { OPT_SINGLE_CORE,            "-s",                SO_NONE  },
-    { OPT_FILE_OUT,               "-o" ,               SO_REQ_SEP},
-    { OPT_FLIP_CLIENT_SERVER,"--flip",SO_NONE  },
-    { OPT_FLOW_FLIP_CLIENT_SERVER,"-p",SO_NONE  },
-    { OPT_FLOW_FLIP_CLIENT_SERVER_SIDE,"-e",SO_NONE  },
-
-    { OPT_NO_CLEAN_FLOW_CLOSE,"--nc",SO_NONE  },
-
-    { OPT_LIMT_NUM_OF_PORTS,"--limit-ports", SO_REQ_SEP },
-    { OPT_CORES     , "-c",         SO_REQ_SEP },
-    { OPT_NODE_DUMP , "-v",         SO_REQ_SEP },
-    { OPT_LATENCY , "-l",         SO_REQ_SEP },
-
-    { OPT_DURATION     , "-d",  SO_REQ_SEP },
-    { OPT_PLATFORM_FACTOR     , "-pm",  SO_REQ_SEP },
-
-    { OPT_PUB_DISABLE     , "-pubd",  SO_NONE },
-
-
-    { OPT_BW_FACTOR     , "-m",  SO_REQ_SEP },
-    { OPT_LATENCY_MASK     , "--lm",  SO_REQ_SEP },
-    { OPT_ONLY_LATENCY, "--lo",  SO_NONE  },
-
-    { OPT_1G_MODE,       "-1g",   SO_NONE   },
-    { OPT_LATENCY_PREVIEW ,       "-k",   SO_REQ_SEP   },
-    { OPT_PCAP,       "--pcap",       SO_NONE   },
-	{ OPT_RX_CHECK,   "--rx-check",  SO_REQ_SEP },
-    { OPT_IO_MODE,   "--iom",  SO_REQ_SEP },      
-    { OPT_RX_CHECK_HOPS, "--hops", SO_REQ_SEP },
-    { OPT_IPV6,       "--ipv6",       SO_NONE   },
-    { OPT_LEARN, "--learn",       SO_NONE   },
-    { OPT_LEARN_MODE, "--learn-mode",       SO_REQ_SEP   },
-    { OPT_LEARN_VERIFY, "--learn-verify",       SO_NONE   },
-    { OPT_L_PKT_MODE, "--l-pkt-mode",       SO_REQ_SEP   },
-    { OPT_NO_FLOW_CONTROL, "--no-flow-control-change",       SO_NONE   },
-    { OPT_VLAN,       "--vlan",       SO_NONE   },
-    { OPT_MAC_FILE, "--mac", SO_REQ_SEP }, 
-    { OPT_NO_KEYBOARD_INPUT ,"--no-key", SO_NONE   },
-    { OPT_VIRT_ONE_TX_RX_QUEUE, "--vm-sim", SO_NONE }, 
-    { OPT_PREFIX, "--prefix", SO_REQ_SEP }, 
-    { OPT_MAC_SPLIT, "--mac-spread", SO_REQ_SEP },
-    { OPT_SEND_DEBUG_PKT, "--send-debug-pkt", SO_REQ_SEP },
-
-    SO_END_OF_OPTIONS
-};
+    {
+        { OPT_HELP,                   "-?",                SO_NONE   },
+        { OPT_HELP,                   "-h",                SO_NONE   },
+        { OPT_HELP,                   "--help",            SO_NONE   },
+        { OPT_UT,                     "--ut",              SO_NONE   },
+        { OPT_MODE_BATCH,             "-f",                SO_REQ_SEP},
+        { OPT_MODE_INTERACTIVE,       "-i",                SO_NONE   },
+        { OPT_PLAT_CFG_FILE,          "--cfg",             SO_REQ_SEP},
+        { OPT_REAL_TIME ,             "-r",                SO_NONE  },
+        { OPT_SINGLE_CORE,            "-s",                SO_NONE  },
+        { OPT_FILE_OUT,               "-o" ,               SO_REQ_SEP},
+        { OPT_FLIP_CLIENT_SERVER,"--flip",SO_NONE  },
+        { OPT_FLOW_FLIP_CLIENT_SERVER,"-p",SO_NONE  },
+        { OPT_FLOW_FLIP_CLIENT_SERVER_SIDE,"-e",SO_NONE  },
+
+        { OPT_NO_CLEAN_FLOW_CLOSE,"--nc",SO_NONE  },
+
+        { OPT_LIMT_NUM_OF_PORTS,"--limit-ports", SO_REQ_SEP },
+        { OPT_CORES     , "-c",         SO_REQ_SEP },
+        { OPT_NODE_DUMP , "-v",         SO_REQ_SEP },
+        { OPT_LATENCY , "-l",         SO_REQ_SEP },
+
+        { OPT_DURATION     , "-d",  SO_REQ_SEP },
+        { OPT_PLATFORM_FACTOR     , "-pm",  SO_REQ_SEP },
+
+        { OPT_PUB_DISABLE     , "-pubd",  SO_NONE },
+
+
+        { OPT_BW_FACTOR     , "-m",  SO_REQ_SEP },
+        { OPT_LATENCY_MASK     , "--lm",  SO_REQ_SEP },
+        { OPT_ONLY_LATENCY, "--lo",  SO_NONE  },
+
+        { OPT_1G_MODE,       "-1g",   SO_NONE   },
+        { OPT_LATENCY_PREVIEW ,       "-k",   SO_REQ_SEP   },
+        { OPT_PCAP,       "--pcap",       SO_NONE   },
+        { OPT_RX_CHECK,   "--rx-check",  SO_REQ_SEP },
+        { OPT_IO_MODE,   "--iom",  SO_REQ_SEP },
+        { OPT_RX_CHECK_HOPS, "--hops", SO_REQ_SEP },
+        { OPT_IPV6,       "--ipv6",       SO_NONE   },
+        { OPT_LEARN, "--learn",       SO_NONE   },
+        { OPT_LEARN_MODE, "--learn-mode",       SO_REQ_SEP   },
+        { OPT_LEARN_VERIFY, "--learn-verify",       SO_NONE   },
+        { OPT_L_PKT_MODE, "--l-pkt-mode",       SO_REQ_SEP   },
+        { OPT_NO_FLOW_CONTROL, "--no-flow-control-change",       SO_NONE   },
+        { OPT_VLAN,       "--vlan",       SO_NONE   },
+        { OPT_MAC_FILE, "--mac", SO_REQ_SEP },
+        { OPT_NO_KEYBOARD_INPUT ,"--no-key", SO_NONE   },
+        { OPT_VIRT_ONE_TX_RX_QUEUE, "--vm-sim", SO_NONE },
+        { OPT_PREFIX, "--prefix", SO_REQ_SEP },
+        { OPT_MAC_SPLIT, "--mac-spread", SO_REQ_SEP },
+        { OPT_SEND_DEBUG_PKT, "--send-debug-pkt", SO_REQ_SEP },
+
+        SO_END_OF_OPTIONS
+    };
 
 
 
@@ -601,7 +601,7 @@ static int usage(){
     printf(" Usage: t-rex-64 [MODE] [OPTION] -f cfg.yaml -c cores   \n");
     printf(" \n");
     printf(" \n");
-    
+
     printf(" mode \n\n");
     printf(" -f [file]                  : YAML file  with template configuration \n");
     printf(" -i                         : launch TRex in interactive mode (RPC server)\n");
@@ -656,8 +656,8 @@ static int usage(){
     printf(" --ipv6                     : work in ipv6 mode\n");
     printf(" --learn (deprecated). Replaced by --learn-mode. To get older behaviour, use --learn-mode 2\n");
     printf(" --learn-mode [1-2]         : Work in NAT environments, learn the dynamic NAT translation and ALG  \n");
-	printf("      1    Use TCP ACK in first SYN to pass NAT translation information. Will work only for TCP streams. Initial SYN packet must be present in stream.\n");
-	printf("      2    Add special IP option to pass NAT translation information. Will not work on certain firewalls if they drop packets with IP options\n");
+    printf("      1    Use TCP ACK in first SYN to pass NAT translation information. Will work only for TCP streams. Initial SYN packet must be present in stream.\n");
+    printf("      2    Add special IP option to pass NAT translation information. Will not work on certain firewalls if they drop packets with IP options\n");
     printf(" --learn-verify             : Learn the translation, but intended for verification of the mechanism in cases that NAT does not exist \n");
     printf("  \n");
     printf(" --l-pkt-mode [0-3]         : Set mode for sending latency packets.\n");
@@ -673,9 +673,9 @@ static int usage(){
     printf("   \n");
     printf("  Warning : This program can generate huge-files (TB ) watch out! try this only on local drive \n");
     printf(" \n");
-	printf("  \n");
-	printf(" --rx-check  [sample]       :  enable rx check thread, using this thread we sample flows 1/sample and check order,latency and more  \n");
-	printf("                              this feature consume another thread  \n");
+    printf("  \n");
+    printf(" --rx-check  [sample]       :  enable rx check thread, using this thread we sample flows 1/sample and check order,latency and more  \n");
+    printf("                              this feature consume another thread  \n");
     printf("  \n");
     printf(" --hops [hops]              :  If rx check is enabled, the hop number can be assigned. The default number of hops is 1\n");
     printf(" --iom  [mode]              :  io mode for interactive mode [0- silent, 1- normal , 2- short]   \n");
@@ -686,8 +686,8 @@ static int usage(){
     printf(" --prefix                   : for multi trex, each instance should have a different name \n");
     printf(" --mac-spread               : Spread the destination mac-order by this factor. e.g 2 will generate the traffic to 2 devices DEST-MAC ,DEST-MAC+1  \n");
     printf("                             maximum is up to 128 devices   \n");
-    
-    
+
+
     printf("\n simulation mode : \n");
     printf(" Using this mode you can generate the traffic into a pcap file and learn how trex works \n");
     printf(" With this version you must be SUDO to use this mode ( I know this is not normal )  \n");
@@ -752,21 +752,21 @@ static void parse_err(const std::string &msg) {
 }
 
 static int parse_options(int argc, char *argv[], CParserOption* po, bool first_time ) {
-     CSimpleOpt args(argc, argv, parser_options);
+    CSimpleOpt args(argc, argv, parser_options);
 
-     bool latency_was_set=false;
-     (void)latency_was_set;
+    bool latency_was_set=false;
+    (void)latency_was_set;
 
-     int a=0;
-     int node_dump=0;
+    int a=0;
+    int node_dump=0;
 
-     po->preview.setFileWrite(true);
-     po->preview.setRealTime(true);
-     uint32_t tmp_data;
+    po->preview.setFileWrite(true);
+    po->preview.setRealTime(true);
+    uint32_t tmp_data;
 
-     po->m_run_mode = CParserOption::RUN_MODE_INVALID;
+    po->m_run_mode = CParserOption::RUN_MODE_INVALID;
 
-     while ( args.Next() ){
+    while ( args.Next() ){
         if (args.LastError() == SO_SUCCESS) {
             switch (args.OptionId()) {
 
@@ -774,7 +774,7 @@ static int parse_options(int argc, char *argv[], CParserOption* po, bool first_t
                 parse_err("Supported only in simulation");
                 break;
 
-            case OPT_HELP: 
+            case OPT_HELP:
                 usage();
                 return -1;
 
@@ -823,25 +823,25 @@ static int parse_options(int argc, char *argv[], CParserOption* po, bool first_t
 
             case OPT_LEARN_MODE :
                 sscanf(args.OptionArg(),"%d", &tmp_data);
-		if (! po->is_valid_opt_val(tmp_data, CParserOption::LEARN_MODE_DISABLED, CParserOption::LEARN_MODE_MAX, "--learn-mode")) {
-		    exit(-1);
-		}
+                if (! po->is_valid_opt_val(tmp_data, CParserOption::LEARN_MODE_DISABLED, CParserOption::LEARN_MODE_MAX, "--learn-mode")) {
+                    exit(-1);
+                }
                 po->m_learn_mode = (uint8_t)tmp_data;
                 break;
 
             case OPT_LEARN_VERIFY :
-		// must configure learn_mode for learn verify to work. If different learn mode will be given later, it will be set instead.
-		if (po->m_learn_mode == 0) {
-		    po->m_learn_mode = CParserOption::LEARN_MODE_IP_OPTION;
-		}
+                // must configure learn_mode for learn verify to work. If different learn mode will be given later, it will be set instead.
+                if (po->m_learn_mode == 0) {
+                    po->m_learn_mode = CParserOption::LEARN_MODE_IP_OPTION;
+                }
                 po->preview.set_learn_and_verify_mode_enable(true);
                 break;
 
             case OPT_L_PKT_MODE :
                 sscanf(args.OptionArg(),"%d", &tmp_data);
-		if (! po->is_valid_opt_val(tmp_data, 0, L_PKT_SUBMODE_0_SEQ, "--l-pkt-mode")) {
-		    exit(-1);
-		}
+                if (! po->is_valid_opt_val(tmp_data, 0, L_PKT_SUBMODE_0_SEQ, "--l-pkt-mode")) {
+                    exit(-1);
+                }
                 po->m_l_pkt_mode=(uint8_t)tmp_data;
                 break;
 
@@ -916,7 +916,7 @@ static int parse_options(int argc, char *argv[], CParserOption* po, bool first_t
 
             case OPT_RX_CHECK :
                 sscanf(args.OptionArg(),"%d", &tmp_data);
-                po->m_rx_check_sampe=(uint16_t)tmp_data;
+                po->m_rx_check_sample=(uint16_t)tmp_data;
                 po->preview.set_rx_check_enable(true);
                 break;
             case OPT_RX_CHECK_HOPS :
@@ -954,17 +954,17 @@ static int parse_options(int argc, char *argv[], CParserOption* po, bool first_t
                 return -1;
                 break;
             } // End of switch
-         }// End of IF
+        }// End of IF
         else {
             usage();
             return -1;
         }
-     } // End of while
+    } // End of while
 
 
     if ((po->m_run_mode ==  CParserOption::RUN_MODE_INVALID) ) {
         parse_err("Please provide single run mode (e.g. batch or interactive)");
-     }
+    }
 
     if ( po->m_mac_splitter > 128 ){
         std::stringstream ss;
@@ -977,15 +977,10 @@ static int parse_options(int argc, char *argv[], CParserOption* po, bool first_t
             parse_err("--learn mode is not supported with --ipv6, beacuse there is not such thing NAT66 ( ipv6-ipv6) \n" \
                       "if you think it is important,open a defect \n");
         }
-        if ( po->is_latency_disabled() ){
-            /* set latency thread */
-            po->m_latency_rate =1000;
-        }
     }
 
-    if (po->preview.get_is_rx_check_enable() &&  ( po->is_latency_disabled() ) ) {
-        printf(" rx check must be enabled with latency check. try adding '-l 1000'   \n");
-        return -1;
+    if (po->preview.get_is_rx_check_enable() ||  po->is_latency_enabled() || CGlobalInfo::is_learn_mode()) {
+        po->set_rx_enabled();
     }
 
     if ( node_dump ){
@@ -1006,7 +1001,7 @@ static int parse_options(int argc, char *argv[], CParserOption* po, bool first_t
         /* only first time read the configuration file */
         if ( po->platform_cfg_file.length() >0  ) {
             if ( node_dump ){
-               printf("load platform configuration file from %s \n",po->platform_cfg_file.c_str());
+                printf("load platform configuration file from %s \n",po->platform_cfg_file.c_str());
             }
             global_platform_cfg_info.load_from_yaml_file(po->platform_cfg_file);
             if ( node_dump ){
@@ -1028,7 +1023,7 @@ static int parse_options(int argc, char *argv[], CParserOption* po, bool first_t
             parse_err("Rx check is not supported with interactive mode ");
         }
 
-        if  ( (! po->is_latency_disabled()) || (po->preview.getOnlyLatency()) ){
+        if  ( (po->is_latency_enabled()) || (po->preview.getOnlyLatency()) ){
             parse_err("Latecny check is not supported with interactive mode ");
         }
 
@@ -1062,7 +1057,7 @@ int main_test(int argc , char * argv[]);
 
 
 struct port_cfg_t {
-    public:
+public:
     port_cfg_t(){
         memset(&m_port_conf,0,sizeof(m_port_conf));
         memset(&m_rx_conf,0,sizeof(m_rx_conf));
@@ -1099,33 +1094,33 @@ struct port_cfg_t {
         get_ex_drv()->update_global_config_fdir(this);
     }
 
-	/* enable FDIR */
+    /* enable FDIR */
     inline void update_global_config_fdir_10g(void){
-		m_port_conf.fdir_conf.mode=RTE_FDIR_MODE_PERFECT_MAC_VLAN;
-		m_port_conf.fdir_conf.pballoc=RTE_FDIR_PBALLOC_64K;
-		m_port_conf.fdir_conf.status=RTE_FDIR_NO_REPORT_STATUS; 
-		/* Offset of flexbytes field in RX packets (in 16-bit word units). */
-		/* Note: divide by 2 to convert byte offset to word offset */
-		if (  CGlobalInfo::m_options.preview.get_ipv6_mode_enable() ){
-			m_port_conf.fdir_conf.flexbytes_offset=(14+6)/2;
-		}else{
-			m_port_conf.fdir_conf.flexbytes_offset=(14+8)/2;
-		}
-                        
-		/* Increment offset 4 bytes for the case where we add VLAN */
-		if (  CGlobalInfo::m_options.preview.get_vlan_mode_enable() ){
-	        	m_port_conf.fdir_conf.flexbytes_offset+=(4/2);
-		}
-		m_port_conf.fdir_conf.drop_queue=1;
+        m_port_conf.fdir_conf.mode=RTE_FDIR_MODE_PERFECT_MAC_VLAN;
+        m_port_conf.fdir_conf.pballoc=RTE_FDIR_PBALLOC_64K;
+        m_port_conf.fdir_conf.status=RTE_FDIR_NO_REPORT_STATUS;
+        /* Offset of flexbytes field in RX packets (in 16-bit word units). */
+        /* Note: divide by 2 to convert byte offset to word offset */
+        if (  CGlobalInfo::m_options.preview.get_ipv6_mode_enable() ){
+            m_port_conf.fdir_conf.flexbytes_offset=(14+6)/2;
+        }else{
+            m_port_conf.fdir_conf.flexbytes_offset=(14+8)/2;
+        }
+
+        /* Increment offset 4 bytes for the case where we add VLAN */
+        if (  CGlobalInfo::m_options.preview.get_vlan_mode_enable() ){
+            m_port_conf.fdir_conf.flexbytes_offset+=(4/2);
+        }
+        m_port_conf.fdir_conf.drop_queue=1;
     }
 
     inline void update_global_config_fdir_40g(void){
         m_port_conf.fdir_conf.mode=RTE_FDIR_MODE_PERFECT;
         m_port_conf.fdir_conf.pballoc=RTE_FDIR_PBALLOC_64K;
-        m_port_conf.fdir_conf.status=RTE_FDIR_NO_REPORT_STATUS; 
+        m_port_conf.fdir_conf.status=RTE_FDIR_NO_REPORT_STATUS;
         /* Offset of flexbytes field in RX packets (in 16-bit word units). */
         /* Note: divide by 2 to convert byte offset to word offset */
-        #if 0
+#if 0
         if (  CGlobalInfo::m_options.preview.get_ipv6_mode_enable() ){
             m_port_conf.fdir_conf.flexbytes_offset=(14+6)/2;
         }else{
@@ -1134,12 +1129,12 @@ struct port_cfg_t {
 
         /* Increment offset 4 bytes for the case where we add VLAN */
         if (  CGlobalInfo::m_options.preview.get_vlan_mode_enable() ){
-                m_port_conf.fdir_conf.flexbytes_offset+=(4/2);
+            m_port_conf.fdir_conf.flexbytes_offset+=(4/2);
         }
-        #endif
+#endif
 
-    // TBD Flow Director does not work with XL710 yet we need to understand why 
-    #if 0
+        // TBD Flow Director does not work with XL710 yet we need to understand why
+#if 0
         struct rte_eth_fdir_flex_conf * lp = &m_port_conf.fdir_conf.flex_conf;
 
         //lp->nb_flexmasks=1;
@@ -1147,11 +1142,11 @@ struct port_cfg_t {
         //memset(lp->flex_mask[0].mask,0xff,RTE_ETH_FDIR_MAX_FLEXLEN);
 
         lp->nb_payloads=1;
-        lp->flex_set[0].type = RTE_ETH_L3_PAYLOAD; 
+        lp->flex_set[0].type = RTE_ETH_L3_PAYLOAD;
         lp->flex_set[0].src_offset[0]=8;
 
         //m_port_conf.fdir_conf.drop_queue=1;
-    #endif
+#endif
     }
 
     struct rte_eth_conf     m_port_conf;
@@ -1161,13 +1156,13 @@ struct port_cfg_t {
 };
 
 
-/* this object is per core / per port / per queue 
-   each core will have 2 ports to send too  
+/* this object is per core / per port / per queue
+   each core will have 2 ports to send too
 
 
-       port0                                port1
+   port0                                port1
 
- 0,1,2,3,..15 out queue ( per core )       0,1,2,3,..15 out queue ( per core )
+   0,1,2,3,..15 out queue ( per core )       0,1,2,3,..15 out queue ( per core )
 
 */
 
@@ -1181,7 +1176,7 @@ typedef struct cnt_name_ {
 
 void CPhyEthIFStats::Clear(){
     ipackets = 0;
-    ibytes = 0; 
+    ibytes = 0;
     f_ipackets = 0;
     f_ibytes = 0;
     opackets = 0;
@@ -1191,114 +1186,113 @@ void CPhyEthIFStats::Clear(){
     imcasts = 0;
     rx_nombuf = 0;
     memset(m_rx_per_flow, 0, sizeof(m_rx_per_flow));
-    m_fdir_stats_first_time = true;
 }
 
 
 void CPhyEthIFStats::DumpAll(FILE *fd){
 
-    #define DP_A4(f) printf(" %-40s : %llu \n",#f, (unsigned long long)f)
-    #define DP_A(f) if (f) printf(" %-40s : %llu \n",#f, (unsigned long long)f)
-    DP_A4(opackets);  
-    DP_A4(obytes);      
-    DP_A4(ipackets);  
-    DP_A4(ibytes);      
-    DP_A(ierrors);     
-    DP_A(oerrors);     
+#define DP_A4(f) printf(" %-40s : %llu \n",#f, (unsigned long long)f)
+#define DP_A(f) if (f) printf(" %-40s : %llu \n",#f, (unsigned long long)f)
+    DP_A4(opackets);
+    DP_A4(obytes);
+    DP_A4(ipackets);
+    DP_A4(ibytes);
+    DP_A(ierrors);
+    DP_A(oerrors);
 
 }
 
 
 void CPhyEthIFStats::Dump(FILE *fd){
 
-    DP_A(opackets);  
-    DP_A(obytes);      
+    DP_A(opackets);
+    DP_A(obytes);
 
-    DP_A(f_ipackets);  
-    DP_A(f_ibytes);      
+    DP_A(f_ipackets);
+    DP_A(f_ibytes);
 
-    DP_A(ipackets);  
-    DP_A(ibytes);      
-    DP_A(ierrors);     
-    DP_A(oerrors);     
-    DP_A(imcasts);     
-    DP_A(rx_nombuf);   
+    DP_A(ipackets);
+    DP_A(ibytes);
+    DP_A(ierrors);
+    DP_A(oerrors);
+    DP_A(imcasts);
+    DP_A(rx_nombuf);
 }
 
 void CPhyEthIF::flush_rx_queue(void){
 
-        rte_mbuf_t * rx_pkts[32];
-        int j=0;
-        uint16_t cnt=0;
-
-        while (true) {
-            j++;
-            cnt = rx_burst(m_rx_queue,rx_pkts,32);
-            if ( cnt ) {
-                int i;
-                for (i=0; i<(int)cnt;i++) {
-                    rte_mbuf_t * m=rx_pkts[i];
-                    /*printf("rx--\n");
-                    rte_pktmbuf_dump(stdout,m, rte_pktmbuf_pkt_len(m));*/
-                    rte_pktmbuf_free(m);
-                }
-            }
-            if ( ((cnt==0) && (j>10)) || (j>15) ) {
-                break;
+    rte_mbuf_t * rx_pkts[32];
+    int j=0;
+    uint16_t cnt=0;
+
+    while (true) {
+        j++;
+        cnt = rx_burst(m_rx_queue,rx_pkts,32);
+        if ( cnt ) {
+            int i;
+            for (i=0; i<(int)cnt;i++) {
+                rte_mbuf_t * m=rx_pkts[i];
+                /*printf("rx--\n");
+                  rte_pktmbuf_dump(stdout,m, rte_pktmbuf_pkt_len(m));*/
+                rte_pktmbuf_free(m);
             }
         }
-        if (cnt>0) {
-            printf(" Warning can't flush rx-queue for port %d \n",(int)get_port_id());
+        if ( ((cnt==0) && (j>10)) || (j>15) ) {
+            break;
         }
+    }
+    if (cnt>0) {
+        printf(" Warning can't flush rx-queue for port %d \n",(int)get_port_id());
+    }
 }
 
 
 void CPhyEthIF::dump_stats_extended(FILE *fd){
 
     cnt_name_t reg[]={
-            MY_REG(IXGBE_GPTC), /* total packet */
-            MY_REG(IXGBE_GOTCL), /* total bytes */
-            MY_REG(IXGBE_GOTCH), 
-
-            MY_REG(IXGBE_GPRC),
-            MY_REG(IXGBE_GORCL),
-            MY_REG(IXGBE_GORCH),
-
-
-
-            MY_REG(IXGBE_RXNFGPC),
-            MY_REG(IXGBE_RXNFGBCL),
-            MY_REG(IXGBE_RXNFGBCH),     
-            MY_REG(IXGBE_RXDGPC  ),     
-            MY_REG(IXGBE_RXDGBCL ),     
-            MY_REG(IXGBE_RXDGBCH  ),    
-             MY_REG(IXGBE_RXDDGPC ),    
-             MY_REG(IXGBE_RXDDGBCL ),   
-             MY_REG(IXGBE_RXDDGBCH  ),  
-             MY_REG(IXGBE_RXLPBKGPC ),  
-             MY_REG(IXGBE_RXLPBKGBCL),  
-             MY_REG(IXGBE_RXLPBKGBCH ), 
-             MY_REG(IXGBE_RXDLPBKGPC ), 
-             MY_REG(IXGBE_RXDLPBKGBCL), 
-             MY_REG(IXGBE_RXDLPBKGBCH ),
-             MY_REG(IXGBE_TXDGPC      ),
-             MY_REG(IXGBE_TXDGBCL     ),
-             MY_REG(IXGBE_TXDGBCH     ),
-		MY_REG(IXGBE_FDIRUSTAT ),
-	 MY_REG(IXGBE_FDIRFSTAT ),
-	 MY_REG(IXGBE_FDIRMATCH ),
-	 MY_REG(IXGBE_FDIRMISS )
+        MY_REG(IXGBE_GPTC), /* total packet */
+        MY_REG(IXGBE_GOTCL), /* total bytes */
+        MY_REG(IXGBE_GOTCH),
+
+        MY_REG(IXGBE_GPRC),
+        MY_REG(IXGBE_GORCL),
+        MY_REG(IXGBE_GORCH),
+
+
+
+        MY_REG(IXGBE_RXNFGPC),
+        MY_REG(IXGBE_RXNFGBCL),
+        MY_REG(IXGBE_RXNFGBCH),
+        MY_REG(IXGBE_RXDGPC  ),
+        MY_REG(IXGBE_RXDGBCL ),
+        MY_REG(IXGBE_RXDGBCH  ),
+        MY_REG(IXGBE_RXDDGPC ),
+        MY_REG(IXGBE_RXDDGBCL ),
+        MY_REG(IXGBE_RXDDGBCH  ),
+        MY_REG(IXGBE_RXLPBKGPC ),
+        MY_REG(IXGBE_RXLPBKGBCL),
+        MY_REG(IXGBE_RXLPBKGBCH ),
+        MY_REG(IXGBE_RXDLPBKGPC ),
+        MY_REG(IXGBE_RXDLPBKGBCL),
+        MY_REG(IXGBE_RXDLPBKGBCH ),
+        MY_REG(IXGBE_TXDGPC      ),
+        MY_REG(IXGBE_TXDGBCL     ),
+        MY_REG(IXGBE_TXDGBCH     ),
+        MY_REG(IXGBE_FDIRUSTAT ),
+        MY_REG(IXGBE_FDIRFSTAT ),
+        MY_REG(IXGBE_FDIRMATCH ),
+        MY_REG(IXGBE_FDIRMISS )
 
     };
     fprintf (fd," extended counters \n");
     int i;
     for (i=0; i<sizeof(reg)/sizeof(reg[0]); i++) {
         cnt_name_t *lp=&reg[i];
-		uint32_t c=pci_reg_read(lp->offset);
+        uint32_t c=pci_reg_read(lp->offset);
         // xl710 bug. Counter values are -559038737 when they should be 0
-		if (c && c != -559038737 ) {
-			fprintf (fd," %s  : %d \n",lp->name,c);
-		}
+        if (c && c != -559038737 ) {
+            fprintf (fd," %s  : %d \n",lp->name,c);
+        }
     }
 }
 
@@ -1310,15 +1304,15 @@ void CPhyEthIF::configure(uint16_t nb_rx_queue,
                           uint16_t nb_tx_queue,
                           const struct rte_eth_conf *eth_conf){
     int ret;
-    ret = rte_eth_dev_configure(m_port_id, 
+    ret = rte_eth_dev_configure(m_port_id,
                                 nb_rx_queue,
-                                nb_tx_queue, 
+                                nb_tx_queue,
                                 eth_conf);
 
     if (ret < 0)
         rte_exit(EXIT_FAILURE, "Cannot configure device: "
-                "err=%d, port=%u\n",
-              ret, m_port_id);
+                 "err=%d, port=%u\n",
+                 ret, m_port_id);
 
     /* get device info */
     rte_eth_dev_info_get(m_port_id, &m_dev_info);
@@ -1328,13 +1322,13 @@ void CPhyEthIF::configure(uint16_t nb_rx_queue,
 
 /*
 
-rx-queue 0 - default- all traffic not goint to queue 1
-             will be drop as queue is disable 
-             
+  rx-queue 0 - default- all traffic not goint to queue 1
+  will be drop as queue is disable
+
 
-rx-queue 1 - Latency measurement packets will go here  
+  rx-queue 1 - Latency measurement packets will go here
 
-            pci_reg_write(IXGBE_L34T_IMIR(0),(1<<21));
+  pci_reg_write(IXGBE_L34T_IMIR(0),(1<<21));
 
 */
 
@@ -1356,7 +1350,7 @@ void CPhyEthIF::configure_rx_drop_queue(){
     if ( get_vm_one_queue_enable() || (CGlobalInfo::m_options.m_debug_pkt_proto != 0)) {
         return;
     }
-    if ( CGlobalInfo::m_options.is_latency_disabled()==false ) {
+    if ( CGlobalInfo::m_options.is_rx_enabled() ) {
         if ( (!get_ex_drv()->is_hardware_support_drop_queue())  ) {
             printf(" ERROR latency feature is not supported with current hardware  \n");
             exit(1);
@@ -1367,38 +1361,38 @@ void CPhyEthIF::configure_rx_drop_queue(){
 
 
 void CPhyEthIF::rx_queue_setup(uint16_t rx_queue_id,
-                               uint16_t nb_rx_desc, 
+                               uint16_t nb_rx_desc,
                                unsigned int socket_id,
                                const struct rte_eth_rxconf *rx_conf,
                                struct rte_mempool *mb_pool){
 
-    int ret = rte_eth_rx_queue_setup(m_port_id , rx_queue_id, 
+    int ret = rte_eth_rx_queue_setup(m_port_id , rx_queue_id,
                                      nb_rx_desc,
-                                     socket_id, 
+                                     socket_id,
                                      rx_conf,
                                      mb_pool);
     if (ret < 0)
         rte_exit(EXIT_FAILURE, "rte_eth_rx_queue_setup: "
-                "err=%d, port=%u\n",
-              ret, m_port_id);
+                 "err=%d, port=%u\n",
+                 ret, m_port_id);
 }
 
 
 
 void CPhyEthIF::tx_queue_setup(uint16_t tx_queue_id,
-                               uint16_t nb_tx_desc, 
+                               uint16_t nb_tx_desc,
                                unsigned int socket_id,
                                const struct rte_eth_txconf *tx_conf){
 
     int ret = rte_eth_tx_queue_setup( m_port_id,
-                                     tx_queue_id, 
+                                      tx_queue_id,
                                       nb_tx_desc,
-                                      socket_id, 
+                                      socket_id,
                                       tx_conf);
     if (ret < 0)
         rte_exit(EXIT_FAILURE, "rte_eth_tx_queue_setup: "
-                "err=%d, port=%u queue=%u\n",
-              ret, m_port_id, tx_queue_id);
+                 "err=%d, port=%u queue=%u\n",
+                 ret, m_port_id, tx_queue_id);
 
 }
 
@@ -1418,7 +1412,7 @@ void CPhyEthIF::start(){
     m_bw_rx.reset();
 
     m_stats.Clear();
-    int i; 
+    int i;
     for (i=0;i<10; i++ ) {
         ret = rte_eth_dev_start(m_port_id);
         if (ret==0) {
@@ -1428,32 +1422,32 @@ void CPhyEthIF::start(){
     }
     if (ret < 0)
         rte_exit(EXIT_FAILURE, "rte_eth_dev_start: "
-                "err=%d, port=%u\n",
-              ret, m_port_id);
+                 "err=%d, port=%u\n",
+                 ret, m_port_id);
 
 }
 
 // Disabling flow control on interface
 void CPhyEthIF::disable_flow_control(){
-       int ret;
-        // see trex-64 issue with loopback on the same NIC
-        struct rte_eth_fc_conf fc_conf;
-        memset(&fc_conf,0,sizeof(fc_conf));
-        fc_conf.mode=RTE_FC_NONE;
-        fc_conf.autoneg=1;
-        fc_conf.pause_time=100;
-        int i;
-        for (i=0; i<5; i++) {
-            ret=rte_eth_dev_flow_ctrl_set(m_port_id,&fc_conf);
-            if (ret==0) {
-                break;
-            }
-            delay(1000);
+    int ret;
+    // see trex-64 issue with loopback on the same NIC
+    struct rte_eth_fc_conf fc_conf;
+    memset(&fc_conf,0,sizeof(fc_conf));
+    fc_conf.mode=RTE_FC_NONE;
+    fc_conf.autoneg=1;
+    fc_conf.pause_time=100;
+    int i;
+    for (i=0; i<5; i++) {
+        ret=rte_eth_dev_flow_ctrl_set(m_port_id,&fc_conf);
+        if (ret==0) {
+            break;
         }
-        if (ret < 0)
-          rte_exit(EXIT_FAILURE, "rte_eth_dev_flow_ctrl_set: "
-                  "err=%d, port=%u\n probably link is down. Please check your link activity, or skip flow-control disabling, using: --no-flow-control-change option\n",
-                ret, m_port_id);
+        delay(1000);
+    }
+    if (ret < 0)
+        rte_exit(EXIT_FAILURE, "rte_eth_dev_flow_ctrl_set: "
+                 "err=%d, port=%u\n probably link is down. Please check your link activity, or skip flow-control disabling, using: --no-flow-control-change option\n",
+                 ret, m_port_id);
 }
 
 
@@ -1465,9 +1459,9 @@ void CPhyEthIF::dump_link(FILE *fd){
     fprintf(fd,"link         : ");
     if (m_link.link_status) {
         fprintf(fd," link : Link Up - speed %u Mbps - %s\n",
-               (unsigned) m_link.link_speed,
-               (m_link.link_duplex == ETH_LINK_FULL_DUPLEX) ?
-               ("full-duplex") : ("half-duplex\n"));
+                (unsigned) m_link.link_speed,
+                (m_link.link_duplex == ETH_LINK_FULL_DUPLEX) ?
+                ("full-duplex") : ("half-duplex\n"));
     } else {
         fprintf(fd," Link Down\n");
     }
@@ -1491,7 +1485,7 @@ void CPhyEthIF::set_promiscuous(bool enable){
     if (enable) {
         rte_eth_promiscuous_enable(m_port_id);
     }else{
-       rte_eth_promiscuous_disable(m_port_id);
+        rte_eth_promiscuous_disable(m_port_id);
     }
 }
 
@@ -1499,7 +1493,7 @@ bool CPhyEthIF::get_promiscuous(){
     int ret=rte_eth_promiscuous_get(m_port_id);
     if (ret<0) {
         rte_exit(EXIT_FAILURE, "rte_eth_promiscuous_get: "
-                "err=%d, port=%u\n",
+                 "err=%d, port=%u\n",
                  ret, m_port_id);
 
     }
@@ -1508,51 +1502,51 @@ bool CPhyEthIF::get_promiscuous(){
 
 
 void CPhyEthIF::macaddr_get(struct ether_addr *mac_addr){
-       rte_eth_macaddr_get(m_port_id , mac_addr);
+    rte_eth_macaddr_get(m_port_id , mac_addr);
 }
 
 
-void CPhyEthIF::get_stats_1g(CPhyEthIFStats *stats){ 
+void CPhyEthIF::get_stats_1g(CPhyEthIFStats *stats){
 
-   stats->ipackets     +=  pci_reg_read(E1000_GPRC) ;
+    stats->ipackets     +=  pci_reg_read(E1000_GPRC) ;
 
-   stats->ibytes       +=  (pci_reg_read(E1000_GORCL) );
-   stats->ibytes       +=  (((uint64_t)pci_reg_read(E1000_GORCH))<<32);
-                        
+    stats->ibytes       +=  (pci_reg_read(E1000_GORCL) );
+    stats->ibytes       +=  (((uint64_t)pci_reg_read(E1000_GORCH))<<32);
 
-   stats->opackets     +=  pci_reg_read(E1000_GPTC);
-   stats->obytes       +=  pci_reg_read(E1000_GOTCL) ;
-   stats->obytes       +=  ( (((uint64_t)pci_reg_read(IXGBE_GOTCH))<<32) );
 
-   stats->f_ipackets   +=  0;
-   stats->f_ibytes     += 0;
+    stats->opackets     +=  pci_reg_read(E1000_GPTC);
+    stats->obytes       +=  pci_reg_read(E1000_GOTCL) ;
+    stats->obytes       +=  ( (((uint64_t)pci_reg_read(IXGBE_GOTCH))<<32) );
 
+    stats->f_ipackets   +=  0;
+    stats->f_ibytes     += 0;
 
-   stats->ierrors      +=  ( pci_reg_read(E1000_RNBC) +
-                             pci_reg_read(E1000_CRCERRS) + 
-                             pci_reg_read(E1000_ALGNERRC ) +
-                             pci_reg_read(E1000_SYMERRS ) +
-                             pci_reg_read(E1000_RXERRC ) +
 
-                             pci_reg_read(E1000_ROC)+
-                             pci_reg_read(E1000_RUC)+ 
-                             pci_reg_read(E1000_RJC) +
+    stats->ierrors      +=  ( pci_reg_read(E1000_RNBC) +
+                              pci_reg_read(E1000_CRCERRS) +
+                              pci_reg_read(E1000_ALGNERRC ) +
+                              pci_reg_read(E1000_SYMERRS ) +
+                              pci_reg_read(E1000_RXERRC ) +
 
-                             pci_reg_read(E1000_XONRXC)+   
-                            pci_reg_read(E1000_XONTXC)+
-                            pci_reg_read(E1000_XOFFRXC)+
-                            pci_reg_read(E1000_XOFFTXC)+
-                            pci_reg_read(E1000_FCRUC)
-                             );
+                              pci_reg_read(E1000_ROC)+
+                              pci_reg_read(E1000_RUC)+
+                              pci_reg_read(E1000_RJC) +
 
-   stats->oerrors      +=  0;
-   stats->imcasts      =  0;
-   stats->rx_nombuf    =  0;
+                              pci_reg_read(E1000_XONRXC)+
+                              pci_reg_read(E1000_XONTXC)+
+                              pci_reg_read(E1000_XOFFRXC)+
+                              pci_reg_read(E1000_XOFFTXC)+
+                              pci_reg_read(E1000_FCRUC)
+                              );
 
-   m_last_tx_rate      =  m_bw_tx.add(stats->obytes);
-   m_last_rx_rate      =  m_bw_rx.add(stats->ibytes);
-   m_last_tx_pps       =  m_pps_tx.add(stats->opackets);
-   m_last_rx_pps       =  m_pps_rx.add(stats->ipackets);
+    stats->oerrors      +=  0;
+    stats->imcasts      =  0;
+    stats->rx_nombuf    =  0;
+
+    m_last_tx_rate      =  m_bw_tx.add(stats->obytes);
+    m_last_rx_rate      =  m_bw_rx.add(stats->ibytes);
+    m_last_tx_pps       =  m_pps_tx.add(stats->opackets);
+    m_last_rx_pps       =  m_pps_rx.add(stats->ipackets);
 
 }
 
@@ -1560,61 +1554,15 @@ int CPhyEthIF::dump_fdir_global_stats(FILE *fd) {
     return get_ex_drv()->dump_fdir_global_stats(this, fd);
 }
 
-// get/reset flow director counters
-// return 0 if OK. -1 if operation not supported.
-// stats - If not NULL, returning counter numbers in it.
-// index - If non negative, get only counter with this index
-// reset - If true, reset counter value after reading
-int CPhyEthIF::get_rx_stats(uint64_t *stats, int index, bool reset) {
-    uint32_t diff_stats[TREX_FDIR_STAT_SIZE];
-    int start, len;
-
-    if (index >= 0) {
-        start = index;
-        len = 1;
-    } else {
-        start = 0;
-        len = TREX_FDIR_STAT_SIZE;
-    }
-    
-    if (get_ex_drv()->get_rx_stats(this, diff_stats, m_stats.m_fdir_prev_stats, index) < 0) {
-        return -1;
-    }
-
-    // First time, just syncing the counters
-    if (m_stats.m_fdir_stats_first_time) {
-        m_stats.m_fdir_stats_first_time = false;
-        if (stats) {
-            memset(stats, 0, sizeof(uint64_t) * TREX_FDIR_STAT_SIZE);
-        }
-        return 0;
-    }
-
-    for (int i = start; i < (start + len); i++) {
-        if ( reset ) {
-            // return value so far, and reset
-            stats[i] = m_stats.m_rx_per_flow[i] + diff_stats[i];
-            m_stats.m_rx_per_flow[i] = 0;
-        } else {
-            m_stats.m_rx_per_flow[i] += diff_stats[i];
-            if (stats != NULL) {
-                stats[i] = m_stats.m_rx_per_flow[i];
-            }
-        }
-    }
-    
-    return 0;
-}
-
 void dump_hw_state(FILE *fd,struct ixgbe_hw_stats *hs ){
 
-    #define DP_A1(f) if (hs->f) fprintf(fd," %-40s : %llu \n",#f, (unsigned long long)hs->f)
-    #define DP_A2(f,m) for (i=0;i<m; i++) { if (hs->f[i]) fprintf(fd," %-40s[%d] : %llu \n",#f,i, (unsigned long long)hs->f[i]); }
-       int i;
+#define DP_A1(f) if (hs->f) fprintf(fd," %-40s : %llu \n",#f, (unsigned long long)hs->f)
+#define DP_A2(f,m) for (i=0;i<m; i++) { if (hs->f[i]) fprintf(fd," %-40s[%d] : %llu \n",#f,i, (unsigned long long)hs->f[i]); }
+    int i;
 
     //for (i=0;i<8; i++) { if (hs->mpc[i]) fprintf(fd," %-40s[%d] : %llu \n","mpc",i,hs->mpc[i]); }
-     DP_A2(mpc,8);
-     DP_A1(crcerrs);
+    DP_A2(mpc,8);
+    DP_A1(crcerrs);
     DP_A1(illerrc);
     //DP_A1(errbc);
     DP_A1(mspdc);
@@ -1634,7 +1582,7 @@ void dump_hw_state(FILE *fd,struct ixgbe_hw_stats *hs ){
     //DP_A1(prc64);
     //DP_A1(prc127);
     //DP_A1(prc255);
-   // DP_A1(prc511);
+    // DP_A1(prc511);
     //DP_A1(prc1023);
     //DP_A1(prc1522);
 
@@ -1665,7 +1613,7 @@ void dump_hw_state(FILE *fd,struct ixgbe_hw_stats *hs ){
     DP_A1(bptc);
     DP_A1(xec);
     DP_A2(qprc,16)
-    DP_A2(qptc,16);
+        DP_A2(qptc,16);
     DP_A2(qbrc,16);
     DP_A2(qbtc,16);
     DP_A2(qprdc,16);
@@ -1694,19 +1642,19 @@ void dump_hw_state(FILE *fd,struct ixgbe_hw_stats *hs ){
 }
 
 
-void CPhyEthIF::update_counters(){ 
-   get_ex_drv()->get_extended_stats(this, &m_stats);
+void CPhyEthIF::update_counters(){
+    get_ex_drv()->get_extended_stats(this, &m_stats);
 
-   m_last_tx_rate      =  m_bw_tx.add(m_stats.obytes);
-   m_last_rx_rate      =  m_bw_rx.add(m_stats.ibytes);
-   m_last_tx_pps       =  m_pps_tx.add(m_stats.opackets);
-   m_last_rx_pps       =  m_pps_rx.add(m_stats.ipackets);
+    m_last_tx_rate      =  m_bw_tx.add(m_stats.obytes);
+    m_last_rx_rate      =  m_bw_rx.add(m_stats.ibytes);
+    m_last_tx_pps       =  m_pps_tx.add(m_stats.opackets);
+    m_last_rx_pps       =  m_pps_rx.add(m_stats.ipackets);
 }
 
-void CPhyEthIF::dump_stats(FILE *fd){ 
+void CPhyEthIF::dump_stats(FILE *fd){
 
     update_counters();
-    
+
     fprintf(fd,"port : %d \n",(int)m_port_id);
     fprintf(fd,"------------\n");
     m_stats.DumpAll(fd);
@@ -1734,7 +1682,7 @@ public:
     uint16_t                m_tx_queue_id;
     uint16_t                m_len;
     rte_mbuf_t *            m_table[MAX_PKT_BURST];
-    CPhyEthIF  *            m_port;  
+    CPhyEthIF  *            m_port;
 };
 
 
@@ -1753,7 +1701,7 @@ public:
 public:
     bool Create(uint8_t             core_id,
                 uint16_t            tx_client_queue_id,
-                CPhyEthIF  *        tx_client_port, 
+                CPhyEthIF  *        tx_client_port,
 
                 uint16_t            tx_server_queue_id,
                 CPhyEthIF  *        tx_server_port);
@@ -1809,7 +1757,7 @@ protected:
 
 protected:
     uint8_t      m_core_id;
-    uint16_t     m_mbuf_cache; 
+    uint16_t     m_mbuf_cache;
     CCorePerPort m_ports[CS_NUM]; /* each core has 2 tx queues 1. client side and server side */
     CNodeRing *  m_ring_to_rx;
 
@@ -1822,7 +1770,7 @@ public:
 
 bool CCoreEthIF::Create(uint8_t             core_id,
                         uint16_t            tx_client_queue_id,
-                        CPhyEthIF  *        tx_client_port, 
+                        CPhyEthIF  *        tx_client_port,
 
                         uint16_t            tx_server_queue_id,
                         CPhyEthIF  *        tx_server_port){
@@ -1841,7 +1789,7 @@ bool CCoreEthIF::Create(uint8_t             core_id,
 
 void CCoreEthIF::flush_rx_queue(void){
     pkt_dir_t   dir ;
-    bool is_latency=get_is_latency_thread_enable();
+    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;
@@ -1856,7 +1804,7 @@ void CCoreEthIF::flush_rx_queue(void){
                 int i;
                 for (i=0; i<(int)cnt;i++) {
                     rte_mbuf_t * m=rx_pkts[i];
-                    if ( is_latency ){
+                    if ( is_rx ){
                         if (!process_rx_pkt(dir,m)){
                             rte_pktmbuf_free(m);
                         }
@@ -1880,7 +1828,7 @@ int CCoreEthIF::flush_tx_queue(void){
         CVirtualIFPerSideStats  * lp_stats= &m_stats[dir];
         if ( likely(lp_port->m_len > 0) ) {
             send_burst(lp_port,lp_port->m_len,lp_stats);
-             lp_port->m_len = 0;
+            lp_port->m_len = 0;
         }
     }
 
@@ -1948,18 +1896,18 @@ int CCoreEthIF::send_burst(CCorePerPort * lp_port,
                            CVirtualIFPerSideStats  * lp_stats){
 
     uint16_t ret = lp_port->m_port->tx_burst(lp_port->m_tx_queue_id,lp_port->m_table,len);
-    #ifdef DELAY_IF_NEEDED
+#ifdef DELAY_IF_NEEDED
     while ( unlikely( ret<len ) ){
         rte_delay_us(1);
         //rte_pause();
         //rte_pause();
         lp_stats->m_tx_queue_full += 1;
         uint16_t ret1=lp_port->m_port->tx_burst(lp_port->m_tx_queue_id,
-                                        &lp_port->m_table[ret],
-                                        len-ret);
+                                                &lp_port->m_table[ret],
+                                                len-ret);
         ret+=ret1;
     }
-    #endif
+#endif
 
     /* CPU has burst of packets , more that TX can send need to drop them !!*/
     if ( unlikely(ret < len) ) {
@@ -1973,7 +1921,7 @@ int CCoreEthIF::send_burst(CCorePerPort * lp_port,
 
     return (0);
 }
-                         
+
 
 int CCoreEthIF::send_pkt(CCorePerPort * lp_port,
                          rte_mbuf_t      *m,
@@ -2002,7 +1950,7 @@ int CCoreEthIF::send_pkt(CCorePerPort * lp_port,
 
 
 
-void CCoreEthIF::send_one_pkt(pkt_dir_t       dir, 
+void CCoreEthIF::send_one_pkt(pkt_dir_t       dir,
                               rte_mbuf_t      *m){
     CCorePerPort *  lp_port=&m_ports[dir];
     CVirtualIFPerSideStats  * lp_stats = &m_stats[dir];
@@ -2037,7 +1985,6 @@ void CCoreEthIF::update_mac_addr(CGenNode * node,uint8_t *p){
 
 int CCoreEthIFStateless::send_node(CGenNode * no){
     CGenNodeStateless * node_sl=(CGenNodeStateless *) no;
-
     /* check that we have mbuf  */
     rte_mbuf_t *    m=node_sl->get_cache_mbuf();
     pkt_dir_t dir=(pkt_dir_t)node_sl->get_mbuf_cache_dir();
@@ -2051,6 +1998,15 @@ int CCoreEthIFStateless::send_node(CGenNode * no){
         assert(m);
     }
 
+    if (unlikely(node_sl->is_stat_needed())) {
+        uint16_t hw_id = node_sl->get_stat_hw_id();
+        if (hw_id > max_stat_hw_id_seen) {
+            max_stat_hw_id_seen = hw_id;
+        }
+        tx_per_flow_t *lp_s = &lp_stats->m_tx_per_flow[hw_id];
+        lp_s->add_pkts(1);
+        lp_s->add_bytes(m->pkt_len);
+    }
     send_pkt(lp_port,m,lp_stats);
 
     return (0);
@@ -2070,7 +2026,7 @@ int CCoreEthIF::send_node(CGenNode * node){
         send_pkt(lp_port,m,lp_stats);
         return (0);
     }
-    
+
 
     CFlowPktInfo *  lp=node->m_pkt_info;
     rte_mbuf_t *    m=lp->generate_new_mbuf(node);
@@ -2087,18 +2043,18 @@ int CCoreEthIF::send_node(CGenNode * node){
 
         /* set the vlan */
         m->ol_flags = PKT_TX_VLAN_PKT;
-		m->l2_len   =14;
-		uint16_t vlan_id = CGlobalInfo::m_options.m_vlan_port[vlan_port];
+        m->l2_len   =14;
+        uint16_t vlan_id = CGlobalInfo::m_options.m_vlan_port[vlan_port];
 
 
-		if (likely( vlan_id >0 ) ) {
-			m->vlan_tci = vlan_id;
-			dir = dir ^ vlan_port;
-		}else{
-			/* both from the same dir but with VLAN0 */
-			m->vlan_tci = CGlobalInfo::m_options.m_vlan_port[0];
-			dir = dir ^ 0;
-		}
+        if (likely( vlan_id >0 ) ) {
+            m->vlan_tci = vlan_id;
+            dir = dir ^ vlan_port;
+        }else{
+            /* both from the same dir but with VLAN0 */
+            m->vlan_tci = CGlobalInfo::m_options.m_vlan_port[0];
+            dir = dir ^ 0;
+        }
     }
 
     CCorePerPort *  lp_port=&m_ports[dir];
@@ -2108,7 +2064,7 @@ int CCoreEthIF::send_node(CGenNode * node){
         lp_stats->m_tx_alloc_error++;
         return(0);
     }
-    
+
     /* update mac addr dest/src 12 bytes */
     uint8_t *p=rte_pktmbuf_mtod(m, uint8_t*);
     uint8_t p_id=lp_port->m_port->get_port_id();
@@ -2116,19 +2072,19 @@ int CCoreEthIF::send_node(CGenNode * node){
 
     memcpy(p,CGlobalInfo::m_options.get_dst_src_mac_addr(p_id),12);
 
-    /* if customer enables both mac_file and get_mac_ip_overide, 
+    /* if customer enables both mac_file and get_mac_ip_overide,
      * we will apply mac_file.
      */
     if ( unlikely(CGlobalInfo::m_options.preview.get_mac_ip_features_enable() ) ) {
         update_mac_addr(node,p);
     }
 
-	if ( unlikely( node->is_rx_check_enabled() ) ) {
+    if ( unlikely( node->is_rx_check_enabled() ) ) {
         lp_stats->m_tx_rx_check_pkt++;
         lp->do_generate_new_mbuf_rxcheck(m, node, single_port);
         lp_stats->m_template.inc_template( node->get_template_id( ));
-	}else{
-        // cache only if it is not sample as this is more complex mbuf struct 
+    }else{
+        // cache only if it is not sample as this is more complex mbuf struct
         if ( unlikely( node->can_cache_mbuf() ) ) {
             if ( !CGlobalInfo::m_options.preview.isMbufCacheDisabled() ){
                 m_mbuf_cache++;
@@ -2143,7 +2099,7 @@ int CCoreEthIF::send_node(CGenNode * node){
     }
 
     /*printf("send packet -- \n");
-    rte_pktmbuf_dump(stdout,m, rte_pktmbuf_pkt_len(m));*/
+      rte_pktmbuf_dump(stdout,m, rte_pktmbuf_pkt_len(m));*/
 
     /* send the packet */
     send_pkt(lp_port,m,lp_stats);
@@ -2161,7 +2117,7 @@ int CCoreEthIF::update_mac_addr_from_global_cfg(pkt_dir_t  dir, uint8_t * p){
     return (0);
 }
 
-pkt_dir_t 
+pkt_dir_t
 CCoreEthIF::port_id_to_dir(uint8_t port_id) {
 
     for (pkt_dir_t dir = 0; dir < CS_NUM; dir++) {
@@ -2187,11 +2143,11 @@ public:
         rte_mbuf_t * tx_pkts[2];
         tx_pkts[0]=m;
         if ( likely( CGlobalInfo::m_options.preview.get_vlan_mode_enable() ) ){
-             /* vlan mode is the default */
-             /* set the vlan */
-             m->ol_flags = PKT_TX_VLAN_PKT;
-             m->vlan_tci =CGlobalInfo::m_options.m_vlan_port[0];
-			 m->l2_len   =14;
+            /* vlan mode is the default */
+            /* set the vlan */
+            m->ol_flags = PKT_TX_VLAN_PKT;
+            m->vlan_tci =CGlobalInfo::m_options.m_vlan_port[0];
+            m->l2_len   =14;
         }
         uint16_t res=m_port->tx_burst(m_tx_queue_id,tx_pkts,1);
         if ( res == 0 ) {
@@ -2199,13 +2155,13 @@ public:
             //printf(" queue is full for latency packet !!\n");
             return (-1);
 
-        }                        
-        #if 0
+        }
+#if 0
         fprintf(stdout," ==> %f.03 send packet ..\n",now_sec());
         uint8_t *p1=rte_pktmbuf_mtod(m, uint8_t*);
         uint16_t pkt_size1=rte_pktmbuf_pkt_len(m);
         utl_DumpBuffer(stdout,p1,pkt_size1,0);
-        #endif
+#endif
 
         return (0);
     }
@@ -2219,8 +2175,8 @@ public:
         }
     }
 
-    virtual uint16_t rx_burst(struct rte_mbuf **rx_pkts, 
-                               uint16_t nb_pkts){
+    virtual uint16_t rx_burst(struct rte_mbuf **rx_pkts,
+                              uint16_t nb_pkts){
         uint16_t cnt=m_port->rx_burst(m_rx_queue_id,rx_pkts,nb_pkts);
         return (cnt);
     }
@@ -2236,7 +2192,7 @@ private:
 class CLatencyVmPort : public CPortLatencyHWBase {
 public:
     void Create(uint8_t port_index,CNodeRing * ring,
-                        CLatencyManager * mgr){
+                CLatencyManager * mgr){
         m_dir        = (port_index%2);
         m_ring_to_dp = ring;
         m_mgr        = mgr;
@@ -2244,11 +2200,11 @@ public:
 
     virtual int tx(rte_mbuf_t * m){
         if ( likely( CGlobalInfo::m_options.preview.get_vlan_mode_enable() ) ){
-             /* vlan mode is the default */
-             /* set the vlan */
-             m->ol_flags = PKT_TX_VLAN_PKT;
-             m->vlan_tci =CGlobalInfo::m_options.m_vlan_port[0];
-			 m->l2_len   =14;
+            /* vlan mode is the default */
+            /* set the vlan */
+            m->ol_flags = PKT_TX_VLAN_PKT;
+            m->vlan_tci =CGlobalInfo::m_options.m_vlan_port[0];
+            m->l2_len   =14;
         }
 
         /* allocate node */
@@ -2267,11 +2223,11 @@ public:
     }
 
     virtual rte_mbuf_t * rx(){
-            return (0);
+        return (0);
     }
 
-    virtual uint16_t rx_burst(struct rte_mbuf **rx_pkts, 
-                               uint16_t nb_pkts){
+    virtual uint16_t rx_burst(struct rte_mbuf **rx_pkts,
+                              uint16_t nb_pkts){
         return (0);
     }
 
@@ -2287,11 +2243,13 @@ private:
 class CPerPortStats {
 public:
     uint64_t opackets;
-    uint64_t obytes;      
-    uint64_t ipackets;  
+    uint64_t obytes;
+    uint64_t ipackets;
     uint64_t ibytes;
     uint64_t ierrors;
-    uint64_t oerrors;     
+    uint64_t oerrors;
+    tx_per_flow_t m_tx_per_flow[MAX_FLOW_STATS];
+    tx_per_flow_t m_prev_tx_per_flow[MAX_FLOW_STATS];
 
     float     m_total_tx_bps;
     float     m_total_tx_pps;
@@ -2304,7 +2262,7 @@ class CGlobalStats {
 public:
     enum DumpFormat {
         dmpSTANDARD,
-        dmpTABLE 
+        dmpTABLE
     };
 
     uint64_t  m_total_tx_pkts;
@@ -2346,7 +2304,7 @@ public:
     uint8_t m_threads;
 
     uint32_t      m_num_of_ports;
-    CPerPortStats m_port[BP_MAX_PORTS];
+    CPerPortStats m_port[TREX_MAX_PORTS];
 public:
     void Dump(FILE *fd,DumpFormat mode);
     void DumpAllPorts(FILE *fd);
@@ -2387,8 +2345,8 @@ std::string CGlobalStats::get_field_port(int port,std::string name,uint64_t &f){
 void CGlobalStats::dump_json(std::string & json){
     json="{\"name\":\"trex-global\",\"type\":0,\"data\":{";
 
-    #define GET_FIELD(f) get_field(std::string(#f),f)
-    #define GET_FIELD_PORT(p,f) get_field_port(p,std::string(#f),lp->f)
+#define GET_FIELD(f) get_field(std::string(#f),f)
+#define GET_FIELD_PORT(p,f) get_field_port(p,std::string(#f),lp->f)
 
     json+=GET_FIELD(m_cpu_util);
     json+=GET_FIELD(m_platform_factor);
@@ -2450,7 +2408,7 @@ void CGlobalStats::DumpAllPorts(FILE *fd){
     //fprintf (fd," Total-Rx-Bytes  : %s  \n",double_to_human_str((double)m_total_rx_bytes,"bytes",KBYE_1000).c_str());
 
 
-    
+
     fprintf (fd," Cpu Utilization : %2.1f  %%  %2.1f Gb/core \n",m_cpu_util,(2*(m_tx_bps/1e9)*100.0/(m_cpu_util*m_threads)));
     fprintf (fd," Platform_factor : %2.1f  \n",m_platform_factor);
     fprintf (fd," Total-Tx        : %s  ",double_to_human_str(m_tx_bps,"bps",KBYE_1000).c_str());
@@ -2527,76 +2485,76 @@ void CGlobalStats::Dump(FILE *fd,DumpFormat mode){
 
 
     if ( mode== dmpSTANDARD ){
-        fprintf (fd," --------------- \n"); 
+        fprintf (fd," --------------- \n");
         for (i=0; i<(int)port_to_show; i++) {
             CPerPortStats * lp=&m_port[i];
             fprintf(fd,"port : %d \n",(int)i);
             fprintf(fd,"------------\n");
-            #define GS_DP_A4(f) fprintf(fd," %-40s : %llu \n",#f, (unsigned long long)lp->f)
-            #define GS_DP_A(f) if (lp->f) fprintf(fd," %-40s : %llu \n",#f, (unsigned long long)lp->f)
-            GS_DP_A4(opackets);  
-            GS_DP_A4(obytes);      
-            GS_DP_A4(ipackets);  
-            GS_DP_A4(ibytes);      
-            GS_DP_A(ierrors);     
-            GS_DP_A(oerrors);     
+#define GS_DP_A4(f) fprintf(fd," %-40s : %llu \n",#f, (unsigned long long)lp->f)
+#define GS_DP_A(f) if (lp->f) fprintf(fd," %-40s : %llu \n",#f, (unsigned long long)lp->f)
+            GS_DP_A4(opackets);
+            GS_DP_A4(obytes);
+            GS_DP_A4(ipackets);
+            GS_DP_A4(ibytes);
+            GS_DP_A(ierrors);
+            GS_DP_A(oerrors);
             fprintf (fd," Tx : %s  \n",double_to_human_str((double)lp->m_total_tx_bps,"bps",KBYE_1000).c_str());
         }
     }else{
-            fprintf(fd," %10s ","ports");
-            for (i=0; i<(int)port_to_show; i++) {
-                fprintf(fd,"| %15d ",i);
-            }
-            fprintf(fd,"\n");
-            fprintf(fd," -----------------------------------------------------------------------------------------\n");
-            std::string names[]={"opackets","obytes","ipackets","ibytes","ierrors","oerrors","Tx Bw"
-            };
-            for (i=0; i<7; i++) {
-                fprintf(fd," %10s ",names[i].c_str());
-                int j=0;
-                for (j=0; j<port_to_show;j++) {
-                    CPerPortStats * lp=&m_port[j];
-                    uint64_t cnt;
-                    switch (i) {
-                    case 0:
-                        cnt=lp->opackets;
-                        fprintf(fd,"| %15lu ",cnt);
+        fprintf(fd," %10s ","ports");
+        for (i=0; i<(int)port_to_show; i++) {
+            fprintf(fd,"| %15d ",i);
+        }
+        fprintf(fd,"\n");
+        fprintf(fd," -----------------------------------------------------------------------------------------\n");
+        std::string names[]={"opackets","obytes","ipackets","ibytes","ierrors","oerrors","Tx Bw"
+        };
+        for (i=0; i<7; i++) {
+            fprintf(fd," %10s ",names[i].c_str());
+            int j=0;
+            for (j=0; j<port_to_show;j++) {
+                CPerPortStats * lp=&m_port[j];
+                uint64_t cnt;
+                switch (i) {
+                case 0:
+                    cnt=lp->opackets;
+                    fprintf(fd,"| %15lu ",cnt);
 
-                        break;
-                    case 1:
-                        cnt=lp->obytes;
-                        fprintf(fd,"| %15lu ",cnt);
+                    break;
+                case 1:
+                    cnt=lp->obytes;
+                    fprintf(fd,"| %15lu ",cnt);
 
-                        break;
-                    case 2:
-                        cnt=lp->ipackets;
-                        fprintf(fd,"| %15lu ",cnt);
+                    break;
+                case 2:
+                    cnt=lp->ipackets;
+                    fprintf(fd,"| %15lu ",cnt);
 
-                        break;
-                    case 3:
-                        cnt=lp->ibytes;
-                        fprintf(fd,"| %15lu ",cnt);
+                    break;
+                case 3:
+                    cnt=lp->ibytes;
+                    fprintf(fd,"| %15lu ",cnt);
 
-                        break;
-                    case 4:
-                        cnt=lp->ierrors;
-                        fprintf(fd,"| %15lu ",cnt);
+                    break;
+                case 4:
+                    cnt=lp->ierrors;
+                    fprintf(fd,"| %15lu ",cnt);
 
-                        break;
-                    case 5:
-                        cnt=lp->oerrors;
-                        fprintf(fd,"| %15lu ",cnt);
+                    break;
+                case 5:
+                    cnt=lp->oerrors;
+                    fprintf(fd,"| %15lu ",cnt);
 
-                        break;
-                    case 6:
-                        fprintf(fd,"| %15s ",double_to_human_str((double)lp->m_total_tx_bps,"bps",KBYE_1000).c_str());
-                        break;
-                    default:
-                        cnt=0xffffff;
-                    }
-                } /* ports */
-        		fprintf(fd, "\n");
-        	}/* fields*/
+                    break;
+                case 6:
+                    fprintf(fd,"| %15s ",double_to_human_str((double)lp->m_total_tx_bps,"bps",KBYE_1000).c_str());
+                    break;
+                default:
+                    cnt=0xffffff;
+                }
+            } /* ports */
+            fprintf(fd, "\n");
+        }/* fields*/
     }
 
 
@@ -2606,15 +2564,15 @@ class CGlobalTRex  {
 
 public:
     CGlobalTRex (){
-       m_max_ports=4;
-       m_max_cores=1;
-       m_cores_to_dual_ports=0;
-       m_max_queues_per_port=0;
-       m_fl_was_init=false;
-       m_expected_pps=0.0;                
-       m_expected_cps=0.0;
-       m_expected_bps=0.0;
-       m_trex_stateless = NULL;
+        m_max_ports=4;
+        m_max_cores=1;
+        m_cores_to_dual_ports=0;
+        m_max_queues_per_port=0;
+        m_fl_was_init=false;
+        m_expected_pps=0.0;
+        m_expected_cps=0.0;
+        m_expected_bps=0.0;
+        m_trex_stateless = NULL;
     }
 
     bool Create();
@@ -2634,25 +2592,24 @@ private:
     /* send message to all dp cores */
     int  send_message_all_dp(TrexStatelessCpToDpMsgBase *msg);
     void check_for_dp_message_from_core(int thread_id);
-    void check_for_dp_messages();
 
 public:
+    void check_for_dp_messages();
     int start_master_statefull();
     int start_master_stateless();
     int run_in_core(virtual_thread_id_t virt_core_id);
     int stop_core(virtual_thread_id_t virt_core_id);
-    int core_for_latency(){
-        if ( (!get_is_latency_thread_enable()) ){
-            return (-1);
+    int core_for_rx(){
+        if ( (! get_is_rx_thread_enabled()) ) {
+            return -1;
         }else{
-                return ( m_max_cores - 1 );
-            }
-
+            return m_max_cores - 1;
+        }
     }
-    int run_in_laterncy_core();
+    int run_in_rx_core();
     int run_in_master();
     int stop_master();
-    /* return the minimum number of dp cores needed to support the active ports 
+    /* return the minimum number of dp cores needed to support the active ports
        this is for c==1 or  m_cores_mul==1
     */
     int get_base_num_cores(){
@@ -2660,12 +2617,12 @@ public:
     }
 
     int get_cores_tx(){
-        /* 0 - master 
-           num_of_cores - 
+        /* 0 - master
+           num_of_cores -
            last for latency */
-        if ( (!get_is_latency_thread_enable()) ){
+        if ( (! get_is_rx_thread_enabled()) ) {
             return (m_max_cores - 1 );
-        }else{
+        } else {
             return (m_max_cores - BP_MASTER_AND_LATENCY );
         }
     }
@@ -2675,7 +2632,7 @@ private:
 
 public:
 
-    void publish_async_data();
+    void publish_async_data(bool sync_now);
     void publish_async_barrier(uint32_t key);
 
     void dump_stats(FILE *fd,
@@ -2683,6 +2640,8 @@ public:
     void dump_template_info(std::string & json);
     bool sanity_check();
     void update_stats(void);
+    tx_per_flow_t get_flow_tx_stats(uint8_t port, uint16_t hw_id);
+    void clear_flow_tx_stats(uint8_t port, uint16_t index);
     void get_stats(CGlobalStats & stats);
     void dump_post_test_stats(FILE *fd);
     void dump_config(FILE *fd);
@@ -2695,14 +2654,14 @@ public:
     uint32_t    m_max_queues_per_port;
     uint32_t    m_cores_to_dual_ports; /* number of ports that will handle dual ports */
     uint16_t    m_latency_tx_queue_id;
-    // statistic 
+    // statistic
     CPPSMeasure  m_cps;
-    float        m_expected_pps;                
-    float        m_expected_cps;                
-    float        m_expected_bps;//bps           
+    float        m_expected_pps;
+    float        m_expected_cps;
+    float        m_expected_bps;//bps
     float        m_last_total_cps;
 
-    CPhyEthIF   m_ports[BP_MAX_PORTS];
+    CPhyEthIF   m_ports[TREX_MAX_PORTS];
     CCoreEthIF          m_cores_vif_sf[BP_MAX_CORES]; /* counted from 1 , 2,3 core zero is reserved - stateful */
     CCoreEthIFStateless m_cores_vif_sl[BP_MAX_CORES]; /* counted from 1 , 2,3 core zero is reserved - stateless*/
     CCoreEthIF *        m_cores_vif[BP_MAX_CORES];
@@ -2714,14 +2673,16 @@ public:
     CTrexGlobalIoMode   m_io_modes;
 
 private:
-    CLatencyHWPort      m_latency_vports[BP_MAX_PORTS];    /* read hardware driver */
-    CLatencyVmPort      m_latency_vm_vports[BP_MAX_PORTS]; /* vm driver */
+    CLatencyHWPort      m_latency_vports[TREX_MAX_PORTS];    /* read hardware driver */
+    CLatencyVmPort      m_latency_vm_vports[TREX_MAX_PORTS]; /* vm driver */
     CLatencyPktInfo     m_latency_pkt;
     TrexPublisher       m_zmq_publisher;
     CGlobalStats        m_stats;
+    std::mutex          m_cp_lock;
 
 public:
     TrexStateless       *m_trex_stateless;
+
 };
 
 int  CGlobalTRex::reset_counters(){
@@ -2736,12 +2697,12 @@ int  CGlobalTRex::reset_counters(){
 
 /**
  * check for a single core
- * 
+ *
  * @author imarom (19-Nov-15)
- * 
- * @param thread_id 
+ *
+ * @param thread_id
  */
-void 
+void
 CGlobalTRex::check_for_dp_message_from_core(int thread_id) {
 
     CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingDpToCp(thread_id);
@@ -2767,10 +2728,11 @@ CGlobalTRex::check_for_dp_message_from_core(int thread_id) {
 
 /**
  * check for messages that arrived from DP to CP
- * 
+ *
  */
-void 
+void
 CGlobalTRex::check_for_dp_messages() {
+
     /* for all the cores - check for a new message */
     for (int i = 0; i < get_cores_tx(); i++) {
         check_for_dp_message_from_core(i);
@@ -2801,7 +2763,7 @@ void CGlobalTRex::try_stop_all_dp(){
     send_message_all_dp(msg);
     delete msg;
     bool all_core_finished = false;
-    int i; 
+    int i;
     for (i=0; i<20; i++) {
         if ( is_all_cores_finished() ){
             all_core_finished =true;
@@ -2820,14 +2782,14 @@ void CGlobalTRex::try_stop_all_dp(){
 
 int  CGlobalTRex::send_message_all_dp(TrexStatelessCpToDpMsgBase *msg){
 
-        int max_threads=(int)CMsgIns::Ins()->getCpDp()->get_num_threads();
-        int i;
+    int max_threads=(int)CMsgIns::Ins()->getCpDp()->get_num_threads();
+    int i;
 
-        for (i=0; i<max_threads; i++) {
-            CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingCpToDp((uint8_t)i);
-            ring->Enqueue((CGenNode*)msg->clone());
-        }
-        return (0);
+    for (i=0; i<max_threads; i++) {
+        CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingCpToDp((uint8_t)i);
+        ring->Enqueue((CGenNode*)msg->clone());
+    }
+    return (0);
 }
 
 
@@ -2851,7 +2813,7 @@ int  CGlobalTRex::ixgbe_configure_mg(void){
     if ( latency_rate ) {
         mg_cfg.m_cps = (double)latency_rate ;
     }else{
-        mg_cfg.m_cps = 100.0;
+        mg_cfg.m_cps = 1.0;
     }
 
     if ( get_vm_one_queue_enable() ) {
@@ -2860,7 +2822,7 @@ int  CGlobalTRex::ixgbe_configure_mg(void){
 
             CMessagingManager * rx_dp=CMsgIns::Ins()->getRxDp();
 
-            uint8_t thread_id = (i>>1); 
+            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);
@@ -2915,7 +2877,7 @@ int  CGlobalTRex::ixgbe_start(void){
             _if->set_rx_queue(0);
             _if->rx_queue_setup(0,
                                 RTE_TEST_RX_DESC_VM_DEFAULT,
-                                socket_id, 
+                                socket_id,
                                 &m_port_cfg.m_rx_conf,
                                 CGlobalInfo::m_mem_pool[socket_id].m_mbuf_pool_2048);
 
@@ -2923,15 +2885,15 @@ int  CGlobalTRex::ixgbe_start(void){
             for ( qid=0; qid<(m_max_queues_per_port); qid++) {
                 _if->tx_queue_setup((uint16_t)qid,
                                     RTE_TEST_TX_DESC_VM_DEFAULT ,
-                                    socket_id, 
+                                    socket_id,
                                     &m_port_cfg.m_tx_conf);
 
             }
 
         }else{
             _if->configure(2,
-                          m_cores_to_dual_ports+1,
-                          &m_port_cfg.m_port_conf);
+                           m_cores_to_dual_ports+1,
+                           &m_port_cfg.m_port_conf);
 
             /* the latency queue for latency measurement packets */
             m_latency_tx_queue_id= m_cores_to_dual_ports;
@@ -2943,7 +2905,7 @@ int  CGlobalTRex::ixgbe_start(void){
             /* drop queue */
             _if->rx_queue_setup(0,
                                 RTE_TEST_RX_DESC_DEFAULT,
-                                socket_id, 
+                                socket_id,
                                 &m_port_cfg.m_rx_conf,
                                 CGlobalInfo::m_mem_pool[socket_id].m_mbuf_pool_2048);
 
@@ -2953,7 +2915,7 @@ int  CGlobalTRex::ixgbe_start(void){
             /* latency measurement ring is 1 */
             _if->rx_queue_setup(1,
                                 RTE_TEST_RX_LATENCY_DESC_DEFAULT,
-                                socket_id, 
+                                socket_id,
                                 &m_port_cfg.m_rx_conf,
                                 CGlobalInfo::m_mem_pool[socket_id].m_mbuf_pool_9k);
 
@@ -2961,7 +2923,7 @@ int  CGlobalTRex::ixgbe_start(void){
             for ( qid=0; qid<(m_max_queues_per_port+1); qid++) {
                 _if->tx_queue_setup((uint16_t)qid,
                                     RTE_TEST_TX_DESC_DEFAULT ,
-                                    socket_id, 
+                                    socket_id,
                                     &m_port_cfg.m_tx_conf);
 
             }
@@ -2975,15 +2937,15 @@ int  CGlobalTRex::ixgbe_start(void){
         _if->configure_rx_drop_queue();
         _if->configure_rx_duplicate_rules();
 
-       if ( ! get_vm_one_queue_enable()  && ! CGlobalInfo::m_options.preview.get_is_disable_flow_control_setting() 
-	    && get_ex_drv()->flow_control_disable_supported()) {
-	   _if->disable_flow_control();
-       }
+        if ( ! get_vm_one_queue_enable()  && ! CGlobalInfo::m_options.preview.get_is_disable_flow_control_setting()
+             && get_ex_drv()->flow_control_disable_supported()) {
+            _if->disable_flow_control();
+        }
 
         _if->update_link_status();
 
         _if->dump_link(stdout);
-        
+
         _if->add_mac((char *)CGlobalInfo::m_options.get_src_mac_addr(i));
 
         fflush(stdout);
@@ -2995,7 +2957,7 @@ int  CGlobalTRex::ixgbe_start(void){
 
         if ( !is_all_links_are_up(true) ){
             rte_exit(EXIT_FAILURE, " "
-                    " one of the link is down \n");
+                     " one of the link is down \n");
         }
     } else {
         get_ex_drv()->wait_after_link_up();
@@ -3009,9 +2971,9 @@ int  CGlobalTRex::ixgbe_start(void){
 
     /* core 0 - control
        core 1 - port 0-0,1-0,
-       core 2 - port 2-0,3-0, 
-       core 3 - port 0-1,1-1, 
-       core 4 - port 2-1,3-1, 
+       core 2 - port 2-0,3-0,
+       core 3 - port 0-1,1-1,
+       core 4 - port 2-1,3-1,
 
     */
     int port_offset=0;
@@ -3024,16 +2986,16 @@ int  CGlobalTRex::ixgbe_start(void){
             m_cores_vif[j]=&m_cores_vif_sf[j];
         }
         m_cores_vif[j]->Create(j,
-                              queue_id,
-                              &m_ports[port_offset], /* 0,2*/
-                              queue_id,
-                              &m_ports[port_offset+1] /*1,3*/
-                              );
+                               queue_id,
+                               &m_ports[port_offset], /* 0,2*/
+                               queue_id,
+                               &m_ports[port_offset+1] /*1,3*/
+                               );
         port_offset+=2;
         if (port_offset == m_max_ports) {
             port_offset = 0;
-        }    
-     }
+        }
+    }
 
     fprintf(stdout," -------------------------------\n");
     CCoreEthIF::DumpIfCfgHeader(stdout);
@@ -3058,22 +3020,22 @@ bool CGlobalTRex::Create(){
         return (false);
     }
 
-   if ( pre_yaml_info.m_vlan_info.m_enable ){
-       CGlobalInfo::m_options.preview.set_vlan_mode_enable(true);
-   }
-   /* End update pre flags */
+    if ( pre_yaml_info.m_vlan_info.m_enable ){
+        CGlobalInfo::m_options.preview.set_vlan_mode_enable(true);
+    }
+    /* End update pre flags */
 
-   ixgbe_prob_init();
-   cores_prob_init();
-   queues_prob_init();
+    ixgbe_prob_init();
+    cores_prob_init();
+    queues_prob_init();
 
-   /* allocate rings */
-   assert( CMsgIns::Ins()->Create(get_cores_tx()) );
+    /* allocate rings */
+    assert( CMsgIns::Ins()->Create(get_cores_tx()) );
 
-   if ( sizeof(CGenNodeNatInfo) != sizeof(CGenNode)  ) {
-       printf("ERROR sizeof(CGenNodeNatInfo) %lu != sizeof(CGenNode) %lu must be the same size \n",sizeof(CGenNodeNatInfo),sizeof(CGenNode));
-       assert(0);
-   }
+    if ( sizeof(CGenNodeNatInfo) != sizeof(CGenNode)  ) {
+        printf("ERROR sizeof(CGenNodeNatInfo) %lu != sizeof(CGenNode) %lu must be the same size \n",sizeof(CGenNodeNatInfo),sizeof(CGenNode));
+        assert(0);
+    }
 
     if ( sizeof(CGenNodeLatencyPktInfo) != sizeof(CGenNode)  ) {
         printf("ERROR sizeof(CGenNodeLatencyPktInfo) %lu != sizeof(CGenNode) %lu must be the same size \n",sizeof(CGenNodeLatencyPktInfo),sizeof(CGenNode));
@@ -3082,36 +3044,36 @@ bool CGlobalTRex::Create(){
 
     /* allocate the memory */
 
-   uint32_t rx_mbuf = 0 ;
+    uint32_t rx_mbuf = 0 ;
 
-   if ( get_vm_one_queue_enable() ) {
+    if ( get_vm_one_queue_enable() ) {
         rx_mbuf = (m_max_ports * RTE_TEST_RX_DESC_VM_DEFAULT);
-   }else{
+    }else{
         rx_mbuf = (m_max_ports * (RTE_TEST_RX_LATENCY_DESC_DEFAULT+RTE_TEST_RX_DESC_DEFAULT));
-   }
+    }
 
-   CGlobalInfo::init_pools(rx_mbuf);
-   ixgbe_start();
-   dump_config(stdout);
+    CGlobalInfo::init_pools(rx_mbuf);
+    ixgbe_start();
+    dump_config(stdout);
 
-   /* start stateless */
-   if (get_is_stateless()) {
+    /* start stateless */
+    if (get_is_stateless()) {
 
-       TrexStatelessCfg cfg;
+        TrexStatelessCfg cfg;
 
-       TrexRpcServerConfig rpc_req_resp_cfg(TrexRpcServerConfig::RPC_PROT_TCP, global_platform_cfg_info.m_zmq_rpc_port);
+        TrexRpcServerConfig rpc_req_resp_cfg(TrexRpcServerConfig::RPC_PROT_TCP, global_platform_cfg_info.m_zmq_rpc_port);
 
-       cfg.m_port_count         = CGlobalInfo::m_options.m_expected_portd;
-       cfg.m_rpc_req_resp_cfg   = &rpc_req_resp_cfg;
-       cfg.m_rpc_async_cfg      = NULL;
-       cfg.m_rpc_server_verbose = false;
-       cfg.m_platform_api       = new TrexDpdkPlatformApi();
-       cfg.m_publisher          = &m_zmq_publisher;
+        cfg.m_port_count         = CGlobalInfo::m_options.m_expected_portd;
+        cfg.m_rpc_req_resp_cfg   = &rpc_req_resp_cfg;
+        cfg.m_rpc_server_verbose = false;
+        cfg.m_platform_api       = new TrexDpdkPlatformApi();
+        cfg.m_publisher          = &m_zmq_publisher;
+        cfg.m_global_lock        = &m_cp_lock;
 
-       m_trex_stateless = new TrexStateless(cfg);
-   }
+        m_trex_stateless = new TrexStateless(cfg);
+    }
 
-   return (true);
+    return (true);
 
 }
 void CGlobalTRex::Delete(){
@@ -3122,9 +3084,9 @@ void CGlobalTRex::Delete(){
 
 int  CGlobalTRex::ixgbe_prob_init(void){
 
-	m_max_ports  = rte_eth_dev_count();
-	if (m_max_ports == 0)
-		rte_exit(EXIT_FAILURE, "No Ethernet ports - bye\n");
+    m_max_ports  = rte_eth_dev_count();
+    if (m_max_ports == 0)
+        rte_exit(EXIT_FAILURE, "No Ethernet ports - bye\n");
 
     printf(" Number of ports found: %d \n",m_max_ports);
 
@@ -3133,8 +3095,8 @@ int  CGlobalTRex::ixgbe_prob_init(void){
                  m_max_ports);
     }
 
-    if ( CGlobalInfo::m_options.get_expected_ports() >BP_MAX_PORTS ){
-        rte_exit(EXIT_FAILURE, " Maximum ports supported are %d, use the configuration file to set the expected number of ports   \n",BP_MAX_PORTS);
+    if ( CGlobalInfo::m_options.get_expected_ports() > TREX_MAX_PORTS ) {
+        rte_exit(EXIT_FAILURE, " Maximum ports supported are %d, use the configuration file to set the expected number of ports   \n",TREX_MAX_PORTS);
     }
 
     if ( CGlobalInfo::m_options.get_expected_ports() > m_max_ports ){
@@ -3146,7 +3108,7 @@ int  CGlobalTRex::ixgbe_prob_init(void){
         /* limit the number of ports */
         m_max_ports=CGlobalInfo::m_options.get_expected_ports();
     }
-    assert(m_max_ports <= BP_MAX_PORTS);
+    assert(m_max_ports <= TREX_MAX_PORTS);
 
     struct rte_eth_dev_info dev_info;
     rte_eth_dev_info_get((uint8_t) 0,&dev_info);
@@ -3218,23 +3180,23 @@ int  CGlobalTRex::queues_prob_init(){
 
     m_cores_to_dual_ports  = m_cores_mul;
 
-    /* core 0 - control 
+    /* core 0 - control
        -core 1 - port 0/1
        -core 2 - port 2/3
        -core 3 - port 0/1
        -core 4 - port 2/3
 
        m_cores_to_dual_ports = 2;
-     */
-    
+    */
+
     /* number of queue - 1 per core for dual ports*/
     m_max_queues_per_port  = m_cores_to_dual_ports;
 
     if (m_max_queues_per_port > BP_MAX_TX_QUEUE) {
-        rte_exit(EXIT_FAILURE, 
-         "maximum number of queue should be maximum %d  \n",BP_MAX_TX_QUEUE);
+        rte_exit(EXIT_FAILURE,
+                 "maximum number of queue should be maximum %d  \n",BP_MAX_TX_QUEUE);
     }
-    
+
     assert(m_max_queues_per_port>0);
     return (0);
 }
@@ -3275,7 +3237,7 @@ void CGlobalTRex::dump_post_test_stats(FILE *fd){
         pkt_out +=_if->get_stats().opackets;
         pkt_out_bytes +=_if->get_stats().obytes;
     }
-    if ( !CGlobalInfo::m_options.is_latency_disabled() ){
+    if ( CGlobalInfo::m_options.is_latency_enabled() ){
         sw_pkt_out += m_mg.get_total_pkt();
         sw_pkt_out_bytes +=m_mg.get_total_bytes();
     }
@@ -3283,13 +3245,13 @@ void CGlobalTRex::dump_post_test_stats(FILE *fd){
 
     fprintf (fd," summary stats \n");
     fprintf (fd," -------------- \n");
-    
+
     if (pkt_in > pkt_out)
-    {
-        fprintf (fd, " Total-pkt-drop       : 0 pkts \n");
-        if (pkt_in > pkt_out * 1.01)
-            fprintf (fd, " Warning : number of rx packets exceeds 101%% of tx packets!\n");
-    }
+        {
+            fprintf (fd, " Total-pkt-drop       : 0 pkts \n");
+            if (pkt_in > pkt_out * 1.01)
+                fprintf (fd, " Warning : number of rx packets exceeds 101%% of tx packets!\n");
+        }
     else
         fprintf (fd, " Total-pkt-drop       : %llu pkts \n", (unsigned long long) (pkt_out - pkt_in));
     fprintf (fd," Total-tx-bytes       : %llu bytes \n", (unsigned long long)pkt_out_bytes);
@@ -3304,7 +3266,7 @@ void CGlobalTRex::dump_post_test_stats(FILE *fd){
     fprintf (fd," Total-sw-err         : %llu pkts \n", (unsigned long long)sw_pkt_out_err);
 
 
-    if ( !CGlobalInfo::m_options.is_latency_disabled() ){
+    if ( CGlobalInfo::m_options.is_latency_enabled() ){
         fprintf (fd," maximum-latency   : %.0f usec \n",m_mg.get_max_latency());
         fprintf (fd," average-latency   : %.0f usec \n",m_mg.get_avr_latency());
         fprintf (fd," latency-any-error : %s  \n",m_mg.is_any_error()?"ERROR":"OK");
@@ -3334,6 +3296,13 @@ void CGlobalTRex::update_stats(){
 
 }
 
+tx_per_flow_t CGlobalTRex::get_flow_tx_stats(uint8_t port, uint16_t index) {
+    return m_stats.m_port[port].m_tx_per_flow[index] - m_stats.m_port[port].m_prev_tx_per_flow[index];
+}
+
+void CGlobalTRex::clear_flow_tx_stats(uint8_t port, uint16_t index) {
+    m_stats.m_port[port].m_prev_tx_per_flow[index] = m_stats.m_port[port].m_tx_per_flow[index];
+}
 
 void CGlobalTRex::get_stats(CGlobalStats & stats){
 
@@ -3363,17 +3332,17 @@ void CGlobalTRex::get_stats(CGlobalStats & stats){
         CPhyEthIFStats & st =_if->get_stats();
 
         stp->opackets = st.opackets;
-        stp->obytes   = st.obytes;     
-        stp->ipackets = st.ipackets; 
+        stp->obytes   = st.obytes;
+        stp->ipackets = st.ipackets;
         stp->ibytes   = st.ibytes;
         stp->ierrors  = st.ierrors;
-        stp->oerrors  = st.oerrors;    
+        stp->oerrors  = st.oerrors;
         stp->m_total_tx_bps = _if->get_last_tx_rate()*_1Mb_DOUBLE;
         stp->m_total_tx_pps = _if->get_last_tx_pps_rate();
         stp->m_total_rx_bps = _if->get_last_rx_rate()*_1Mb_DOUBLE;
         stp->m_total_rx_pps = _if->get_last_rx_pps_rate();
 
-        stats.m_total_tx_pkts  += st.opackets; 
+        stats.m_total_tx_pkts  += st.opackets;
         stats.m_total_rx_pkts  += st.ipackets;
         stats.m_total_tx_bytes += st.obytes;
         stats.m_total_rx_bytes += st.ibytes;
@@ -3383,6 +3352,9 @@ void CGlobalTRex::get_stats(CGlobalStats & stats){
         total_tx_pps +=_if->get_last_tx_pps_rate();
         total_rx_pps +=_if->get_last_rx_pps_rate();
 
+        for (uint16_t flow = 0; flow <= max_stat_hw_id_seen; flow++) {
+            stats.m_port[i].m_tx_per_flow[flow].clear();
+        }
     }
 
     uint64_t total_open_flows=0;
@@ -3400,22 +3372,20 @@ void CGlobalTRex::get_stats(CGlobalStats & stats){
     uint64_t total_nat_open     =0;
     uint64_t total_nat_learn_error=0;
 
-
     CFlowGenListPerThread   * lpt;
     stats.m_template.Clear();
-
     for (i=0; i<get_cores_tx(); i++) {
         lpt = m_fl.m_threads_info[i];
         total_open_flows +=   lpt->m_stats.m_total_open_flows ;
         total_active_flows += (lpt->m_stats.m_total_open_flows-lpt->m_stats.m_total_close_flows) ;
 
         stats.m_total_alloc_error += lpt->m_node_gen.m_v_if->m_stats[0].m_tx_alloc_error+
-                               lpt->m_node_gen.m_v_if->m_stats[1].m_tx_alloc_error;
+            lpt->m_node_gen.m_v_if->m_stats[1].m_tx_alloc_error;
         stats.m_total_queue_full +=lpt->m_node_gen.m_v_if->m_stats[0].m_tx_queue_full+
-                               lpt->m_node_gen.m_v_if->m_stats[1].m_tx_queue_full;
+            lpt->m_node_gen.m_v_if->m_stats[1].m_tx_queue_full;
 
-        stats.m_total_queue_drop =lpt->m_node_gen.m_v_if->m_stats[0].m_tx_drop+
-                               lpt->m_node_gen.m_v_if->m_stats[1].m_tx_drop;
+        stats.m_total_queue_drop +=lpt->m_node_gen.m_v_if->m_stats[0].m_tx_drop+
+            lpt->m_node_gen.m_v_if->m_stats[1].m_tx_drop;
 
         stats.m_template.Add(&lpt->m_node_gen.m_v_if->m_stats[0].m_template);
         stats.m_template.Add(&lpt->m_node_gen.m_v_if->m_stats[1].m_template);
@@ -3431,6 +3401,13 @@ void CGlobalTRex::get_stats(CGlobalStats & stats){
         total_nat_active   +=lpt->m_stats.m_nat_lookup_add_flow_id - lpt->m_stats.m_nat_lookup_remove_flow_id;
         total_nat_open     +=lpt->m_stats.m_nat_lookup_add_flow_id;
         total_nat_learn_error   +=lpt->m_stats.m_nat_flow_learn_error;
+        uint8_t port0 = lpt->getDualPortId() *2;
+        for (uint16_t flow = 0; flow <= max_stat_hw_id_seen; flow++) {
+            stats.m_port[port0].m_tx_per_flow[flow] +=
+                lpt->m_node_gen.m_v_if->m_stats[0].m_tx_per_flow[flow];
+            stats.m_port[port0 + 1].m_tx_per_flow[flow] +=
+                lpt->m_node_gen.m_v_if->m_stats[1].m_tx_per_flow[flow];
+        }
     }
 
     stats.m_total_nat_time_out = total_nat_time_out;
@@ -3448,7 +3425,7 @@ void CGlobalTRex::get_stats(CGlobalStats & stats){
     } else {
         stats.m_socket_util = 0;
     }
-    
+
 
 
     float drop_rate=total_tx-total_rx;
@@ -3516,27 +3493,27 @@ void CGlobalTRex::dump_stats(FILE *fd, CGlobalStats::DumpFormat format){
     if (format==CGlobalStats::dmpTABLE) {
         if ( m_io_modes.m_g_mode == CTrexGlobalIoMode::gNORMAL ){
             switch (m_io_modes.m_pp_mode ){
-                case CTrexGlobalIoMode::ppDISABLE:
-                    fprintf(fd,"\n+Per port stats disabled \n");
-                    break;
-                case CTrexGlobalIoMode::ppTABLE:
-                    fprintf(fd,"\n-Per port stats table \n");
-                    m_stats.Dump(fd,CGlobalStats::dmpTABLE);
-                    break;
-                case CTrexGlobalIoMode::ppSTANDARD:
-                    fprintf(fd,"\n-Per port stats - standard\n");
-                    m_stats.Dump(fd,CGlobalStats::dmpSTANDARD);
-                    break;
+            case CTrexGlobalIoMode::ppDISABLE:
+                fprintf(fd,"\n+Per port stats disabled \n");
+                break;
+            case CTrexGlobalIoMode::ppTABLE:
+                fprintf(fd,"\n-Per port stats table \n");
+                m_stats.Dump(fd,CGlobalStats::dmpTABLE);
+                break;
+            case CTrexGlobalIoMode::ppSTANDARD:
+                fprintf(fd,"\n-Per port stats - standard\n");
+                m_stats.Dump(fd,CGlobalStats::dmpSTANDARD);
+                break;
             };
 
             switch (m_io_modes.m_ap_mode ){
-                case   CTrexGlobalIoMode::apDISABLE:
-                    fprintf(fd,"\n+Global stats disabled \n");
-                    break;
-                case   CTrexGlobalIoMode::apENABLE:
-                    fprintf(fd,"\n-Global stats enabled \n");
-                    m_stats.DumpAllPorts(fd);
-                    break;
+            case   CTrexGlobalIoMode::apDISABLE:
+                fprintf(fd,"\n+Global stats disabled \n");
+                break;
+            case   CTrexGlobalIoMode::apENABLE:
+                fprintf(fd,"\n-Global stats enabled \n");
+                m_stats.DumpAllPorts(fd);
+                break;
             };
         }
     }else{
@@ -3544,60 +3521,66 @@ void CGlobalTRex::dump_stats(FILE *fd, CGlobalStats::DumpFormat format){
         m_stats.Dump(fd,format);
         m_stats.DumpAllPorts(fd);
     }
-    
+
 }
 
+void
+CGlobalTRex::publish_async_data(bool sync_now) {
+    std::string json;
 
-void 
-CGlobalTRex::publish_async_data() {
-     std::string json;
+    /* refactor to update, dump, and etc. */
+    if (sync_now) {
+        update_stats();
+        get_stats(m_stats);
+    }
 
-     m_stats.dump_json(json);
-     m_zmq_publisher.publish_json(json);
+    m_stats.dump_json(json);
+    m_zmq_publisher.publish_json(json);
 
-     /* generator json , all cores are the same just sample the first one */
-     m_fl.m_threads_info[0]->m_node_gen.dump_json(json);
-     m_zmq_publisher.publish_json(json);
+    /* generator json , all cores are the same just sample the first one */
+    m_fl.m_threads_info[0]->m_node_gen.dump_json(json);
+    m_zmq_publisher.publish_json(json);
 
-        
-     if ( !get_is_stateless() ){
-         dump_template_info(json);
-         m_zmq_publisher.publish_json(json);
-     }
 
-     if ( get_is_rx_check_mode() ) {
-         m_mg.rx_check_dump_json(json );
-         m_zmq_publisher.publish_json(json);
-     }
+    if ( !get_is_stateless() ){
+        dump_template_info(json);
+        m_zmq_publisher.publish_json(json);
+    }
+
+    if ( get_is_rx_check_mode() ) {
+        m_mg.rx_check_dump_json(json );
+        m_zmq_publisher.publish_json(json);
+    }
 
-     /* backward compatible */
-     m_mg.dump_json(json );
-     m_zmq_publisher.publish_json(json);
+    /* backward compatible */
+    m_mg.dump_json(json );
+    m_zmq_publisher.publish_json(json);
 
-     /* more info */
-     m_mg.dump_json_v2(json );
-     m_zmq_publisher.publish_json(json);
+    /* more info */
+    m_mg.dump_json_v2(json );
+    m_zmq_publisher.publish_json(json);
 
-     if (get_is_stateless()) {
-         if (m_trex_stateless->m_rx_flow_stat.dump_json(json))
-             m_zmq_publisher.publish_json(json);
-     }
+    if (get_is_stateless()) {
+        if (m_trex_stateless->m_rx_flow_stat.dump_json(json))
+            m_zmq_publisher.publish_json(json);
+    }
 }
 
-void 
+void
 CGlobalTRex::publish_async_barrier(uint32_t key) {
     m_zmq_publisher.publish_barrier(key);
 }
 
 int CGlobalTRex::run_in_master() {
-
-   
     bool was_stopped=false;
 
     if ( get_is_stateless() ) {
         m_trex_stateless->launch_control_plane();
     }
 
+    /* exception and scope safe */
+    std::unique_lock<std::mutex> cp_lock(m_cp_lock);
+
     while ( true ) {
 
         if ( CGlobalInfo::m_options.preview.get_no_keyboard() ==false ){
@@ -3626,7 +3609,7 @@ int CGlobalTRex::run_in_master() {
             }
         }
 
-        
+
         if (m_io_modes.m_g_mode == CTrexGlobalIoMode::gHELP ) {
             m_io_modes.DumpHelp(stdout);
         }
@@ -3634,17 +3617,17 @@ int CGlobalTRex::run_in_master() {
         dump_stats(stdout,CGlobalStats::dmpTABLE);
 
         if (m_io_modes.m_g_mode == CTrexGlobalIoMode::gNORMAL ) {
-    		fprintf (stdout," current time    : %.1f sec  \n",now_sec());
-    		float d= CGlobalInfo::m_options.m_duration - now_sec();
-    		if (d<0) {
-    			d=0;
-    
-    		}
-    		fprintf (stdout," test duration   : %.1f sec  \n",d);
+            fprintf (stdout," current time    : %.1f sec  \n",now_sec());
+            float d= CGlobalInfo::m_options.m_duration - now_sec();
+            if (d<0) {
+                d=0;
+
+            }
+            fprintf (stdout," test duration   : %.1f sec  \n",d);
         }
 
 
-        if ( !CGlobalInfo::m_options.is_latency_disabled() ){
+        if ( CGlobalInfo::m_options.is_rx_enabled() ){
             m_mg.update();
 
             if ( m_io_modes.m_g_mode ==  CTrexGlobalIoMode::gNORMAL ){
@@ -3662,43 +3645,45 @@ int CGlobalTRex::run_in_master() {
                     break;
                 }
 
-             if ( get_is_rx_check_mode() ) {
+                if ( get_is_rx_check_mode() ) {
 
-                switch (m_io_modes.m_rc_mode) {
-                case CTrexGlobalIoMode::rcDISABLE:
-                    fprintf(stdout,"\n+Rx Check stats disabled \n");
-                    break;
-                case CTrexGlobalIoMode::rcENABLE:
-                    fprintf(stdout,"\n-Rx Check stats enabled \n");
-                    m_mg.DumpShortRxCheck(stdout);
-                    break;
-                case CTrexGlobalIoMode::rcENABLE_Extended:
-                    fprintf(stdout,"\n-Rx Check stats enhanced \n");
-                    m_mg.DumpRxCheck(stdout);
-                    break;
-                }
+                    switch (m_io_modes.m_rc_mode) {
+                    case CTrexGlobalIoMode::rcDISABLE:
+                        fprintf(stdout,"\n+Rx Check stats disabled \n");
+                        break;
+                    case CTrexGlobalIoMode::rcENABLE:
+                        fprintf(stdout,"\n-Rx Check stats enabled \n");
+                        m_mg.DumpShortRxCheck(stdout);
+                        break;
+                    case CTrexGlobalIoMode::rcENABLE_Extended:
+                        fprintf(stdout,"\n-Rx Check stats enhanced \n");
+                        m_mg.DumpRxCheck(stdout);
+                        break;
+                    }
 
-              }/* ex checked */
+                }
 
             }
-
-          
-
         }
 
         /* publish data */
-        publish_async_data();
+        publish_async_data(false);
 
         /* check from messages from DP */
         check_for_dp_messages();
 
+        cp_lock.unlock();
         delay(500);
+        cp_lock.lock();
 
         if ( is_all_cores_finished() ) {
             break;
         }
     }
 
+    /* on exit release the lock */
+    cp_lock.unlock();
+
     if (!is_all_cores_finished()) {
         /* probably CLTR-C */
         try_stop_all_dp();
@@ -3715,10 +3700,11 @@ int CGlobalTRex::run_in_master() {
 
 
 
-int CGlobalTRex::run_in_laterncy_core(void){
-    if ( !CGlobalInfo::m_options.is_latency_disabled() ){
+int CGlobalTRex::run_in_rx_core(void){
+    if ( CGlobalInfo::m_options.is_rx_enabled() ){
         m_mg.start(0);
     }
+    // ??? start stateless rx
     return (0);
 }
 
@@ -3731,8 +3717,8 @@ int CGlobalTRex::stop_core(virtual_thread_id_t virt_core_id){
 int CGlobalTRex::run_in_core(virtual_thread_id_t virt_core_id){
 
     CPreviewMode *lp=&CGlobalInfo::m_options.preview;
-    if ( lp->getSingleCore() && 
-         (virt_core_id==2 ) && 
+    if ( lp->getSingleCore() &&
+         (virt_core_id==2 ) &&
          (lp-> getCores() ==1) ){
         printf(" bypass this core \n");
         m_signal[virt_core_id]=1;
@@ -3771,7 +3757,7 @@ int CGlobalTRex::stop_master(){
 
     CFlowGenListPerThread   * lpt;
     uint64_t total_tx_rx_check=0;
-    
+
     int i;
     for (i=0; i<get_cores_tx(); i++) {
         lpt = m_fl.m_threads_info[i];
@@ -3780,7 +3766,7 @@ int CGlobalTRex::stop_master(){
         erf_vif->DumpCoreStats(stdout);
         erf_vif->DumpIfStats(stdout);
         total_tx_rx_check+=erf_vif->m_stats[CLIENT_SIDE].m_tx_rx_check_pkt+
-                           erf_vif->m_stats[SERVER_SIDE].m_tx_rx_check_pkt;
+            erf_vif->m_stats[SERVER_SIDE].m_tx_rx_check_pkt;
     }
 
     fprintf(stdout," ==================\n");
@@ -3791,7 +3777,7 @@ int CGlobalTRex::stop_master(){
         lpt->m_node_gen.DumpHist(stdout);
         lpt->DumpStats(stdout);
     }
-    if ( !CGlobalInfo::m_options.is_latency_disabled() ){
+    if ( CGlobalInfo::m_options.is_latency_enabled() ){
         fprintf(stdout," ==================\n");
         fprintf(stdout," latency \n");
         fprintf(stdout," ==================\n");
@@ -3801,7 +3787,7 @@ int CGlobalTRex::stop_master(){
         m_mg.DumpRxCheck(stdout);
         m_mg.DumpRxCheckVerification(stdout,total_tx_rx_check);
     }
-    
+
     dump_stats(stdout,CGlobalStats::dmpSTANDARD);
     dump_post_test_stats(stdout);
     m_fl.Delete();
@@ -3859,9 +3845,9 @@ int CGlobalTRex::start_master_statefull() {
     } else {
         m_fl.m_mac_info.set_configured(false);
     }
- 
-    m_expected_pps = m_fl.get_total_pps();     
-    m_expected_cps = 1000.0*m_fl.get_total_kcps();               
+
+    m_expected_pps = m_fl.get_total_pps();
+    m_expected_cps = 1000.0*m_fl.get_total_kcps();
     m_expected_bps = m_fl.get_total_tx_bps();
     if ( m_fl.get_total_repeat_flows() > 2000) {
         /* disable flows cache */
@@ -3873,14 +3859,14 @@ int CGlobalTRex::start_master_statefull() {
     m_mg.set_ip( tg->m_client_pool[0].get_ip_start(),
                  tg->m_server_pool[0].get_ip_start(),
                  tg->m_client_pool[0].getDualMask()
-               );
+                 );
 
     if (  CGlobalInfo::m_options.preview.getVMode() >0 ) {
-      m_fl.DumpCsv(stdout);
-      for (i=0; i<100; i++) {
-        fprintf(stdout,"\n");
-      }
-      fflush(stdout);
+        m_fl.DumpCsv(stdout);
+        for (i=0; i<100; i++) {
+            fprintf(stdout,"\n");
+        }
+        fflush(stdout);
     }
 
     m_fl.generate_p_thread_info(get_cores_tx());
@@ -3905,6 +3891,56 @@ int CGlobalTRex::start_master_statefull() {
 
 static CGlobalTRex g_trex;
 
+// The HW counters start from some random values. The driver give us the diffs from previous,
+// each time we do get_rx_stats. We need to make one first call, at system startup,
+// and ignore the returned diffs
+int CPhyEthIF::reset_hw_flow_stats() {
+    uint32_t diff_stats[MAX_FLOW_STATS];
+
+    if (get_ex_drv()->get_rx_stats(this, diff_stats, m_stats.m_fdir_prev_stats, 0, MAX_FLOW_STATS - 1) < 0) {
+        return -1;
+    }
+
+    return 0;
+}
+
+// get/reset flow director counters
+// return 0 if OK. -1 if operation not supported.
+// stats - If not NULL, returning counter numbers in it.
+// index - If non negative, get only counter with this index
+// reset - If true, reset counter value after reading
+int CPhyEthIF::get_flow_stats(uint64_t *rx_stats, tx_per_flow_t *tx_stats, int min, int max, bool reset) {
+    uint32_t diff_stats[MAX_FLOW_STATS];
+
+    if (get_ex_drv()->get_rx_stats(this, diff_stats, m_stats.m_fdir_prev_stats, min, max) < 0) {
+        return -1;
+    }
+
+    for (int i = min; i <= max; i++) {
+        if ( reset ) {
+            // return value so far, and reset
+            if (rx_stats != NULL) {
+                rx_stats[i] = m_stats.m_rx_per_flow[i] + diff_stats[i];
+            }
+            if (tx_stats != NULL) {
+                tx_stats[i] = g_trex.get_flow_tx_stats(m_port_id, i);
+            }
+            m_stats.m_rx_per_flow[i] = 0;
+            g_trex.clear_flow_tx_stats(m_port_id, i);
+        } else {
+            m_stats.m_rx_per_flow[i] += diff_stats[i];
+            if (rx_stats != NULL) {
+                rx_stats[i] = m_stats.m_rx_per_flow[i];
+            }
+            if (tx_stats != NULL) {
+                tx_stats[i] = g_trex.get_flow_tx_stats(m_port_id, i);
+            }
+        }
+    }
+
+    return 0;
+}
+
 bool CCoreEthIF::process_rx_pkt(pkt_dir_t   dir,
                                 rte_mbuf_t * m){
 
@@ -3923,7 +3959,7 @@ bool CCoreEthIF::process_rx_pkt(pkt_dir_t   dir,
             uint8_t max_ttl = 0xff - get_rx_check_hops();
             uint8_t pkt_ttl = parser.getTTl();
             if ( (pkt_ttl==max_ttl) || (pkt_ttl==(max_ttl-1) ) ) {
-               send=true;
+                send=true;
             }
         }
     }
@@ -3942,15 +3978,15 @@ bool CCoreEthIF::process_rx_pkt(pkt_dir_t   dir,
                 send=false;
             }
 
-            #ifdef LATENCY_QUEUE_TRACE_
+#ifdef LATENCY_QUEUE_TRACE_
             printf("rx to cp --\n");
             rte_pktmbuf_dump(stdout,m, rte_pktmbuf_pkt_len(m));
-            #endif
+#endif
         }else{
             send=false;
         }
     }
-   return (send);
+    return (send);
 }
 
 
@@ -3964,8 +4000,8 @@ static int latency_one_lcore(__attribute__((unused)) void *dummy)
     physical_thread_id_t  phy_id =rte_lcore_id();
 
 
-    if ( lpsock->thread_phy_is_latency( phy_id )  ){
-        g_trex.run_in_laterncy_core();
+    if ( lpsock->thread_phy_is_rx(phy_id) ) {
+        g_trex.run_in_rx_core();
     }else{
 
         if ( lpsock->thread_phy_is_master( phy_id ) ) {
@@ -3977,7 +4013,7 @@ static int latency_one_lcore(__attribute__((unused)) void *dummy)
             g_trex.m_signal[ lpsock->thread_phy_to_virt( phy_id ) ]=1;
         }
     }
-	return 0;
+    return 0;
 }
 
 
@@ -3988,8 +4024,8 @@ static int slave_one_lcore(__attribute__((unused)) void *dummy)
     physical_thread_id_t  phy_id =rte_lcore_id();
 
 
-    if ( lpsock->thread_phy_is_latency( phy_id )  ){
-        g_trex.run_in_laterncy_core();
+    if ( lpsock->thread_phy_is_rx(phy_id) ) {
+        g_trex.run_in_rx_core();
     }else{
         if ( lpsock->thread_phy_is_master( phy_id ) ) {
             g_trex.run_in_master();
@@ -3998,7 +4034,7 @@ static int slave_one_lcore(__attribute__((unused)) void *dummy)
             g_trex.run_in_core( lpsock->thread_phy_to_virt( phy_id ) );
         }
     }
-	return 0;
+    return 0;
 }
 
 
@@ -4012,7 +4048,7 @@ uint32_t get_cores_mask(uint32_t cores,int offset){
     for (i=0; i<(cores-1); i++) {
         res |= mask ;
         mask = mask <<1;
-   }
+    }
     return (res);
 }
 
@@ -4031,7 +4067,7 @@ int update_global_info_from_platform_file(){
 
     CGlobalInfo::m_socket.Create(&cg->m_platform);
 
-    
+
     if (!cg->m_info_exist) {
         /* nothing to do ! */
         return 0;
@@ -4058,8 +4094,8 @@ int update_global_info_from_platform_file(){
 
         int port_size=cg->m_mac_info.size();
 
-        if ( port_size > BP_MAX_PORTS ){
-            port_size = BP_MAX_PORTS;
+        if ( port_size > TREX_MAX_PORTS ){
+            port_size = TREX_MAX_PORTS;
         }
         for (i=0; i<port_size; i++){
             cg->m_mac_info[i].copy_src(( char *)CGlobalInfo::m_options.m_mac_addr[i].u.m_mac.src)   ;
@@ -4078,7 +4114,7 @@ int update_global_info_from_platform_file(){
 
     CGlobalInfo::m_memory_cfg.set(cg->m_memory,mul);
     CGlobalInfo::m_memory_cfg.set_number_of_dp_cors(
-    CGlobalInfo::m_options.get_number_of_dp_cores_needed() );
+                                                    CGlobalInfo::m_options.get_number_of_dp_cores_needed() );
     return (0);
 }
 
@@ -4088,21 +4124,21 @@ int core_mask_calc() {
     uint32_t mask = 0;
     int lcore_id;
 
-	for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++) {
-	    if (eal_cpu_detected(lcore_id)) {
-		mask |= (1 << lcore_id);
-	    }
-	}
+    for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++) {
+        if (eal_cpu_detected(lcore_id)) {
+            mask |= (1 << lcore_id);
+        }
+    }
 
-	return mask;
+    return mask;
 }
 
 // Return number of set bits in i
 uint32_t num_set_bits(uint32_t i)
 {
-     i = i - ((i >> 1) & 0x55555555);
-     i = (i & 0x33333333) + ((i >> 2) & 0x33333333);
-     return (((i + (i >> 4)) & 0x0F0F0F0F) * 0x01010101) >> 24;
+    i = i - ((i >> 1) & 0x55555555);
+    i = (i & 0x33333333) + ((i >> 2) & 0x33333333);
+    return (((i + (i >> 4)) & 0x0F0F0F0F) * 0x01010101) >> 24;
 }
 
 // sanity check if the cores we want to use really exist
@@ -4114,17 +4150,17 @@ int core_mask_sanity(uint32_t wanted_core_mask) {
     calc_core_num = num_set_bits(calc_core_mask);
 
     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_latency_thread_enable() ? "+1 latency thread (because of -l flag)\n" : "\n");
-	printf("       Maybe try smaller -c <num>.\n");
-	return -1;
+        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");
+        return -1;
     }
 
     if (wanted_core_mask != (wanted_core_mask & calc_core_mask)) {
-	printf ("Serious error: Something is wrong with the hardware. Wanted core mask is %x. Existing core mask is %x\n", wanted_core_mask, calc_core_mask);
-	return -1;
+        printf ("Serious error: Something is wrong with the hardware. Wanted core mask is %x. Existing core mask is %x\n", wanted_core_mask, calc_core_mask);
+        return -1;
     }
 
     return 0;
@@ -4135,7 +4171,7 @@ int  update_dpdk_args(void){
     CPlatformSocketInfo * lpsock=&CGlobalInfo::m_socket;
     CParserOption * lpop= &CGlobalInfo::m_options;
 
-    lpsock->set_latency_thread_is_enabled(get_is_latency_thread_enable());
+    lpsock->set_rx_thread_is_enabled(get_is_rx_thread_enabled());
     lpsock->set_number_of_threads_per_ports(lpop->preview.getCores() );
     lpsock->set_number_of_dual_ports(lpop->get_expected_dual_ports());
     if ( !lpsock->sanity_check() ){
@@ -4149,7 +4185,7 @@ int  update_dpdk_args(void){
 
     sprintf(global_cores_str,"0x%llx",(unsigned long long)lpsock->get_cores_mask());
     if (core_mask_sanity(strtol(global_cores_str, NULL, 16)) < 0) {
-	return -1;
+        return -1;
     }
 
     /* set the DPDK options */
@@ -4196,7 +4232,7 @@ int  update_dpdk_args(void){
 
     if ( CGlobalInfo::m_options.preview.getVMode() > 0  ) {
         printf("args \n");
-        int i; 
+        int i;
         for (i=0; i<global_dpdk_args_num; i++) {
             printf(" %s \n",global_dpdk_args[i]);
         }
@@ -4254,7 +4290,7 @@ int main_test(int argc , char * argv[]){
     /* It is not a mistake. Give the user higher priorty over the configuration file */
     parse_options(argc, argv, &CGlobalInfo::m_options ,false);
 
-    
+
     if ( CGlobalInfo::m_options.preview.getVMode() > 0){
         CGlobalInfo::m_options.dump(stdout);
         CGlobalInfo::m_memory_cfg.Dump(stdout);
@@ -4272,7 +4308,7 @@ int main_test(int argc , char * argv[]){
         rte_set_log_level(1);
 
     }
-    uid_t uid; 
+    uid_t uid;
     uid = geteuid ();
     if ( uid != 0 ) {
         printf("ERROR you must run with superuser priviliges \n");
@@ -4291,8 +4327,8 @@ int main_test(int argc , char * argv[]){
     }
 
     time_init();
-    
-        /* check if we are in simulation mode */
+
+    /* check if we are in simulation mode */
     if ( CGlobalInfo::m_options.out_file != "" ){
         printf(" t-rex simulation mode into %s \n",CGlobalInfo::m_options.out_file.c_str());
         return ( sim_load_list_of_cap_files(&CGlobalInfo::m_options) );
@@ -4302,17 +4338,17 @@ int main_test(int argc , char * argv[]){
         exit(1);
     }
 
-    if (po->preview.get_is_rx_check_enable() &&  (po->m_rx_check_sampe< get_min_sample_rate()) ) {
-        po->m_rx_check_sampe = get_min_sample_rate();
+    if (po->preview.get_is_rx_check_enable() &&  (po->m_rx_check_sample< get_min_sample_rate()) ) {
+        po->m_rx_check_sample = get_min_sample_rate();
         printf("Warning:rx check sample rate should not be lower than %d. Setting it to %d\n",get_min_sample_rate(),get_min_sample_rate());
     }
 
     /* set dump mode */
     g_trex.m_io_modes.set_mode((CTrexGlobalIoMode::CliDumpMode)CGlobalInfo::m_options.m_io_mode);
 
-    if ( !CGlobalInfo::m_options.is_latency_disabled() 
-         && (CGlobalInfo::m_options.m_latency_prev>0)  ){
-        uint32_t pkts = CGlobalInfo::m_options.m_latency_prev*
+    if ( CGlobalInfo::m_options.is_latency_enabled()
+         && (CGlobalInfo::m_options.m_latency_prev > 0)) {
+        uint32_t pkts = CGlobalInfo::m_options.m_latency_prev *
             CGlobalInfo::m_options.m_latency_rate;
         printf("Start prev latency check- for %d sec \n",CGlobalInfo::m_options.m_latency_prev);
         g_trex.m_mg.start(pkts);
@@ -4368,7 +4404,7 @@ int main_test(int argc , char * argv[]){
 
 //////////////////////////////////////////////////////////////////////////////////////////////
 //////////////////////////////////////////////////////////////////////////////////////////////
-// driver section 
+// driver section
 //////////////////////////////////////////////////////////////////////////////////////////////
 int CTRexExtendedDriverBase::configure_drop_queue(CPhyEthIF * _if) {
     uint8_t port_id=_if->get_rte_port_id();
@@ -4376,16 +4412,16 @@ int CTRexExtendedDriverBase::configure_drop_queue(CPhyEthIF * _if) {
 }
 
 void wait_x_sec(int sec) {
-        int i; 
-        printf(" wait %d sec ", sec);
-        fflush(stdout);
-        for (i=0; i<sec; i++) {
-            delay(1000);
-            printf(".");
-            fflush(stdout);
-        }
-        printf("\n");
+    int i;
+    printf(" wait %d sec ", sec);
+    fflush(stdout);
+    for (i=0; i<sec; i++) {
+        delay(1000);
+        printf(".");
         fflush(stdout);
+    }
+    printf("\n");
+    fflush(stdout);
 }
 
 // in 1G we need to wait if links became ready to soon
@@ -4400,9 +4436,9 @@ int CTRexExtendedDriverBase1G::wait_for_stable_link(){
 
 void CTRexExtendedDriverBase1G::update_configuration(port_cfg_t * cfg){
 
-        cfg->m_tx_conf.tx_thresh.pthresh = TX_PTHRESH_1G;
-        cfg->m_tx_conf.tx_thresh.hthresh = TX_HTHRESH;
-        cfg->m_tx_conf.tx_thresh.wthresh = 0;
+    cfg->m_tx_conf.tx_thresh.pthresh = TX_PTHRESH_1G;
+    cfg->m_tx_conf.tx_thresh.hthresh = TX_HTHRESH;
+    cfg->m_tx_conf.tx_thresh.wthresh = 0;
 }
 
 void CTRexExtendedDriverBase1G::update_global_config_fdir(port_cfg_t * cfg){
@@ -4429,7 +4465,7 @@ int CTRexExtendedDriverBase1G::configure_rx_filter_rules(CPhyEthIF * _if){
 
 int CTRexExtendedDriverBase1G::configure_rx_filter_rules_statefull(CPhyEthIF * _if) {
     uint16_t hops = get_rx_check_hops();
-    uint16_t v4_hops = (hops << 8)&0xff00; 
+    uint16_t v4_hops = (hops << 8)&0xff00;
     uint8_t protocol;
 
     if (CGlobalInfo::m_options.m_l_pkt_mode == 0) {
@@ -4441,82 +4477,82 @@ int CTRexExtendedDriverBase1G::configure_rx_filter_rules_statefull(CPhyEthIF * _
     _if->pci_reg_write( E1000_IMIR(0), 0x00020000);
     _if->pci_reg_write( E1000_IMIREXT(0), 0x00081000);
     _if->pci_reg_write( E1000_TTQF(0),   protocol
-			| 0x00008100 /* enable */
-			| 0xE0010000 /* RX queue is 1 */
-			);
+                        | 0x00008100 /* enable */
+                        | 0xE0010000 /* RX queue is 1 */
+                        );
 
 
-        /* 16  :   12 MAC , (2)0x0800,2      | DW0 , DW1
-                   6 bytes , TTL , PROTO     | DW2=0 , DW3=0x0000FF06
-        */
-        int i;
-        // IPv4: bytes being compared are {TTL, Protocol}
-        uint16_t ff_rules_v4[6]={
-            (uint16_t)(0xFF06 - v4_hops),
-            (uint16_t)(0xFE11 - v4_hops),
-            (uint16_t)(0xFF11 - v4_hops),
-            (uint16_t)(0xFE06 - v4_hops),
-            (uint16_t)(0xFF01 - v4_hops),
-            (uint16_t)(0xFE01 - v4_hops),
-        }  ;
-        // IPv6: bytes being compared are {NextHdr, HopLimit}
-        uint16_t ff_rules_v6[2]={
-            (uint16_t)(0x3CFF - hops),
-            (uint16_t)(0x3CFE - hops),
-        }  ;
-        uint16_t *ff_rules;
-        uint16_t num_rules;
-        uint32_t mask=0;
-        int  rule_id;
+    /* 16  :   12 MAC , (2)0x0800,2      | DW0 , DW1
+       6 bytes , TTL , PROTO     | DW2=0 , DW3=0x0000FF06
+    */
+    int i;
+    // IPv4: bytes being compared are {TTL, Protocol}
+    uint16_t ff_rules_v4[6]={
+        (uint16_t)(0xFF06 - v4_hops),
+        (uint16_t)(0xFE11 - v4_hops),
+        (uint16_t)(0xFF11 - v4_hops),
+        (uint16_t)(0xFE06 - v4_hops),
+        (uint16_t)(0xFF01 - v4_hops),
+        (uint16_t)(0xFE01 - v4_hops),
+    }  ;
+    // IPv6: bytes being compared are {NextHdr, HopLimit}
+    uint16_t ff_rules_v6[2]={
+        (uint16_t)(0x3CFF - hops),
+        (uint16_t)(0x3CFE - hops),
+    }  ;
+    uint16_t *ff_rules;
+    uint16_t num_rules;
+    uint32_t mask=0;
+    int  rule_id;
+
+    if (  CGlobalInfo::m_options.preview.get_ipv6_mode_enable() ){
+        ff_rules = &ff_rules_v6[0];
+        num_rules = sizeof(ff_rules_v6)/sizeof(ff_rules_v6[0]);
+    }else{
+        ff_rules = &ff_rules_v4[0];
+        num_rules = sizeof(ff_rules_v4)/sizeof(ff_rules_v4[0]);
+    }
 
-        if (  CGlobalInfo::m_options.preview.get_ipv6_mode_enable() ){
-            ff_rules = &ff_rules_v6[0];
-            num_rules = sizeof(ff_rules_v6)/sizeof(ff_rules_v6[0]);
-        }else{
-            ff_rules = &ff_rules_v4[0];
-            num_rules = sizeof(ff_rules_v4)/sizeof(ff_rules_v4[0]);
+    uint8_t len = 24;
+    for (rule_id=0; rule_id<num_rules; rule_id++ ) {
+        /* clear rule all */
+        for (i=0; i<0xff; i+=4) {
+            _if->pci_reg_write( (E1000_FHFT(rule_id)+i) , 0);
         }
 
-        uint8_t len = 24;
-        for (rule_id=0; rule_id<num_rules; rule_id++ ) {
-            /* clear rule all */
-            for (i=0; i<0xff; i+=4) {
-                _if->pci_reg_write( (E1000_FHFT(rule_id)+i) , 0);
+        if (  CGlobalInfo::m_options.preview.get_vlan_mode_enable() ){
+            len += 8;
+            if (  CGlobalInfo::m_options.preview.get_ipv6_mode_enable() ){
+                // IPv6 VLAN: NextHdr/HopLimit offset = 0x18
+                _if->pci_reg_write( (E1000_FHFT(rule_id)+(3*16)+0) , PKT_NTOHS(ff_rules[rule_id]) );
+                _if->pci_reg_write( (E1000_FHFT(rule_id)+(3*16)+8) , 0x03); /* MASK */
+            }else{
+                // IPv4 VLAN: TTL/Protocol offset = 0x1A
+                _if->pci_reg_write( (E1000_FHFT(rule_id)+(3*16)+0) , (PKT_NTOHS(ff_rules[rule_id])<<16) );
+                _if->pci_reg_write( (E1000_FHFT(rule_id)+(3*16)+8) , 0x0C); /* MASK */
             }
-
-            if (  CGlobalInfo::m_options.preview.get_vlan_mode_enable() ){
-                len += 8;
-                if (  CGlobalInfo::m_options.preview.get_ipv6_mode_enable() ){
-                    // IPv6 VLAN: NextHdr/HopLimit offset = 0x18
-                    _if->pci_reg_write( (E1000_FHFT(rule_id)+(3*16)+0) , PKT_NTOHS(ff_rules[rule_id]) );
-                    _if->pci_reg_write( (E1000_FHFT(rule_id)+(3*16)+8) , 0x03); /* MASK */
-                }else{
-                    // IPv4 VLAN: TTL/Protocol offset = 0x1A
-                    _if->pci_reg_write( (E1000_FHFT(rule_id)+(3*16)+0) , (PKT_NTOHS(ff_rules[rule_id])<<16) );
-                    _if->pci_reg_write( (E1000_FHFT(rule_id)+(3*16)+8) , 0x0C); /* MASK */
-                }
+        }else{
+            if (  CGlobalInfo::m_options.preview.get_ipv6_mode_enable() ){
+                // IPv6: NextHdr/HopLimit offset = 0x14
+                _if->pci_reg_write( (E1000_FHFT(rule_id)+(2*16)+4) , PKT_NTOHS(ff_rules[rule_id]) );
+                _if->pci_reg_write( (E1000_FHFT(rule_id)+(2*16)+8) , 0x30); /* MASK */
             }else{
-                if (  CGlobalInfo::m_options.preview.get_ipv6_mode_enable() ){
-                    // IPv6: NextHdr/HopLimit offset = 0x14
-                    _if->pci_reg_write( (E1000_FHFT(rule_id)+(2*16)+4) , PKT_NTOHS(ff_rules[rule_id]) );
-                    _if->pci_reg_write( (E1000_FHFT(rule_id)+(2*16)+8) , 0x30); /* MASK */
-                }else{
-                    // IPv4: TTL/Protocol offset = 0x16
-                    _if->pci_reg_write( (E1000_FHFT(rule_id)+(2*16)+4) , (PKT_NTOHS(ff_rules[rule_id])<<16) );
-                    _if->pci_reg_write( (E1000_FHFT(rule_id)+(2*16)+8) , 0xC0); /* MASK */
-                }
+                // IPv4: TTL/Protocol offset = 0x16
+                _if->pci_reg_write( (E1000_FHFT(rule_id)+(2*16)+4) , (PKT_NTOHS(ff_rules[rule_id])<<16) );
+                _if->pci_reg_write( (E1000_FHFT(rule_id)+(2*16)+8) , 0xC0); /* MASK */
             }
+        }
 
-            // FLEX_PRIO[[18:16] = 1, RQUEUE[10:8] = 1
-            _if->pci_reg_write( (E1000_FHFT(rule_id)+0xFC) , (1<<16) | (1<<8)  | len);
+        // FLEX_PRIO[[18:16] = 1, RQUEUE[10:8] = 1
+        _if->pci_reg_write( (E1000_FHFT(rule_id)+0xFC) , (1<<16) | (1<<8)  | len);
 
-            mask |=(1<<rule_id);
-        }
+        mask |=(1<<rule_id);
+    }
 
-        /* enable all rules */
-        _if->pci_reg_write(E1000_WUFC, (mask<<16) | (1<<14) );
+    /* enable all rules */
+    _if->pci_reg_write(E1000_WUFC, (mask<<16) | (1<<14) );
 
-        return (0);
+    return (0);
 }
 
 // Sadly, DPDK has no support for i350 filters, so we need to implement by writing to registers.
@@ -4535,19 +4571,19 @@ int CTRexExtendedDriverBase1G::configure_rx_filter_rules_stateless(CPhyEthIF * _
             _if->pci_reg_write( (E1000_FHFT(rule_id)+i) , 0);
         }
     }
-    
+
     rule_id = 0;
     // filter for byte 18 of packet (lsb of IP ID) should equal ff
     _if->pci_reg_write( (E1000_FHFT(rule_id)+(2*16)) ,  0x00ff0000);
     _if->pci_reg_write( (E1000_FHFT(rule_id)+(2*16) + 8) , 0x04); /* MASK */
     // + bytes 12 + 13 (ether type) should indicate IP.
     _if->pci_reg_write( (E1000_FHFT(rule_id)+(1*16) + 4) ,  0x00000008);
-    _if->pci_reg_write( (E1000_FHFT(rule_id)+(1*16) + 8) , 0x30); /* MASK */    
+    _if->pci_reg_write( (E1000_FHFT(rule_id)+(1*16) + 8) , 0x30); /* MASK */
     // FLEX_PRIO[[18:16] = 1, RQUEUE[10:8] = 1
     _if->pci_reg_write( (E1000_FHFT(rule_id) + 0xFC) , (1 << 16) | (1 << 8) | len);
 
     // same like 0, but with vlan. type should be vlan. Inside vlan, should be IP with lsb of IP ID equals 0xff
-    rule_id = 1; 
+    rule_id = 1;
     // filter for byte 22 of packet (msb of IP ID) should equal ff
     _if->pci_reg_write( (E1000_FHFT(rule_id)+(2*16) + 4) ,  0x00ff0000);
     _if->pci_reg_write( (E1000_FHFT(rule_id)+(2*16) + 8) , 0x40 | 0x03); /* MASK */
@@ -4557,53 +4593,53 @@ int CTRexExtendedDriverBase1G::configure_rx_filter_rules_stateless(CPhyEthIF * _
     // + bytes 16 + 17 (vlan type) should indicate IP.
     _if->pci_reg_write( (E1000_FHFT(rule_id)+(2*16) ) ,  0x00000080);
     // Was written together with IP ID filter
-    // _if->pci_reg_write( (E1000_FHFT(rule_id)+(2*16) + 8) , 0x03); /* MASK */    
+    // _if->pci_reg_write( (E1000_FHFT(rule_id)+(2*16) + 8) , 0x03); /* MASK */
     // FLEX_PRIO[[18:16] = 1, RQUEUE[10:8] = 1
     _if->pci_reg_write( (E1000_FHFT(rule_id) + 0xFC) , (1 << 16) | (1 << 8) | len);
 
     /* enable rules */
     _if->pci_reg_write(E1000_WUFC, (mask << 16) | (1 << 14) );
-    
+
     return (0);
 }
 
 
-void CTRexExtendedDriverBase1G::get_extended_stats(CPhyEthIF * _if,CPhyEthIFStats *stats){ 
+void CTRexExtendedDriverBase1G::get_extended_stats(CPhyEthIF * _if,CPhyEthIFStats *stats){
+
+    stats->ipackets     +=  _if->pci_reg_read(E1000_GPRC) ;
 
-   stats->ipackets     +=  _if->pci_reg_read(E1000_GPRC) ;
+    stats->ibytes       +=  (_if->pci_reg_read(E1000_GORCL) );
+    stats->ibytes       +=  (((uint64_t)_if->pci_reg_read(E1000_GORCH))<<32);
 
-   stats->ibytes       +=  (_if->pci_reg_read(E1000_GORCL) );
-   stats->ibytes       +=  (((uint64_t)_if->pci_reg_read(E1000_GORCH))<<32);
-                        
 
-   stats->opackets     +=  _if->pci_reg_read(E1000_GPTC);
-   stats->obytes       +=  _if->pci_reg_read(E1000_GOTCL) ;
-   stats->obytes       +=  ( (((uint64_t)_if->pci_reg_read(IXGBE_GOTCH))<<32) );
+    stats->opackets     +=  _if->pci_reg_read(E1000_GPTC);
+    stats->obytes       +=  _if->pci_reg_read(E1000_GOTCL) ;
+    stats->obytes       +=  ( (((uint64_t)_if->pci_reg_read(IXGBE_GOTCH))<<32) );
 
-   stats->f_ipackets   +=  0;
-   stats->f_ibytes     += 0;
+    stats->f_ipackets   +=  0;
+    stats->f_ibytes     += 0;
 
 
-   stats->ierrors      +=  ( _if->pci_reg_read(E1000_RNBC) +
-                             _if->pci_reg_read(E1000_CRCERRS) + 
-                             _if->pci_reg_read(E1000_ALGNERRC ) +
-                             _if->pci_reg_read(E1000_SYMERRS ) +
-                             _if->pci_reg_read(E1000_RXERRC ) +
+    stats->ierrors      +=  ( _if->pci_reg_read(E1000_RNBC) +
+                              _if->pci_reg_read(E1000_CRCERRS) +
+                              _if->pci_reg_read(E1000_ALGNERRC ) +
+                              _if->pci_reg_read(E1000_SYMERRS ) +
+                              _if->pci_reg_read(E1000_RXERRC ) +
 
-                             _if->pci_reg_read(E1000_ROC)+
-                             _if->pci_reg_read(E1000_RUC)+ 
-                             _if->pci_reg_read(E1000_RJC) +
+                              _if->pci_reg_read(E1000_ROC)+
+                              _if->pci_reg_read(E1000_RUC)+
+                              _if->pci_reg_read(E1000_RJC) +
 
-                             _if->pci_reg_read(E1000_XONRXC)+   
-                            _if->pci_reg_read(E1000_XONTXC)+
-                            _if->pci_reg_read(E1000_XOFFRXC)+
-                            _if->pci_reg_read(E1000_XOFFTXC)+
-                            _if->pci_reg_read(E1000_FCRUC)
-                             );
+                              _if->pci_reg_read(E1000_XONRXC)+
+                              _if->pci_reg_read(E1000_XONTXC)+
+                              _if->pci_reg_read(E1000_XOFFRXC)+
+                              _if->pci_reg_read(E1000_XOFFTXC)+
+                              _if->pci_reg_read(E1000_FCRUC)
+                              );
 
-   stats->oerrors      +=  0;
-   stats->imcasts      =  0;
-   stats->rx_nombuf    =  0;
+    stats->oerrors      +=  0;
+    stats->imcasts      =  0;
+    stats->rx_nombuf    =  0;
 }
 
 void CTRexExtendedDriverBase1G::clear_extended_stats(CPhyEthIF * _if){
@@ -4626,108 +4662,108 @@ void CTRexExtendedDriverBase10G::update_configuration(port_cfg_t * cfg){
 }
 
 int CTRexExtendedDriverBase10G::configure_rx_filter_rules(CPhyEthIF * _if){
-        uint8_t port_id=_if->get_rte_port_id();
-        uint16_t hops = get_rx_check_hops();
-        uint16_t v4_hops = (hops << 8)&0xff00; 
-
-	/* enable rule 0 SCTP -> queue 1 for latency  */
-	/* 1<<21 means that queue 1 is for SCTP */
-	_if->pci_reg_write(IXGBE_L34T_IMIR(0),(1<<21));	
-	_if->pci_reg_write(IXGBE_FTQF(0),
-			   IXGBE_FTQF_PROTOCOL_SCTP|
-			   (IXGBE_FTQF_PRIORITY_MASK<<IXGBE_FTQF_PRIORITY_SHIFT)|
-			   ((0x0f)<<IXGBE_FTQF_5TUPLE_MASK_SHIFT)|IXGBE_FTQF_QUEUE_ENABLE);
-
-        // IPv4: bytes being compared are {TTL, Protocol}
-        uint16_t ff_rules_v4[6]={
-            (uint16_t)(0xFF11 - v4_hops),
-            (uint16_t)(0xFE11 - v4_hops),
-            (uint16_t)(0xFF06 - v4_hops),
-            (uint16_t)(0xFE06 - v4_hops),
-            (uint16_t)(0xFF01 - v4_hops),
-            (uint16_t)(0xFE01 - v4_hops),
-        };
-        // IPv6: bytes being compared are {NextHdr, HopLimit}
-        uint16_t ff_rules_v6[6]={
-            (uint16_t)(0x3CFF - hops),
-            (uint16_t)(0x3CFE - hops),
-        };        
+    uint8_t port_id=_if->get_rte_port_id();
+    uint16_t hops = get_rx_check_hops();
+    uint16_t v4_hops = (hops << 8)&0xff00;
+
+    /* enable rule 0 SCTP -> queue 1 for latency  */
+    /* 1<<21 means that queue 1 is for SCTP */
+    _if->pci_reg_write(IXGBE_L34T_IMIR(0),(1<<21));
+    _if->pci_reg_write(IXGBE_FTQF(0),
+                       IXGBE_FTQF_PROTOCOL_SCTP|
+                       (IXGBE_FTQF_PRIORITY_MASK<<IXGBE_FTQF_PRIORITY_SHIFT)|
+                       ((0x0f)<<IXGBE_FTQF_5TUPLE_MASK_SHIFT)|IXGBE_FTQF_QUEUE_ENABLE);
+
+    // IPv4: bytes being compared are {TTL, Protocol}
+    uint16_t ff_rules_v4[6]={
+        (uint16_t)(0xFF11 - v4_hops),
+        (uint16_t)(0xFE11 - v4_hops),
+        (uint16_t)(0xFF06 - v4_hops),
+        (uint16_t)(0xFE06 - v4_hops),
+        (uint16_t)(0xFF01 - v4_hops),
+        (uint16_t)(0xFE01 - v4_hops),
+    };
+    // IPv6: bytes being compared are {NextHdr, HopLimit}
+    uint16_t ff_rules_v6[6]={
+        (uint16_t)(0x3CFF - hops),
+        (uint16_t)(0x3CFE - hops),
+    };
 
-        uint16_t *ff_rules;
-        uint16_t num_rules;
-        int  rule_id;
+    uint16_t *ff_rules;
+    uint16_t num_rules;
+    int  rule_id;
 
+    if (  CGlobalInfo::m_options.preview.get_ipv6_mode_enable() ){
+        ff_rules = &ff_rules_v6[0];
+        num_rules = sizeof(ff_rules_v6)/sizeof(ff_rules_v6[0]);
+    }else{
+        ff_rules = &ff_rules_v4[0];
+        num_rules = sizeof(ff_rules_v4)/sizeof(ff_rules_v4[0]);
+    }
+
+    for (rule_id=0; rule_id<num_rules; rule_id++ ) {
+        struct rte_eth_fdir_filter fdir_filter;
+        uint16_t ff_rule = ff_rules[rule_id];
+        int res = 0;
+
+        memset(&fdir_filter,0,sizeof(fdir_filter));
+        /* TOS/PROTO */
         if (  CGlobalInfo::m_options.preview.get_ipv6_mode_enable() ){
-            ff_rules = &ff_rules_v6[0];
-            num_rules = sizeof(ff_rules_v6)/sizeof(ff_rules_v6[0]);
+            fdir_filter.input.flow_type = RTE_ETH_FLOW_NONFRAG_IPV6_OTHER;
         }else{
-            ff_rules = &ff_rules_v4[0];
-            num_rules = sizeof(ff_rules_v4)/sizeof(ff_rules_v4[0]);
+            fdir_filter.input.flow_type = RTE_ETH_FLOW_NONFRAG_IPV4_OTHER;
         }
+        fdir_filter.soft_id = rule_id;
 
-        for (rule_id=0; rule_id<num_rules; rule_id++ ) {
-            struct rte_eth_fdir_filter fdir_filter;
-            uint16_t ff_rule = ff_rules[rule_id];
-            int res = 0;
+        fdir_filter.input.flow_ext.flexbytes[0] = (ff_rule >> 8) & 0xff;
+        fdir_filter.input.flow_ext.flexbytes[1] = ff_rule & 0xff;
+        fdir_filter.action.rx_queue = 1;
+        fdir_filter.action.behavior = RTE_ETH_FDIR_ACCEPT;
+        fdir_filter.action.report_status = RTE_ETH_FDIR_NO_REPORT_STATUS;
+        res = rte_eth_dev_filter_ctrl(port_id, RTE_ETH_FILTER_FDIR, RTE_ETH_FILTER_ADD, &fdir_filter);
 
-            memset(&fdir_filter,0,sizeof(fdir_filter));
-            /* TOS/PROTO */
-            if (  CGlobalInfo::m_options.preview.get_ipv6_mode_enable() ){
-                fdir_filter.input.flow_type = RTE_ETH_FLOW_NONFRAG_IPV6_OTHER;
-            }else{
-                fdir_filter.input.flow_type = RTE_ETH_FLOW_NONFRAG_IPV4_OTHER;
-            }
-            fdir_filter.soft_id = rule_id;
-            
-            fdir_filter.input.flow_ext.flexbytes[0] = (ff_rule >> 8) & 0xff;
-            fdir_filter.input.flow_ext.flexbytes[1] = ff_rule & 0xff;
-            fdir_filter.action.rx_queue = 1;
-            fdir_filter.action.behavior = RTE_ETH_FDIR_ACCEPT;
-            fdir_filter.action.report_status = RTE_ETH_FDIR_NO_REPORT_STATUS;
-            res = rte_eth_dev_filter_ctrl(port_id, RTE_ETH_FILTER_FDIR, RTE_ETH_FILTER_ADD, &fdir_filter);
-
-            if (res != 0) {
-                 rte_exit(EXIT_FAILURE, " ERROR rte_eth_dev_filter_ctrl : %d\n",res);
-            }
+        if (res != 0) {
+            rte_exit(EXIT_FAILURE, " ERROR rte_eth_dev_filter_ctrl : %d\n",res);
         }
-        return (0);
+    }
+    return (0);
 }
 
-void CTRexExtendedDriverBase10G::get_extended_stats(CPhyEthIF * _if,CPhyEthIFStats *stats){ 
+void CTRexExtendedDriverBase10G::get_extended_stats(CPhyEthIF * _if,CPhyEthIFStats *stats){
 
-   int i;
-   uint64_t t=0;
+    int i;
+    uint64_t t=0;
 
-   if ( !get_is_stateless() ) {
+    if ( !get_is_stateless() ) {
 
         for (i=0; i<8;i++) {
             t+=_if->pci_reg_read(IXGBE_MPC(i));
         }
-   }
+    }
+
+    stats->ipackets     +=  _if->pci_reg_read(IXGBE_GPRC) ;
 
-   stats->ipackets     +=  _if->pci_reg_read(IXGBE_GPRC) ;
-   
-   stats->ibytes       +=  (_if->pci_reg_read(IXGBE_GORCL) +(((uint64_t)_if->pci_reg_read(IXGBE_GORCH))<<32));
+    stats->ibytes       +=  (_if->pci_reg_read(IXGBE_GORCL) +(((uint64_t)_if->pci_reg_read(IXGBE_GORCH))<<32));
 
 
 
-   stats->opackets     +=  _if->pci_reg_read(IXGBE_GPTC);
-   stats->obytes       +=  (_if->pci_reg_read(IXGBE_GOTCL) +(((uint64_t)_if->pci_reg_read(IXGBE_GOTCH))<<32));
+    stats->opackets     +=  _if->pci_reg_read(IXGBE_GPTC);
+    stats->obytes       +=  (_if->pci_reg_read(IXGBE_GOTCL) +(((uint64_t)_if->pci_reg_read(IXGBE_GOTCH))<<32));
 
-   stats->f_ipackets   +=  _if->pci_reg_read(IXGBE_RXDGPC);
-   stats->f_ibytes     += (_if->pci_reg_read(IXGBE_RXDGBCL) +(((uint64_t)_if->pci_reg_read(IXGBE_RXDGBCH))<<32));
+    stats->f_ipackets   +=  _if->pci_reg_read(IXGBE_RXDGPC);
+    stats->f_ibytes     += (_if->pci_reg_read(IXGBE_RXDGBCL) +(((uint64_t)_if->pci_reg_read(IXGBE_RXDGBCH))<<32));
 
 
-   stats->ierrors      +=  ( _if->pci_reg_read(IXGBE_RLEC) +
-                             _if->pci_reg_read(IXGBE_ERRBC) +
-                             _if->pci_reg_read(IXGBE_CRCERRS) + 
-                             _if->pci_reg_read(IXGBE_ILLERRC ) +
-                             _if->pci_reg_read(IXGBE_ROC)+
-                             _if->pci_reg_read(IXGBE_RUC)+t);
+    stats->ierrors      +=  ( _if->pci_reg_read(IXGBE_RLEC) +
+                              _if->pci_reg_read(IXGBE_ERRBC) +
+                              _if->pci_reg_read(IXGBE_CRCERRS) +
+                              _if->pci_reg_read(IXGBE_ILLERRC ) +
+                              _if->pci_reg_read(IXGBE_ROC)+
+                              _if->pci_reg_read(IXGBE_RUC)+t);
 
-   stats->oerrors      +=  0;
-   stats->imcasts      =  0;
-   stats->rx_nombuf    =  0;
+    stats->oerrors      +=  0;
+    stats->imcasts      =  0;
+    stats->rx_nombuf    =  0;
 
 }
 
@@ -4764,8 +4800,8 @@ void CTRexExtendedDriverBase40G::add_del_rules(enum rte_filter_op op, uint8_t po
 
     if ( ret != 0 ){
         rte_exit(EXIT_FAILURE, "rte_eth_dev_filter_supported "
-                "err=%d, port=%u \n",
-              ret, port_id);
+                 "err=%d, port=%u \n",
+                 ret, port_id);
     }
 
     struct rte_eth_fdir_filter filter;
@@ -4776,7 +4812,7 @@ void CTRexExtendedDriverBase40G::add_del_rules(enum rte_filter_op op, uint8_t po
     printf("40g::%s rules: port:%d, type:%d ttl:%d, ip_id:%x, q:%d hw index:%d\n", (op == RTE_ETH_FILTER_ADD) ?  "add" : "del"
            , port_id, type, ttl, ip_id, queue, stat_idx);
 #endif
-    
+
     filter.action.rx_queue = queue;
     filter.action.behavior =RTE_ETH_FDIR_ACCEPT;
     filter.action.report_status =RTE_ETH_FDIR_NO_REPORT_STATUS;
@@ -4807,22 +4843,22 @@ void CTRexExtendedDriverBase40G::add_del_rules(enum rte_filter_op op, uint8_t po
     }
 
     ret=rte_eth_dev_filter_ctrl(port_id, RTE_ETH_FILTER_FDIR,
-                op, (void*)&filter);
+                                op, (void*)&filter);
 
     if (  ret !=0 ){
         rte_exit(EXIT_FAILURE, "rte_eth_dev_filter_ctrl"
-                "err=%d, port=%u \n",
-              ret, port_id);
+                 "err=%d, port=%u \n",
+                 ret, port_id);
     }
 }
 
 // type - rule type. Currently we only support rules in IP ID.
 // proto - Packet protocol: UDP or TCP
-// id - Counter id in HW. We assume it is in the range 0..TREX_FDIR_STAT_SIZE
+// id - Counter id in HW. We assume it is in the range 0..MAX_FLOW_STATS
 int CTRexExtendedDriverBase40G::add_del_rx_flow_stat_rule(uint8_t port_id, enum rte_filter_op op, uint8_t type, uint16_t proto, uint16_t id) {
-    uint32_t rule_id = (port_id % m_if_per_card) * TREX_FDIR_STAT_SIZE + id;
+    uint32_t rule_id = (port_id % m_if_per_card) * MAX_FLOW_STATS + id;
     uint16_t rte_type = RTE_ETH_FLOW_NONFRAG_IPV4_OTHER;
-    
+
     switch(proto) {
     case IPPROTO_TCP:
         rte_type = RTE_ETH_FLOW_NONFRAG_IPV4_TCP;
@@ -4833,7 +4869,7 @@ int CTRexExtendedDriverBase40G::add_del_rx_flow_stat_rule(uint8_t port_id, enum
     default:
         rte_type = RTE_ETH_FLOW_NONFRAG_IPV4_OTHER;
         break;
-    }    
+    }
     add_del_rules(op, port_id, rte_type, 0, IP_ID_RESERVE_BASE + id, MAIN_DPDK_DATA_Q, rule_id);
     return 0;
 }
@@ -4842,13 +4878,13 @@ int CTRexExtendedDriverBase40G::configure_rx_filter_rules_statfull(CPhyEthIF * _
     uint32_t port_id = _if->get_port_id();
     uint16_t hops = get_rx_check_hops();
     int i;
-    
+
     for (i = 0; i < 2; i++) {
         uint8_t ttl = TTL_RESERVE_DUPLICATE - i - hops;
         add_del_rules(RTE_ETH_FILTER_ADD, port_id, RTE_ETH_FLOW_NONFRAG_IPV4_UDP, ttl, 0, MAIN_DPDK_RX_Q, 0);
         add_del_rules(RTE_ETH_FILTER_ADD, port_id, RTE_ETH_FLOW_NONFRAG_IPV4_TCP, ttl, 0, MAIN_DPDK_RX_Q, 0);
         add_del_rules(RTE_ETH_FILTER_ADD, port_id, RTE_ETH_FLOW_NONFRAG_IPV6_UDP, ttl, 0, MAIN_DPDK_RX_Q, 0);
-        add_del_rules(RTE_ETH_FILTER_ADD, port_id, RTE_ETH_FLOW_NONFRAG_IPV6_TCP, ttl, 0, MAIN_DPDK_RX_Q, 0);        
+        add_del_rules(RTE_ETH_FILTER_ADD, port_id, RTE_ETH_FLOW_NONFRAG_IPV6_TCP, ttl, 0, MAIN_DPDK_RX_Q, 0);
     }
 
     /* Configure rules for latency measurement packets */
@@ -4868,20 +4904,12 @@ int CTRexExtendedDriverBase40G::configure_rx_filter_rules(CPhyEthIF * _if) {
 
 // instead of adding this to rte_ethdev.h
 extern "C" int rte_eth_fdir_stats_get(uint8_t port_id, uint32_t *stats, uint32_t start, uint32_t len);
-int CTRexExtendedDriverBase40G::get_rx_stats(CPhyEthIF * _if, uint32_t *stats, uint32_t *prev_stats, int index) {
-    uint32_t hw_stats[TREX_FDIR_STAT_SIZE];
+int CTRexExtendedDriverBase40G::get_rx_stats(CPhyEthIF * _if, uint32_t *stats, uint32_t *prev_stats, int min, int max) {
+    uint32_t hw_stats[MAX_FLOW_STATS];
     uint32_t port_id = _if->get_port_id();
-    uint32_t len, start, loop_start;
-
-    if (index >= 0) {
-        len = 1;
-        start = (port_id % m_if_per_card) * TREX_FDIR_STAT_SIZE + index;
-        loop_start = index;
-    } else {
-        start = (port_id % m_if_per_card) * TREX_FDIR_STAT_SIZE;
-        len = TREX_FDIR_STAT_SIZE;
-        loop_start = 0;
-    }
+    uint32_t start = (port_id % m_if_per_card) * MAX_FLOW_STATS + min;
+    uint32_t len = max - min + 1;
+    uint32_t loop_start = min;
 
     rte_eth_fdir_stats_get(port_id, hw_stats, start, len);
     for (int i = loop_start; i <  loop_start + len; i++) {
@@ -4893,7 +4921,7 @@ int CTRexExtendedDriverBase40G::get_rx_stats(CPhyEthIF * _if, uint32_t *stats, u
         }
         prev_stats[i] = hw_stats[i];
     }
-    
+
     return 0;
 }
 
@@ -4904,7 +4932,7 @@ int CTRexExtendedDriverBase40G::dump_fdir_global_stats(CPhyEthIF * _if, FILE *fd
     uint32_t port_id = _if->get_port_id();
     struct rte_eth_fdir_stats stat;
     int ret;
-    
+
     ret = rte_eth_dev_filter_ctrl(port_id, RTE_ETH_FILTER_FDIR, RTE_ETH_FILTER_STATS, (void*)&stat);
     if (ret == 0) {
         if (fd)
@@ -4916,38 +4944,38 @@ int CTRexExtendedDriverBase40G::dump_fdir_global_stats(CPhyEthIF * _if, FILE *fd
         return -1;
     }
 }
-    
-void CTRexExtendedDriverBase40G::get_extended_stats(CPhyEthIF * _if,CPhyEthIFStats *stats){ 
+
+void CTRexExtendedDriverBase40G::get_extended_stats(CPhyEthIF * _if,CPhyEthIFStats *stats){
 
     struct rte_eth_stats stats1;
     rte_eth_stats_get(_if->get_port_id(), &stats1);
 
 
-   stats->ipackets     =  stats1.ipackets;
-   stats->ibytes       =  stats1.ibytes; 
+    stats->ipackets     =  stats1.ipackets;
+    stats->ibytes       =  stats1.ibytes;
 
-   stats->opackets     =  stats1.opackets;
-   stats->obytes       =  stats1.obytes;
+    stats->opackets     =  stats1.opackets;
+    stats->obytes       =  stats1.obytes;
 
-   stats->f_ipackets   = 0;
-   stats->f_ibytes     = 0;
+    stats->f_ipackets   = 0;
+    stats->f_ibytes     = 0;
 
 
-   stats->ierrors      =  stats1.ierrors + stats1.imissed + stats1.ibadcrc +
-                               stats1.ibadlen      +
-                               stats1.ierrors      +
-                               stats1.oerrors      +
-                               stats1.imcasts      +
-                               stats1.rx_nombuf    +
-                               stats1.tx_pause_xon +
-                               stats1.rx_pause_xon +
-                               stats1.tx_pause_xoff+
-                               stats1.rx_pause_xoff ;
+    stats->ierrors      =  stats1.ierrors + stats1.imissed + stats1.ibadcrc +
+        stats1.ibadlen      +
+        stats1.ierrors      +
+        stats1.oerrors      +
+        stats1.imcasts      +
+        stats1.rx_nombuf    +
+        stats1.tx_pause_xon +
+        stats1.rx_pause_xon +
+        stats1.tx_pause_xoff+
+        stats1.rx_pause_xoff ;
 
 
-   stats->oerrors      =  stats1.oerrors;;
-   stats->imcasts      =  0;
-   stats->rx_nombuf    =  stats1.rx_nombuf;
+    stats->oerrors      =  stats1.oerrors;;
+    stats->imcasts      =  0;
+    stats->rx_nombuf    =  stats1.rx_nombuf;
 
 }
 
@@ -4985,37 +5013,37 @@ int CTRexExtendedDriverBase1GVm::configure_drop_queue(CPhyEthIF * _if){
     return (0);
 }
 
-void CTRexExtendedDriverBase1GVm::get_extended_stats(CPhyEthIF * _if,CPhyEthIFStats *stats){ 
+void CTRexExtendedDriverBase1GVm::get_extended_stats(CPhyEthIF * _if,CPhyEthIFStats *stats){
 
     struct rte_eth_stats stats1;
     rte_eth_stats_get(_if->get_port_id(), &stats1);
 
 
-   stats->ipackets     =  stats1.ipackets;
-   stats->ibytes       =  stats1.ibytes; 
+    stats->ipackets     =  stats1.ipackets;
+    stats->ibytes       =  stats1.ibytes;
 
-   stats->opackets     =  stats1.opackets;
-   stats->obytes       =  stats1.obytes;
+    stats->opackets     =  stats1.opackets;
+    stats->obytes       =  stats1.obytes;
 
-   stats->f_ipackets   = 0;
-   stats->f_ibytes     = 0;
+    stats->f_ipackets   = 0;
+    stats->f_ibytes     = 0;
 
 
-   stats->ierrors      =  stats1.ierrors + stats1.imissed + stats1.ibadcrc +
-                               stats1.ibadlen      +
-                               stats1.ierrors      +
-                               stats1.oerrors      +
-                               stats1.imcasts      +
-                               stats1.rx_nombuf    +
-                               stats1.tx_pause_xon +
-                               stats1.rx_pause_xon +
-                               stats1.tx_pause_xoff+
-                               stats1.rx_pause_xoff ;
+    stats->ierrors      =  stats1.ierrors + stats1.imissed + stats1.ibadcrc +
+        stats1.ibadlen      +
+        stats1.ierrors      +
+        stats1.oerrors      +
+        stats1.imcasts      +
+        stats1.rx_nombuf    +
+        stats1.tx_pause_xon +
+        stats1.rx_pause_xon +
+        stats1.tx_pause_xoff+
+        stats1.rx_pause_xoff ;
 
 
-   stats->oerrors      =  stats1.oerrors;;
-   stats->imcasts      =  0;
-   stats->rx_nombuf    =  stats1.rx_nombuf;
+    stats->oerrors      =  stats1.oerrors;;
+    stats->imcasts      =  0;
+    stats->rx_nombuf    =  stats1.rx_nombuf;
 
 }
 
@@ -5028,13 +5056,13 @@ int CTRexExtendedDriverBase1GVm::wait_for_stable_link(){
 
 /**
  * convert chain of mbuf to one big mbuf
- * 
+ *
  * @param m
- * 
- * @return 
+ *
+ * @return
  */
 struct rte_mbuf *  rte_mbuf_convert_to_one_seg(struct rte_mbuf *m){
-	unsigned int len;
+    unsigned int len;
     struct rte_mbuf * r;
     struct rte_mbuf * old_m;
     old_m=m;
@@ -5049,21 +5077,21 @@ struct rte_mbuf *  rte_mbuf_convert_to_one_seg(struct rte_mbuf *m){
     }
     char *p=rte_pktmbuf_append(r,len);
 
-	while ( m ) {
+    while ( m ) {
         len = m->data_len;
         assert(len);
         memcpy(p,(char *)m->buf_addr, len);
         p+=len;
-		m = m->next;
-	}
+        m = m->next;
+    }
     rte_pktmbuf_free(old_m);
     return(r);
 }
 
 /***********************************************************
- * platfrom API object 
- * TODO: REMOVE THIS TO A SEPERATE FILE 
- * 
+ * platfrom API object
+ * TODO: REMOVE THIS TO A SEPERATE FILE
+ *
  **********************************************************/
 void TrexDpdkPlatformApi::get_port_num(uint8_t &port_num) const {
     port_num = g_trex.m_max_ports;
@@ -5087,12 +5115,12 @@ TrexDpdkPlatformApi::get_global_stats(TrexPlatformGlobalStats &stats) const {
     stats.m_stats.m_total_rx_bytes     = trex_stats.m_total_rx_bytes;
 }
 
-void 
+void
 TrexDpdkPlatformApi::get_interface_stats(uint8_t interface_id, TrexPlatformInterfaceStats &stats) const {
 
 }
 
-uint8_t 
+uint8_t
 TrexDpdkPlatformApi::get_dp_core_count() const {
     return CGlobalInfo::m_options.preview.getCores();
 }
@@ -5128,9 +5156,12 @@ TrexDpdkPlatformApi::get_interface_info(uint8_t interface_id, intf_info_st &info
     /* hardware */
     g_trex.m_ports[interface_id].macaddr_get(&rte_mac_addr);
     assert(ETHER_ADDR_LEN == 6);
+    printf("interface %d speed: %d mac:", interface_id, info.speed);
     for (int i = 0; i < 6; i++) {
         info.mac_info.hw_macaddr[i] = rte_mac_addr.addr_bytes[i];
+        printf("%x:", rte_mac_addr.addr_bytes[i]);
     }
+    printf("\n");
 
     /* software */
     uint8_t sw_macaddr[12];
@@ -5139,7 +5170,7 @@ TrexDpdkPlatformApi::get_interface_info(uint8_t interface_id, intf_info_st &info
     for (int i = 0; i < 6; i++) {
         info.mac_info.dst_macaddr[i] = sw_macaddr[i];
         info.mac_info.src_macaddr[i] = sw_macaddr[6 + i];
-        
+
     }
 
     info.numa_node =  g_trex.m_ports[interface_id].m_dev_info.pci_dev->numa_node;
@@ -5153,7 +5184,7 @@ TrexDpdkPlatformApi::get_interface_info(uint8_t interface_id, intf_info_st &info
 
 void
 TrexDpdkPlatformApi::publish_async_data_now(uint32_t key) const {
-    g_trex.publish_async_data();
+    g_trex.publish_async_data(true);
     g_trex.publish_async_barrier(key);
 }
 
@@ -5163,8 +5194,12 @@ TrexDpdkPlatformApi::get_interface_stat_info(uint8_t interface_id, uint16_t &num
     capabilities = CTRexExtendedDriverDb::Ins()->get_drv()->get_rx_stat_capabilities();
 }
 
-int TrexDpdkPlatformApi::get_rx_stats(uint8 port_id, uint64_t *stats, int index, bool reset) const {
-    return g_trex.m_ports[port_id].get_rx_stats(stats, index, reset);
+int TrexDpdkPlatformApi::get_flow_stats(uint8 port_id, uint64_t *rx_stats, void *tx_stats, int min, int max, bool reset) const {
+    return g_trex.m_ports[port_id].get_flow_stats(rx_stats, (tx_per_flow_t *)tx_stats, min, max, reset);
+}
+
+int TrexDpdkPlatformApi::reset_hw_flow_stats(uint8_t port_id) const {
+    return g_trex.m_ports[port_id].reset_hw_flow_stats();
 }
 
 int TrexDpdkPlatformApi::add_rx_flow_stat_rule(uint8_t port_id, uint8_t type, uint16_t proto, uint16_t id) const {
@@ -5185,4 +5220,6 @@ bool TrexDpdkPlatformApi::get_promiscuous(uint8_t port_id) const {
     return g_trex.m_ports[port_id].get_promiscuous();
 }
 
-
+void TrexDpdkPlatformApi::flush_dp_messages() const {
+    g_trex.check_for_dp_messages();
+}
diff --git a/src/main_dpdk.h b/src/main_dpdk.h
index 7357c0f4..33615636 100644
--- a/src/main_dpdk.h
+++ b/src/main_dpdk.h
@@ -38,10 +38,9 @@ class CPhyEthIFStats {
     uint64_t oerrors;   /**< Total number of failed transmitted packets. */
     uint64_t imcasts;   /**< Total number of multicast received packets. */
     uint64_t rx_nombuf; /**< Total number of RX mbuf allocation failures. */
-    uint64_t  m_rx_per_flow [TREX_FDIR_STAT_SIZE]; // Per flow RX statistics
+    uint64_t  m_rx_per_flow [MAX_FLOW_STATS]; // Per flow RX statistics
     // Previous fdir stats values read from HW. Since on xl710 this is 32 bit, we save old value, to handle wrap around.
-    uint32_t  m_fdir_prev_stats [TREX_FDIR_STAT_SIZE];
-    bool m_fdir_stats_first_time;
+    uint32_t  m_fdir_prev_stats [MAX_FLOW_STATS];
  public:
     void Clear();
     void Dump(FILE *fd);
@@ -53,7 +52,6 @@ class CPhyEthIF  {
     CPhyEthIF (){
         m_port_id=0;
         m_rx_queue=0;
-        m_stats.m_fdir_stats_first_time = true;
     }
     bool Create(uint8_t portid){
         m_port_id      = portid;
@@ -74,7 +72,8 @@ class CPhyEthIF  {
     void macaddr_get(struct ether_addr *mac_addr);
     void get_stats(CPhyEthIFStats *stats);
     int dump_fdir_global_stats(FILE *fd);
-    int get_rx_stats(uint64_t *stats, int index, bool reset);
+    int reset_hw_flow_stats();
+    int get_flow_stats(uint64_t *rx_stats, tx_per_flow_t *tx_stats, int min, int max, bool reset);
     void get_stats_1g(CPhyEthIFStats *stats);
     void rx_queue_setup(uint16_t rx_queue_id,
                         uint16_t nb_rx_desc,
@@ -116,6 +115,7 @@ class CPhyEthIF  {
     float get_last_rx_pps_rate(){
         return (m_last_rx_pps);
     }
+
     CPhyEthIFStats     & get_stats(){
         return ( m_stats );
     }
diff --git a/src/platform_cfg.cpp b/src/platform_cfg.cpp
index 15834544..42b9a990 100755
--- a/src/platform_cfg.cpp
+++ b/src/platform_cfg.cpp
@@ -99,8 +99,8 @@ void CPlatformCoresYamlInfo::Dump(FILE *fd){
        fprintf(fd," no platform info \n");
        return;
     }
-    fprintf(fd," master   thread  : %d  \n",m_master_thread);
-    fprintf(fd," latency  thread  : %d  \n",m_latency_thread);
+    fprintf(fd," master   thread  : %d  \n", m_master_thread);
+    fprintf(fd," rx  thread  : %d  \n", m_rx_thread);
     int i;
     for (i=0; i<m_dual_if.size(); i++) {
         printf(" dual_if : %d \n",i);
@@ -124,7 +124,7 @@ void operator >> (const YAML::Node& node, CPlatformDualIfYamlInfo & plat_info) {
 
 void operator >> (const YAML::Node& node, CPlatformCoresYamlInfo & plat_info) {
      node["master_thread_id"] >> plat_info.m_master_thread;
-     node["latency_thread_id"] >> plat_info.m_latency_thread;    
+     node["rx_thread_id"] >> plat_info.m_rx_thread;
 
      const YAML::Node& dual_info = node["dual_if"];
      for(unsigned i=0;i<dual_info.size();i++) {
diff --git a/src/platform_cfg.h b/src/platform_cfg.h
index e8f93d0b..682f33e8 100755
--- a/src/platform_cfg.h
+++ b/src/platform_cfg.h
@@ -141,7 +141,7 @@ public:
     }
     bool             m_is_exists;
     uint32_t         m_master_thread; 
-    uint32_t         m_latency_thread;  
+    uint32_t         m_rx_thread;  
     std::vector <CPlatformDualIfYamlInfo> m_dual_if;
 public:
     void Dump(FILE *fd);
diff --git a/src/rpc-server/trex_rpc_req_resp_server.cpp b/src/rpc-server/trex_rpc_req_resp_server.cpp
index da7e8c55..5c587e0f 100644
--- a/src/rpc-server/trex_rpc_req_resp_server.cpp
+++ b/src/rpc-server/trex_rpc_req_resp_server.cpp
@@ -173,10 +173,8 @@ void TrexRpcServerReqRes::process_request_raw(const std::string &request, std::s
 
     int index = 0;
 
-    /* if lock was provided, take it  */
-    if (m_lock) {
-        m_lock->lock();
-    }
+    /* expcetion safe */
+    std::unique_lock<std::mutex> lock(*m_lock);
 
     /* for every command parsed - launch it */
     for (auto command : commands) {
@@ -190,9 +188,7 @@ void TrexRpcServerReqRes::process_request_raw(const std::string &request, std::s
     }
 
     /* done with the lock */
-    if (m_lock) {
-        m_lock->unlock();
-    }
+    lock.unlock();
 
     /* write the JSON to string and sever on ZMQ */
 
@@ -254,28 +250,3 @@ TrexRpcServerReqRes::test_inject_request(const std::string &req) {
     return response;
 }
 
-
-/**
- * MOCK req resp server
- */
-TrexRpcServerReqResMock::TrexRpcServerReqResMock(const TrexRpcServerConfig &cfg) : TrexRpcServerReqRes(cfg) {
-}
-
-/**
- * override start
- * 
- */
-void
-TrexRpcServerReqResMock::start() {
-
-}
-
-
-/**
- * override stop
- */
-void
-TrexRpcServerReqResMock::stop() {
-
-}
-
diff --git a/src/rpc-server/trex_rpc_req_resp_server.h b/src/rpc-server/trex_rpc_req_resp_server.h
index 979bf9af..26b3248f 100644
--- a/src/rpc-server/trex_rpc_req_resp_server.h
+++ b/src/rpc-server/trex_rpc_req_resp_server.h
@@ -55,7 +55,6 @@ protected:
     void               *m_socket;
 };
 
-
 /**
  * a mock req resp server (for tests)
  * 
@@ -73,5 +72,6 @@ public:
 
 };
 
+
 #endif /* __TREX_RPC_REQ_RESP_API_H__ */
 
diff --git a/src/rpc-server/trex_rpc_server.cpp b/src/rpc-server/trex_rpc_server.cpp
index 1dfc4494..7d2e31a5 100644
--- a/src/rpc-server/trex_rpc_server.cpp
+++ b/src/rpc-server/trex_rpc_server.cpp
@@ -33,6 +33,9 @@ limitations under the License.
 TrexRpcServerInterface::TrexRpcServerInterface(const TrexRpcServerConfig &cfg, const std::string &name, std::mutex *lock) : m_cfg(cfg), m_name(name), m_lock(lock)  {
     m_is_running = false;
     m_is_verbose = false;
+    if (m_lock == NULL) {
+        m_lock = &m_dummy_lock;
+    }
 }
 
 TrexRpcServerInterface::~TrexRpcServerInterface() {
@@ -117,7 +120,6 @@ get_current_date_time() {
 const std::string TrexRpcServer::s_server_uptime = get_current_date_time();
 
 TrexRpcServer::TrexRpcServer(const TrexRpcServerConfig *req_resp_cfg,
-                             const TrexRpcServerConfig *async_cfg,
                              std::mutex *lock) {
 
     m_req_resp = NULL;
@@ -134,10 +136,6 @@ TrexRpcServer::TrexRpcServer(const TrexRpcServerConfig *req_resp_cfg,
         m_servers.push_back(m_req_resp);
     }
     
-    /* add async publisher */
-    if (async_cfg) {
-        m_servers.push_back(new TrexRpcServerAsync(*async_cfg, lock));
-    }
 }
 
 TrexRpcServer::~TrexRpcServer() {
@@ -187,3 +185,27 @@ TrexRpcServer::test_inject_request(const std::string &req_str) {
         return "";
     }
 }
+
+/**
+ * MOCK req resp server
+ */
+TrexRpcServerReqResMock::TrexRpcServerReqResMock(const TrexRpcServerConfig &cfg) : TrexRpcServerReqRes(cfg) {
+}
+
+/**
+ * override start
+ * 
+ */
+void
+TrexRpcServerReqResMock::start() {
+
+}
+
+
+/**
+ * override stop
+ */
+void
+TrexRpcServerReqResMock::stop() {
+
+}
diff --git a/src/rpc-server/trex_rpc_server_api.h b/src/rpc-server/trex_rpc_server_api.h
index 1ab5dce9..a02b2cc0 100644
--- a/src/rpc-server/trex_rpc_server_api.h
+++ b/src/rpc-server/trex_rpc_server_api.h
@@ -133,6 +133,7 @@ protected:
     std::thread                          *m_thread;
     std::string                          m_name;
     std::mutex                           *m_lock;
+    std::mutex                           m_dummy_lock;
 };
 
 /**
@@ -147,7 +148,6 @@ public:
   
     /* creates the collection of servers using configurations */
     TrexRpcServer(const TrexRpcServerConfig *req_resp_cfg,
-                  const TrexRpcServerConfig *async_cfg,
                   std::mutex *m_lock = NULL);
 
     ~TrexRpcServer();
diff --git a/src/rx_check.cpp b/src/rx_check.cpp
index 7e81ef28..bfaa4ddb 100755
--- a/src/rx_check.cpp
+++ b/src/rx_check.cpp
@@ -388,7 +388,7 @@ void RxCheckManager::handle_packet(CRx_check_header * rxh){
     m_tw.restart_timer(&lf->m_aging_timer_handle,m_cur_time+std::max(rxh->m_aging_sec,(uint16_t)5));
     /* teminate flow if needed */
     if ( lf->is_all_pkts_seen() ){
-            /* handel from termination */
+            /* handle from termination */
             m_tw.stop_timer(&lf->m_aging_timer_handle);
             lf->set_aged_correctly();
             on_flow_end(lf);
diff --git a/src/sim/trex_sim.h b/src/sim/trex_sim.h
index 3a3a62ea..59184b75 100644
--- a/src/sim/trex_sim.h
+++ b/src/sim/trex_sim.h
@@ -77,7 +77,6 @@ public:
     
         cfg.m_port_count         = 2;
         cfg.m_rpc_req_resp_cfg   = NULL;
-        cfg.m_rpc_async_cfg      = NULL;
         cfg.m_rpc_server_verbose = false;
         cfg.m_platform_api       = new SimPlatformApi(1);
         cfg.m_publisher          = NULL;
diff --git a/src/sim/trex_sim_stateless.cpp b/src/sim/trex_sim_stateless.cpp
index 30d60b15..87c61ae2 100644
--- a/src/sim/trex_sim_stateless.cpp
+++ b/src/sim/trex_sim_stateless.cpp
@@ -186,7 +186,6 @@ SimStateless::prepare_control_plane() {
 
     cfg.m_port_count         = m_port_count;
     cfg.m_rpc_req_resp_cfg   = &rpc_req_resp_cfg;
-    cfg.m_rpc_async_cfg      = NULL;
     cfg.m_rpc_server_verbose = false;
     cfg.m_platform_api       = new SimPlatformApi(m_dp_core_count);
     cfg.m_publisher          = m_publisher;
diff --git a/src/stateless/cp/trex_dp_port_events.cpp b/src/stateless/cp/trex_dp_port_events.cpp
index ba327e59..8e098adf 100644
--- a/src/stateless/cp/trex_dp_port_events.cpp
+++ b/src/stateless/cp/trex_dp_port_events.cpp
@@ -20,6 +20,7 @@ limitations under the License.
 */
 
 #include <trex_dp_port_events.h>
+#include <trex_stateless_messaging.h>
 #include <sstream>
 #include <os_time.h>
 #include <trex_stateless.h>
@@ -27,24 +28,20 @@ limitations under the License.
 /**
  * port events
  */
-void 
-TrexDpPortEvents::create(TrexStatelessPort *port) {
+TrexDpPortEvents::TrexDpPortEvents(TrexStatelessPort *port) {
     m_port = port;
-
-    for (int i = 0; i < TrexDpPortEvent::EVENT_MAX; i++) {
-        m_events[i].create((TrexDpPortEvent::event_e) i, port);
-    }
-
     m_event_id_counter = EVENT_ID_INVALID;
 }
 
-/**
- * generate a new event ID
- * 
- */
-int
-TrexDpPortEvents::generate_event_id() {
-    return (++m_event_id_counter);
+TrexDpPortEvent *
+TrexDpPortEvents::lookup(int event_id) {
+    auto search = m_events.find(event_id);
+
+    if (search != m_events.end()) {
+        return search->second;
+    } else {
+        return NULL;
+    }
 }
 
 /**
@@ -52,21 +49,49 @@ TrexDpPortEvents::generate_event_id() {
  * all other events will be disabled 
  * 
  */
-void
-TrexDpPortEvents::wait_for_event(TrexDpPortEvent::event_e ev, int event_id, int timeout_ms) {
+int
+TrexDpPortEvents::create_event(TrexDpPortEvent *event, int timeout_ms) {
+    /* allocate ID for event */
+    int event_id = ++m_event_id_counter;
 
-    /* first disable all events */
-    for (TrexDpPortEvent & e : m_events) {
-        e.disable();
-    }
+    /* init and add */
+    event->init(m_port, event_id, timeout_ms);
+    m_events[event_id] = event;
 
-    /* mark this event as allowed */
-    m_events[ev].wait_for_event(event_id, timeout_ms);
+    return event_id;
 }
 
 void 
-TrexDpPortEvents::disable(TrexDpPortEvent::event_e ev) {
-    m_events[ev].disable();
+TrexDpPortEvents::destroy_event(int event_id) {
+    TrexDpPortEvent *event = lookup(event_id);
+    if (!event) {
+        /* cannot find event */
+        throw TrexException("internal error - cannot find event");
+    }
+
+    m_events.erase(event_id);
+    delete event;
+}
+
+class DPBarrier : public TrexDpPortEvent {
+protected:
+    virtual void on_event() {
+        /* do nothing */
+    }
+};
+
+void
+TrexDpPortEvents::barrier() {
+    int barrier_id = create_event(new DPBarrier());
+
+    TrexStatelessCpToDpMsgBase *barrier_msg = new TrexStatelessDpBarrier(m_port->m_port_id, barrier_id);
+    m_port->send_message_to_all_dp(barrier_msg);
+
+    get_stateless_obj()->get_platform_api()->flush_dp_messages();
+    while (lookup(barrier_id) != NULL) {
+        delay(1);
+        get_stateless_obj()->get_platform_api()->flush_dp_messages();
+    }
 }
 
 /**
@@ -74,39 +99,33 @@ TrexDpPortEvents::disable(TrexDpPortEvent::event_e ev) {
  * 
  */
 void 
-TrexDpPortEvents::handle_event(TrexDpPortEvent::event_e ev, int thread_id, int event_id) {
-    m_events[ev].handle_event(thread_id, event_id);
-}
-
-/*********** 
- * single event object
- * 
- */
+TrexDpPortEvents::on_core_reporting_in(int event_id, int thread_id) {
+    TrexDpPortEvent *event = lookup(event_id);
+    /* event might have been deleted */
+    if (!event) {
+        return;
+    }
 
-void
-TrexDpPortEvent::create(event_e type, TrexStatelessPort *port) {
-    m_event_type = type;
-    m_port = port;
+    bool done = event->on_core_reporting_in(thread_id);
 
-    /* add the core ids to the hash */
-    m_signal.clear();
-    for (int core_id : m_port->get_core_id_list()) {
-        m_signal[core_id] = false;
+    if (done) {
+        destroy_event(event_id);
     }
-
-    /* event is disabled */
-    disable();
 }
 
 
-/**
- * wait the event using event id and timeout
+/***************************
+ * event
  * 
- */
-void
-TrexDpPortEvent::wait_for_event(int event_id, int timeout_ms) {
+ **************************/
+TrexDpPortEvent::TrexDpPortEvent() {
+    m_port = NULL;
+    m_event_id = -1;
+}
 
-    /* set a new event id */
+void 
+TrexDpPortEvent::init(TrexStatelessPort *port, int event_id, int timeout_ms) {
+    m_port = port;
     m_event_id = event_id;
 
     /* do we have a timeout ? */
@@ -118,103 +137,33 @@ TrexDpPortEvent::wait_for_event(int event_id, int timeout_ms) {
 
     /* prepare the signal array */
     m_pending_cnt = 0;
-    for (auto & core_pair : m_signal) {
-        core_pair.second = false;
+    for (int core_id : m_port->get_core_id_list()) {
+        m_signal[core_id] = false;
         m_pending_cnt++;
     }
 }
 
-void
-TrexDpPortEvent::disable() {
-    m_event_id = TrexDpPortEvents::EVENT_ID_INVALID;
-}
-
-/**
- * get the event status
- * 
- */
-
-TrexDpPortEvent::event_status_e
-TrexDpPortEvent::status() {
-
-    /* is it even active ? */
-    if (m_event_id == TrexDpPortEvents::EVENT_ID_INVALID) {
-        return (EVENT_DISABLE);
-    }
-
-    /* did it occured ? */
-    if (m_pending_cnt == 0) {
-        return (EVENT_OCCURED);
-    }
-
-    /* so we are enabled and the event did not occur - maybe we timed out ? */
-    if ( (m_expire_limit_ms > 0) && (os_get_time_msec() > m_expire_limit_ms) ) {
-        return (EVENT_TIMED_OUT);
-    }
-
-    /* so we are still waiting... */
-    return (EVENT_PENDING);
-
-}
-
-void
-TrexDpPortEvent::err(int thread_id, int event_id, const std::string &err_msg)  {
-    std::stringstream err;
-    err << "DP event '" << event_name(m_event_type)  << "' on thread id '" << thread_id << "' with key '" << event_id <<"' - ";
-}
-
-/**
- * event occured
- * 
- */
-void 
-TrexDpPortEvent::handle_event(int thread_id, int event_id) {
-
-    /* if the event is disabled - we don't care */
-    if (!is_active()) {
-        return;
-    }
-
-    /* check the event id is matching the required event - if not maybe its an old signal */
-    if (event_id != m_event_id) {
-        return;
-    }
-
+bool
+TrexDpPortEvent::on_core_reporting_in(int thread_id) {
     /* mark sure no double signal */
     if (m_signal.at(thread_id)) {
-        err(thread_id, event_id, "double signal");
+        std::stringstream err;
+        err << "double signal detected on event id: " << m_event_id;
+        throw TrexException(err.str());
 
-    } else {
-        /* mark */
-        m_signal.at(thread_id) = true;
-        m_pending_cnt--;
     }
 
+    /* mark */
+    m_signal.at(thread_id) = true;
+    m_pending_cnt--;
+
     /* event occured */
     if (m_pending_cnt == 0) {
-        m_port->on_dp_event_occured(m_event_type);
-        m_event_id = TrexDpPortEvents::EVENT_ID_INVALID;
+        on_event();
+        return true;
+    } else {
+        return false;
     }
 }
 
-bool
-TrexDpPortEvent::is_active() {
-    return (status() != EVENT_DISABLE);
-}
-
-bool 
-TrexDpPortEvent::has_timeout_expired() {
-    return (status() == EVENT_TIMED_OUT);
-}
-
-const char *
-TrexDpPortEvent::event_name(event_e type) {
-    switch (type) {
-    case EVENT_STOP:
-        return "DP STOP";
 
-    default:
-        throw TrexException("unknown event type");
-    }
-
-}
diff --git a/src/stateless/cp/trex_dp_port_events.h b/src/stateless/cp/trex_dp_port_events.h
index 557e590b..3b8c8633 100644
--- a/src/stateless/cp/trex_dp_port_events.h
+++ b/src/stateless/cp/trex_dp_port_events.h
@@ -25,95 +25,43 @@ limitations under the License.
 #include <string>
 
 class TrexStatelessPort;
+class TrexDpPortEvents;
 
 /**
- * describes a single DP event related to port
+ * interface class for DP events
  * 
- * @author imarom (18-Nov-15)
+ * @author imarom (29-Feb-16)
  */
 class TrexDpPortEvent {
-public:
-
-    enum event_e {
-        EVENT_STOP   = 1,
-        EVENT_MAX
-    };
-
-    /**
-     * status of the event for the port
-     */
-    enum event_status_e {
-        EVENT_DISABLE,
-        EVENT_PENDING,
-        EVENT_TIMED_OUT,
-        EVENT_OCCURED
-    };
-
-    /**
-     * init for the event
-     * 
-     */
-    void create(event_e type, TrexStatelessPort *port);
-
-    /**
-     * create a new pending event
-     * 
-     */
-    void wait_for_event(int event_id, int timeout_ms = -1);
-
-    /**
-     * mark event as not allowed to happen
-     * 
-     */
-    void disable();
-
-    /**
-     * get the event status
-     * 
-     */
-    event_status_e status();
-
-    /**
-     * event occured
-     * 
-     */
-    void handle_event(int thread_id, int event_id);
+    friend TrexDpPortEvents;
 
-    /**
-     * returns true if event is active
-     * 
-     */
-    bool is_active();
-
-    /**
-     * has timeout already expired ?
-     * 
-     */
-    bool has_timeout_expired();
-
-    /**
-     * generate error
-     * 
-     */
-    void err(int thread_id, int event_id, const std::string &err_msg);
+public:
+    TrexDpPortEvent();
+    virtual ~TrexDpPortEvent() {}
 
+protected:
     /**
-     * event to name
+     * what to do when an event has been completed (all cores 
+     * reported in 
      * 
+     * @author imarom (29-Feb-16)
      */
-    static const char * event_name(event_e type);
+    virtual void on_event() = 0;
 
+    TrexStatelessPort *get_port() {
+        return m_port;
+    }
 
 private:
+    void init(TrexStatelessPort *port, int event_id, int timeout_ms);
+    bool on_core_reporting_in(int thread_id);
 
-    event_e                        m_event_type;
     std::unordered_map<int, bool>  m_signal;
     int                            m_pending_cnt;
 
     TrexStatelessPort             *m_port;
     int                            m_event_id;
     int                            m_expire_limit_ms;
-    
 };
 
 /**
@@ -124,44 +72,39 @@ class TrexDpPortEvents {
 public:
     friend class TrexDpPortEvent;
 
-    void create(TrexStatelessPort *port);
+    static const int INVALID_ID = -1;
 
-    /**
-     * generate a new event ID to be used with wait_for_event
-     * 
-     */
-    int generate_event_id();
+    TrexDpPortEvents(TrexStatelessPort *port);
 
     /**
      * wait a new DP event on the port 
      * returns a key which will be used to identify 
      * the event happened 
      * 
-     * @author imarom (18-Nov-15)
-     * 
-     * @param ev - type of event
-     * @param event_id - a unique identifier for the event
-     * @param timeout_ms - does it has a timeout ?
-     * 
      */
-    void wait_for_event(TrexDpPortEvent::event_e ev, int event_id, int timeout_ms = -1);
+    int create_event(TrexDpPortEvent *event, int timeout_ms = -1);
 
     /**
-     * disable an event (don't care)
+     * destroy an event
      * 
      */
-    void disable(TrexDpPortEvent::event_e ev);
+    void destroy_event(int event_id);
 
     /**
-     * event has occured
-     * 
+     * return when all DP cores have responsed on a barrier
+     */
+    void barrier();
+
+    /**
+     * a core has reached the event 
      */
-    void handle_event(TrexDpPortEvent::event_e ev, int thread_id, int event_id);
+    void on_core_reporting_in(int event_id, int thread_id);
 
 private:
-    static const int EVENT_ID_INVALID = -1;
+    TrexDpPortEvent *lookup(int event_id);
 
-    TrexDpPortEvent m_events[TrexDpPortEvent::EVENT_MAX];
+    static const int EVENT_ID_INVALID = -1;
+    std::unordered_map<int, TrexDpPortEvent *> m_events;
     int m_event_id_counter;
 
     TrexStatelessPort *m_port;
diff --git a/src/stateless/cp/trex_stateless.cpp b/src/stateless/cp/trex_stateless.cpp
index a4522837..9e24802b 100644
--- a/src/stateless/cp/trex_stateless.cpp
+++ b/src/stateless/cp/trex_stateless.cpp
@@ -40,7 +40,7 @@ TrexStateless::TrexStateless(const TrexStatelessCfg &cfg) {
     /* create RPC servers */
 
     /* set both servers to mutex each other */
-    m_rpc_server = new TrexRpcServer(cfg.m_rpc_req_resp_cfg, cfg.m_rpc_async_cfg, &m_global_cp_lock);
+    m_rpc_server = new TrexRpcServer(cfg.m_rpc_req_resp_cfg, cfg.m_global_lock);
     m_rpc_server->set_verbose(cfg.m_rpc_server_verbose);
 
     /* configure ports */
diff --git a/src/stateless/cp/trex_stateless.h b/src/stateless/cp/trex_stateless.h
index cc47da6b..6e5e0c44 100644
--- a/src/stateless/cp/trex_stateless.h
+++ b/src/stateless/cp/trex_stateless.h
@@ -92,18 +92,19 @@ public:
     TrexStatelessCfg() {
         m_port_count          = 0;
         m_rpc_req_resp_cfg    = NULL;
-        m_rpc_async_cfg       = NULL;
         m_rpc_server_verbose  = false;
         m_platform_api        = NULL;
         m_publisher           = NULL;
+        m_global_lock         = NULL;
     }
 
     const TrexRpcServerConfig  *m_rpc_req_resp_cfg;
-    const TrexRpcServerConfig  *m_rpc_async_cfg;
     const TrexPlatformApi      *m_platform_api;
     bool                        m_rpc_server_verbose;
     uint8_t                     m_port_count;
     TrexPublisher              *m_publisher;
+    std::mutex                 *m_global_lock;
+
 };
 
 /**
@@ -186,7 +187,6 @@ protected:
 
     TrexPublisher                        *m_publisher;
 
-    std::mutex m_global_cp_lock;
 };
 
 /**
diff --git a/src/stateless/cp/trex_stateless_port.cpp b/src/stateless/cp/trex_stateless_port.cpp
index c60b0e85..7302e05d 100644
--- a/src/stateless/cp/trex_stateless_port.cpp
+++ b/src/stateless/cp/trex_stateless_port.cpp
@@ -48,11 +48,35 @@ port_id_to_cores(uint8_t port_id, std::vector<std::pair<uint8_t, uint8_t>> &core
 
 using namespace std;
 
+
+
+/***************************
+ * trex DP events handlers
+ * 
+ **************************/
+class AsyncStopEvent : public TrexDpPortEvent {
+
+protected:
+    /**
+     * when an async event occurs (all cores have reported in)
+     * 
+     * @author imarom (29-Feb-16)
+     */
+    virtual void on_event() {
+        get_port()->change_state(TrexStatelessPort::PORT_STATE_STREAMS);
+
+        get_port()->common_port_stop_actions(true);
+
+        assert(get_port()->m_pending_async_stop_event != TrexDpPortEvents::INVALID_ID);
+        get_port()->m_pending_async_stop_event = TrexDpPortEvents::INVALID_ID;
+    }
+};
+
 /***************************
  * trex stateless port
  * 
  **************************/
-TrexStatelessPort::TrexStatelessPort(uint8_t port_id, const TrexPlatformApi *api) {
+TrexStatelessPort::TrexStatelessPort(uint8_t port_id, const TrexPlatformApi *api) : m_dp_events(this) {
     std::vector<std::pair<uint8_t, uint8_t>> core_pair_list;
 
     m_port_id = port_id;
@@ -73,16 +97,20 @@ TrexStatelessPort::TrexStatelessPort(uint8_t port_id, const TrexPlatformApi *api
         m_cores_id_list.push_back(core_pair.first);
     }
 
-    /* init the events DP DB */
-    m_dp_events.create(this);
-
     m_graph_obj = NULL;
+
+    m_pending_async_stop_event = TrexDpPortEvents::INVALID_ID;
 }
 
 TrexStatelessPort::~TrexStatelessPort() {
     if (m_graph_obj) {
         delete m_graph_obj;
     }
+
+    if (m_pending_async_stop_event != TrexDpPortEvents::INVALID_ID) {
+        m_dp_events.destroy_event(m_pending_async_stop_event);
+        m_pending_async_stop_event = TrexDpPortEvents::INVALID_ID;
+    }
 }
 
 /**
@@ -170,16 +198,14 @@ TrexStatelessPort::start_traffic(const TrexPortMultiplier &mul, double duration,
     }
 
     /* generate a message to all the relevant DP cores to start transmitting */
-    int event_id = m_dp_events.generate_event_id();
-
-    /* mark that DP event of stoppped is possible */
-    m_dp_events.wait_for_event(TrexDpPortEvent::EVENT_STOP, event_id);
-
+    assert(m_pending_async_stop_event == TrexDpPortEvents::INVALID_ID);
+    m_pending_async_stop_event = m_dp_events.create_event(new AsyncStopEvent());
 
     /* update object status */
     m_factor = factor;
     m_last_all_streams_continues = compiled_objs[0]->get_all_streams_continues();
     m_last_duration = duration;
+
     change_state(PORT_STATE_TX);
 
     /* update the DP - messages will be freed by the DP */
@@ -188,17 +214,23 @@ TrexStatelessPort::start_traffic(const TrexPortMultiplier &mul, double duration,
 
         /* was the core assigned a compiled object ? */
         if (compiled_objs[index]) {
-            TrexStatelessCpToDpMsgBase *start_msg = new TrexStatelessDpStart(m_port_id, event_id, compiled_objs[index], duration);
+            TrexStatelessCpToDpMsgBase *start_msg = new TrexStatelessDpStart(m_port_id,
+                                                                             m_pending_async_stop_event,
+                                                                             compiled_objs[index],
+                                                                             duration);
             send_message_to_dp(core_id, start_msg);
         } else {
 
             /* mimic an end event */
-            m_dp_events.handle_event(TrexDpPortEvent::EVENT_STOP, core_id, event_id);
+            m_dp_events.on_core_reporting_in(m_pending_async_stop_event, core_id);
         }
 
         index++;
     }
     
+    /* for debug - this can be turn on */
+    //m_dp_events.barrier();
+
     /* update subscribers */    
     Json::Value data;
     data["port_id"] = m_port_id;
@@ -225,17 +257,23 @@ TrexStatelessPort::stop_traffic(void) {
     /* delete any previous graphs */
     delete_streams_graph();
 
-    /* mask out the DP stop event */
-    m_dp_events.disable(TrexDpPortEvent::EVENT_STOP);
+    /* to avoid race, first destroy any previous stop/pause events */
+    if (m_pending_async_stop_event != TrexDpPortEvents::INVALID_ID) {
+        m_dp_events.destroy_event(m_pending_async_stop_event);
+        m_pending_async_stop_event = TrexDpPortEvents::INVALID_ID;
 
+    }
+    
     /* generate a message to all the relevant DP cores to start transmitting */
     TrexStatelessCpToDpMsgBase *stop_msg = new TrexStatelessDpStop(m_port_id);
-
     send_message_to_all_dp(stop_msg);
 
-    /* continue to general actions */
+    /* a barrier - make sure all the DP cores stopped */
+    m_dp_events.barrier();
+
+    change_state(PORT_STATE_STREAMS);
+
     common_port_stop_actions(false);
-  
 }
 
 /**
@@ -243,14 +281,12 @@ TrexStatelessPort::stop_traffic(void) {
  * 
  */
 void
-TrexStatelessPort::common_port_stop_actions(bool event_triggered) {
+TrexStatelessPort::common_port_stop_actions(bool async) {
 
-    change_state(PORT_STATE_STREAMS);
-    
     Json::Value data;
     data["port_id"] = m_port_id;
 
-    if (event_triggered) {
+    if (async) {
         get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_FINISHED_TX, data);
     } else {
         get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_STOPPED, data);
@@ -274,10 +310,14 @@ TrexStatelessPort::pause_traffic(void) {
         throw TrexException(" pause is supported when duration is not enable is start command ");
     }
 
+    /* send a pause message */
     TrexStatelessCpToDpMsgBase *pause_msg = new TrexStatelessDpPause(m_port_id);
-
     send_message_to_all_dp(pause_msg);
 
+    /* make sure all DP cores paused */
+    m_dp_events.barrier();
+
+    /* change state */
     change_state(PORT_STATE_PAUSE);
 
     Json::Value data;
@@ -285,6 +325,7 @@ TrexStatelessPort::pause_traffic(void) {
     get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_PAUSED, data);
 }
 
+
 void
 TrexStatelessPort::resume_traffic(void) {
 
@@ -294,7 +335,6 @@ TrexStatelessPort::resume_traffic(void) {
     TrexStatelessCpToDpMsgBase *resume_msg = new TrexStatelessDpResume(m_port_id);
 
     send_message_to_all_dp(resume_msg);
-
     change_state(PORT_STATE_TX);
 
 
@@ -335,7 +375,6 @@ TrexStatelessPort::update_traffic(const TrexPortMultiplier &mul, bool force) {
     }
 
     TrexStatelessCpToDpMsgBase *update_msg = new TrexStatelessDpUpdate(m_port_id, factor);
-
     send_message_to_all_dp(update_msg);
 
     m_factor *= factor;
@@ -439,25 +478,6 @@ TrexStatelessPort::send_message_to_dp(uint8_t core_id, TrexStatelessCpToDpMsgBas
     ring->Enqueue((CGenNode *)msg);
 }
 
-/**
- * when a DP (async) event occurs - handle it
- * 
- */
-void 
-TrexStatelessPort::on_dp_event_occured(TrexDpPortEvent::event_e event_type) {
-    Json::Value data;
-
-    switch (event_type) {
-
-    case TrexDpPortEvent::EVENT_STOP:
-        common_port_stop_actions(true);
-        break;
-
-    default:
-        assert(0);
-
-    }
-}
 
 uint64_t
 TrexStatelessPort::get_port_speed_bps() const {
diff --git a/src/stateless/cp/trex_stateless_port.h b/src/stateless/cp/trex_stateless_port.h
index 192d0d19..d3c4dcb9 100644
--- a/src/stateless/cp/trex_stateless_port.h
+++ b/src/stateless/cp/trex_stateless_port.h
@@ -102,13 +102,17 @@ private:
 };
 
 
+class AsyncStopEvent;
+
 /**
  * describes a stateless port
  * 
  * @author imarom (31-Aug-15)
  */
 class TrexStatelessPort {
-    friend class TrexDpPortEvent;
+    friend TrexDpPortEvents;
+    friend TrexDpPortEvent;
+    friend AsyncStopEvent;
 
 public:
 
@@ -363,18 +367,12 @@ private:
      */
     void send_message_to_dp(uint8_t core_id, TrexStatelessCpToDpMsgBase *msg);
 
-    /**
-     * triggered when event occurs
-     * 
-     */
-    void on_dp_event_occured(TrexDpPortEvent::event_e event_type);
-
 
     /**
      * when a port stops, perform various actions
      * 
      */
-    void common_port_stop_actions(bool event_triggered);
+    void common_port_stop_actions(bool async);
 
     /**
      * calculate effective M per core
@@ -421,6 +419,8 @@ private:
 
     /* owner information */
     TrexPortOwner       m_owner;
+
+    int m_pending_async_stop_event;
 };
 
 
diff --git a/src/stateless/cp/trex_stream.h b/src/stateless/cp/trex_stream.h
index 161e9592..1abf0c04 100644
--- a/src/stateless/cp/trex_stream.h
+++ b/src/stateless/cp/trex_stream.h
@@ -514,7 +514,7 @@ public:
         bool      m_seq_enabled;
         bool      m_latency;
         uint32_t  m_user_id;
-
+        uint16_t  m_hw_id;
     } m_rx_check;
 
     uint32_t   m_burst_total_pkts; /* valid in case of burst stSINGLE_BURST,stMULTI_BURST*/
diff --git a/src/stateless/cp/trex_streams_compiler.cpp b/src/stateless/cp/trex_streams_compiler.cpp
index 7c91754c..be5002da 100644
--- a/src/stateless/cp/trex_streams_compiler.cpp
+++ b/src/stateless/cp/trex_streams_compiler.cpp
@@ -476,7 +476,8 @@ TrexStreamsCompiler::compile_stream(TrexStream *stream,
     }
 
     TrexStream *fixed_rx_flow_stat_stream = stream->clone(true);
-    get_stateless_obj()->m_rx_flow_stat.start_stream(fixed_rx_flow_stat_stream);
+
+    get_stateless_obj()->m_rx_flow_stat.start_stream(fixed_rx_flow_stat_stream, fixed_rx_flow_stat_stream->m_rx_check.m_hw_id); //???? check for errors
 
     /* can this stream be split to many cores ? */
     if (!stream->is_splitable(dp_core_count)) {
diff --git a/src/stateless/dp/trex_stateless_dp_core.cpp b/src/stateless/dp/trex_stateless_dp_core.cpp
index 549f923f..f8d6d828 100644
--- a/src/stateless/dp/trex_stateless_dp_core.cpp
+++ b/src/stateless/dp/trex_stateless_dp_core.cpp
@@ -265,9 +265,9 @@ bool TrexStatelessDpPerPort::pause_traffic(uint8_t port_id){
 }
 
 
-bool TrexStatelessDpPerPort::stop_traffic(uint8_t port_id,
-                                          bool stop_on_id, 
-                                          int event_id){
+bool TrexStatelessDpPerPort::stop_traffic(uint8_t  port_id,
+                                          bool     stop_on_id, 
+                                          int      event_id){
 
 
     if (m_state == TrexStatelessDpPerPort::ppSTATE_IDLE) {
@@ -394,12 +394,27 @@ bool TrexStatelessDpCore::set_stateless_next_node(CGenNodeStateless * cur_node,
 void 
 TrexStatelessDpCore::idle_state_loop() {
 
+    const int SHORT_DELAY_MS    = 2;
+    const int LONG_DELAY_MS     = 50;
+    const int DEEP_SLEEP_LIMIT  = 2000;
+
+    int counter = 0;
+
     while (m_state == STATE_IDLE) {
         bool had_msg = periodic_check_for_cp_messages();
-        /* if no message - backoff for some time */
-        if (!had_msg) {
-            delay(200);
+        if (had_msg) {
+            counter = 0;
+            continue;
+        }
+
+        /* enter deep sleep only if enough time had passed */
+        if (counter < DEEP_SLEEP_LIMIT) {
+            delay(SHORT_DELAY_MS);
+            counter++;
+        } else {
+            delay(LONG_DELAY_MS);
         }
+        
     }
 }
 
@@ -565,7 +580,6 @@ void
 TrexStatelessDpCore::add_stream(TrexStatelessDpPerPort * lp_port,
                                 TrexStream * stream,
                                 TrexStreamsCompiledObj *comp) {
-
     CGenNodeStateless *node = m_core->create_node_sl();
 
     /* add periodic */
@@ -581,7 +595,6 @@ TrexStatelessDpCore::add_stream(TrexStatelessDpPerPort * lp_port,
 
     node->m_next_stream=0; /* will be fixed later */
 
-
     if ( stream->m_self_start ){
         /* if self start it is in active mode */
         node->m_state =CGenNodeStateless::ss_ACTIVE;
@@ -597,7 +610,12 @@ TrexStatelessDpCore::add_stream(TrexStatelessDpPerPort * lp_port,
     node->m_src_port =0;
     node->m_original_packet_data_prefix = 0;
 
-
+    if (stream->m_rx_check.m_enabled) {
+        node->set_stat_needed();
+        uint8_t hw_id = stream->m_rx_check.m_hw_id;
+        assert (hw_id < MAX_FLOW_STATS);
+        node->set_stat_hw_id(hw_id);
+    }
 
     /* set socket id */
     node->set_socket_id(m_core->m_node_gen.m_socket_id);
@@ -826,9 +844,9 @@ TrexStatelessDpCore::update_traffic(uint8_t port_id, double factor) {
 
 
 void
-TrexStatelessDpCore::stop_traffic(uint8_t port_id,
-                                  bool stop_on_id, 
-                                  int event_id) {
+TrexStatelessDpCore::stop_traffic(uint8_t  port_id,
+                                  bool     stop_on_id, 
+                                  int      event_id) {
     /* we cannot remove nodes not from the top of the queue so
        for every active node - make sure next time
        the scheduler invokes it, it will be free */
@@ -840,20 +858,19 @@ TrexStatelessDpCore::stop_traffic(uint8_t port_id,
         //printf(" skip .. %f\n",m_core->m_cur_time_sec);
         return;
     }
-
-#if 0
-    if ( are_all_ports_idle() ) {
-        /* just a place holder if we will need to do somthing in that case */
-    }
-#endif
  
     /* inform the control plane we stopped - this might be a async stop
        (streams ended)
-     */
+    */
+    #if 0
+    if ( are_all_ports_idle() ) {
+        /* just a place holder if we will need to do somthing in that case */
+    }
+    #endif
+
     CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingDpToCp(m_core->m_thread_id);
     TrexStatelessDpToCpMsgBase *event_msg = new TrexDpPortEventMsg(m_core->m_thread_id,
                                                                    port_id,
-                                                                   TrexDpPortEvent::EVENT_STOP,
                                                                    lp_port->get_event_id());
     ring->Enqueue((CGenNode *)event_msg);
 
@@ -869,3 +886,12 @@ TrexStatelessDpCore::handle_cp_msg(TrexStatelessCpToDpMsgBase *msg) {
     delete msg;
 }
 
+void
+TrexStatelessDpCore::barrier(uint8_t port_id, int event_id) {
+
+    CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingDpToCp(m_core->m_thread_id);
+    TrexStatelessDpToCpMsgBase *event_msg = new TrexDpPortEventMsg(m_core->m_thread_id,
+                                                                   port_id,
+                                                                   event_id);
+    ring->Enqueue((CGenNode *)event_msg);
+}
diff --git a/src/stateless/dp/trex_stateless_dp_core.h b/src/stateless/dp/trex_stateless_dp_core.h
index 3d214655..cb102b8d 100644
--- a/src/stateless/dp/trex_stateless_dp_core.h
+++ b/src/stateless/dp/trex_stateless_dp_core.h
@@ -235,7 +235,8 @@ public:
         return (&m_ports[local_port_id]);
     }
         
-
+    /* simply sends a message back (acts as a barrier for previous messages) */
+    void barrier(uint8_t port_id, int event_id);
 
 private:
 
diff --git a/src/stateless/dp/trex_stream_node.h b/src/stateless/dp/trex_stream_node.h
index b366a770..104e4d3b 100644
--- a/src/stateless/dp/trex_stream_node.h
+++ b/src/stateless/dp/trex_stream_node.h
@@ -62,7 +62,8 @@ public:
 
         SL_NODE_CONST_MBUF                 =4,
 
-        SL_NODE_VAR_PKT_SIZE               =8
+        SL_NODE_VAR_PKT_SIZE               =8,
+        SL_NODE_STATS_NEEDED               = 0x10
 
     };
 
@@ -82,7 +83,8 @@ private:
 
     double              m_next_time_offset; /* in sec */
     uint16_t            m_action_counter;
-    uint16_t            m_pad11;
+    uint8_t             m_stat_hw_id; // hw id used to count rx and tx stats
+    uint8_t             m_pad11;
     uint32_t            m_pad12;
 
     stream_state_t      m_state;
@@ -269,6 +271,22 @@ public:
         return ( m_socket_id );
     }
 
+    void set_stat_hw_id(uint16_t hw_id) {
+        m_stat_hw_id = hw_id;
+    }
+
+    socket_id_t get_stat_hw_id() {
+        return ( m_stat_hw_id );
+    }
+
+    inline void set_stat_needed() {
+        m_flags |= SL_NODE_STATS_NEEDED;
+    }
+
+    inline bool is_stat_needed() {
+        return ((m_flags & SL_NODE_STATS_NEEDED) != 0);
+    }
+
     inline void set_mbuf_cache_dir(pkt_dir_t  dir){
         if (dir) {
             m_flags |=NODE_FLAGS_DIR;
diff --git a/src/stateless/messaging/trex_stateless_messaging.cpp b/src/stateless/messaging/trex_stateless_messaging.cpp
index 257de168..333aec88 100644
--- a/src/stateless/messaging/trex_stateless_messaging.cpp
+++ b/src/stateless/messaging/trex_stateless_messaging.cpp
@@ -180,11 +180,29 @@ TrexStatelessDpUpdate::clone() {
     return new_msg;
 }
 
+/*************************
+  barrier message
+ ************************/
+
+bool
+TrexStatelessDpBarrier::handle(TrexStatelessDpCore *dp_core) {
+    dp_core->barrier(m_port_id, m_event_id);
+    return true;
+}
+
+TrexStatelessCpToDpMsgBase *
+TrexStatelessDpBarrier::clone() {
+
+    TrexStatelessCpToDpMsgBase *new_msg = new TrexStatelessDpBarrier(m_port_id, m_event_id);
+
+    return new_msg;
+}
+
 /************************* messages from DP to CP **********************/
 bool
 TrexDpPortEventMsg::handle() {
     TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(m_port_id);
-    port->get_dp_events().handle_event(m_event_type, m_thread_id, m_event_id);
+    port->get_dp_events().on_core_reporting_in(m_event_id, m_thread_id);
 
     return (true);
 }
diff --git a/src/stateless/messaging/trex_stateless_messaging.h b/src/stateless/messaging/trex_stateless_messaging.h
index d56596bf..dda086b7 100644
--- a/src/stateless/messaging/trex_stateless_messaging.h
+++ b/src/stateless/messaging/trex_stateless_messaging.h
@@ -145,7 +145,7 @@ public:
 
     TrexStatelessDpStop(uint8_t port_id) : m_port_id(port_id) {
         m_stop_only_for_event_id=false;
-        m_event_id=0;
+        m_event_id = 0;
         m_core = NULL;
     }
 
@@ -245,6 +245,26 @@ private:
     double   m_factor;
 };
 
+/**
+ * barrier message for DP core
+ * 
+ */
+class TrexStatelessDpBarrier : public TrexStatelessCpToDpMsgBase {
+public:
+
+    TrexStatelessDpBarrier(uint8_t port_id, int event_id) {
+        m_port_id  = port_id;
+        m_event_id = event_id;
+    }
+
+    virtual bool handle(TrexStatelessDpCore *dp_core);
+
+    virtual TrexStatelessCpToDpMsgBase * clone();
+
+private:
+    uint8_t m_port_id;
+    int     m_event_id;
+};
 
 /************************* messages from DP to CP **********************/
 
@@ -282,10 +302,9 @@ public:
 class TrexDpPortEventMsg : public TrexStatelessDpToCpMsgBase {
 public:
 
-    TrexDpPortEventMsg(int thread_id, uint8_t port_id, TrexDpPortEvent::event_e type, int event_id) {
+    TrexDpPortEventMsg(int thread_id, uint8_t port_id, int event_id) {
         m_thread_id  = thread_id;
         m_port_id    = port_id;
-        m_event_type = type;
         m_event_id   = event_id;
     }
 
@@ -299,10 +318,6 @@ public:
         return m_port_id;
     }
 
-    TrexDpPortEvent::event_e get_event_type() {
-        return m_event_type;
-    }
-
     int get_event_id() {
         return m_event_id;
     }
@@ -310,7 +325,6 @@ public:
 private:
     int                         m_thread_id;
     uint8_t                     m_port_id;
-    TrexDpPortEvent::event_e    m_event_type;
     int                         m_event_id;
     
 };
diff --git a/src/trex_defs.h b/src/trex_defs.h
index bb8510fa..ace3618d 100644
--- a/src/trex_defs.h
+++ b/src/trex_defs.h
@@ -17,7 +17,8 @@ limitations under the License.
 #define __TREX_DEFS_H__
 
 #define TREX_MAX_PORTS 12
-#define TREX_FDIR_STAT_SIZE 128
+
+#define MAX_FLOW_STATS 128
 
 #ifndef UINT8_MAX
     #define UINT8_MAX 255