RX stats two stage API (stop and remove RX filters)

3 files changed, 101 insertions, 8 deletions

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 6fdc3454..f6b5d49c 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
@@ -852,8 +852,26 @@ class STLClient(object):
         return RC_OK()
 
 
- 
- 
+    # remove all RX filters in a safe manner
+    def _remove_rx_filters (self, ports, rx_delay_ms):
+
+        # get the enabled RX ports
+        rx_ports = [port_id for port_id in ports if self.ports[port_id].has_rx_enabled()]
+
+        if not rx_ports:
+            return RC_OK()
+
+        # block while any RX configured port has not yet have it's delay expired
+        while any([not self.ports[port_id].has_rx_delay_expired(rx_delay_ms) for port_id in rx_ports]):
+            time.sleep(0.01)
+
+        # remove RX filters
+        rc = RC()
+        for port_id in rx_ports:
+            rc.add(self.ports[port_id].remove_rx_filters())
+
+        return rc
+
 
     #################################
     # ------ private methods ------ #
@@ -1089,6 +1107,7 @@ class STLClient(object):
                 for port_id, port_obj in self.ports.items()
                 if port_obj.is_active()]
 
+
     # get paused ports
     def get_paused_ports (self):
         return [port_id
@@ -1336,7 +1355,7 @@ class STLClient(object):
         ports = self._validate_port_list(ports)
 
         self.acquire(ports, force = True)
-        self.stop(ports)
+        self.stop(ports, rx_delay_ms = 0)
         self.remove_all_streams(ports)
         self.clear_stats(ports)
 
@@ -1535,7 +1554,7 @@ class STLClient(object):
 
     
     @__api_check(True)
-    def stop (self, ports = None):
+    def stop (self, ports = None, rx_delay_ms = 10):
         """
             stop port(s)
 
@@ -1543,6 +1562,13 @@ class STLClient(object):
                 ports : list
                     ports to execute the command
 
+                rx_delay_ms : int
+                    time to wait until RX filters are removed
+                    this value should reflect the time it takes
+                    packets which were transmitted to arrive
+                    to the destination.
+                    after this time the RX filters will be removed
+
             :raises:
                 + :exc:`STLError`
 
@@ -1561,6 +1587,11 @@ class STLClient(object):
         if not rc:
             raise STLError(rc)
 
+        # remove any RX filters
+        rc = self._remove_rx_filters(ports, rx_delay_ms = rx_delay_ms)
+        if not rc:
+            raise STLError(rc)
+
         
     @__api_check(True)
     def update (self, ports = None, mult = "1", total = False, force = False):
@@ -2025,7 +2056,7 @@ class STLClient(object):
             self.logger.log(format_text("No active traffic on provided ports\n", 'bold'))
             return
 
-        self.stop(ports)
+        self.stop(ports, rx_delay_ms = 2000)
 
         # true means print time
         return True
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 47124114..049929ae 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
@@ -7,8 +7,8 @@ from .trex_stl_types import *
 from . import trex_stl_stats
 
 import base64
-import time
 import copy
+from datetime import datetime, timedelta
 
 StreamOnPort = namedtuple('StreamOnPort', ['compiled_stream', 'metadata'])
 
@@ -61,6 +61,8 @@ class Port(object):
         self.port_stats = trex_stl_stats.CPortStats(self)
 
         self.next_available_id = 1
+        self.tx_stopped_ts = None
+        self.has_rx_streams = False
 
 
     def err(self, msg):
@@ -239,6 +241,9 @@ class Port(object):
                                            'rate' : streams_list[i].get_rate()}
 
                 ret.add(RC_OK(data = stream_id))
+
+                self.has_rx_streams = self.has_rx_streams or streams_list[i].has_flow_stats()
+
             else:
                 ret.add(RC(*single_rc))
 
@@ -283,6 +288,9 @@ class Port(object):
 
         self.state = self.STATE_STREAMS if (len(self.streams) > 0) else self.STATE_IDLE
 
+        # recheck if any RX stats streams present on the port
+        self.has_rx_streams = any([stream.has_flow_stats() for stream in self.streams])
+
         return self.ok() if rc else self.err(rc.err())
 
 
@@ -305,6 +313,7 @@ class Port(object):
         self.streams = {}
 
         self.state = self.STATE_IDLE
+        self.has_rx_streams = False
 
         return self.ok()
 
@@ -351,7 +360,7 @@ class Port(object):
     # stop traffic
     # with force ignores the cached state and sends the command
     def stop (self, force = False):
-
+        
         if not self.is_acquired():
             return self.err("port is not owned")
 
@@ -360,7 +369,6 @@ class Port(object):
             if (self.state == self.STATE_IDLE) or (self.state == self.state == self.STATE_STREAMS):
                 return self.ok()
 
-
         params = {"handler": self.handler,
                   "port_id": self.port_id}
 
@@ -370,8 +378,56 @@ class Port(object):
 
         self.state = self.STATE_STREAMS
 
+        # timestamp for last tx
+        self.tx_stopped_ts = datetime.now()
+
+        return self.ok()
+
+
+    # return True if port has any stream configured with RX stats
+    def has_rx_enabled (self):
+        return self.has_rx_streams
+
+
+    # return true if rx_delay_ms has passed since the last port stop
+    def has_rx_delay_expired (self, rx_delay_ms):
+        assert(self.has_rx_enabled())
+
+        # if active - it's not safe to remove RX filters
+        if self.is_active():
+            return False
+
+        # either no timestamp present or time has already passed
+        return not self.tx_stopped_ts or (datetime.now() - self.tx_stopped_ts) > timedelta(milliseconds = rx_delay_ms)
+
+
+
+    def remove_rx_filters (self):
+        assert(self.has_rx_enabled())
+
+        if not self.is_acquired():
+            return self.err("port is not owned")
+
+        if self.state == self.STATE_DOWN:
+            return self.err("Unable to remove RX filters - port is down")
+
+        if self.state == self.STATE_TX:
+            return self.err("Unable to remove RX filters - port is transmitting")
+
+        if self.state == self.STATE_IDLE:
+            return self.ok()
+
+
+        params = {"handler": self.handler,
+                  "port_id": self.port_id}
+
+        rc = self.transmit("remove_rx_filters", params)
+        if rc.bad():
+            return self.err(rc.err())
+
         return self.ok()
 
+
     def pause (self):
 
         if not self.is_acquired():
@@ -597,6 +653,8 @@ class Port(object):
 
   ################# events handler ######################
     def async_event_port_job_done (self):
+        # until thread is locked - order is important
+        self.tx_stopped_ts = datetime.now()
         self.state = self.STATE_STREAMS
 
     # rest of the events are used for TUI / read only sessions
diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py
index 4f8ce3e6..b6780c5e 100644
--- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py
+++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py
@@ -430,6 +430,10 @@ class STLStream(object):
         return self.next
 
 
+    def has_flow_stats (self):
+        """ Return True if stream was configured with flow stats """
+        return self.fields['flow_stats']['enabled']
+
     def get_pkt (self):
         """ Get packet as string """
         return self.pkt