Migrate Trex TG to use new NSPerf-RFC2544 scenario

JIRA: YARDSTICK-1592

Change-Id: Ic51a52385ac53df9b5f56387af4f1b661bb0d86f
Signed-off-by: Mytnyk, Volodymyr <volodymyrx.mytnyk@intel.com>

3 files changed, 89 insertions, 42 deletions

diff --git a/yardstick/network_services/vnf_generic/vnf/sample_vnf.py b/yardstick/network_services/vnf_generic/vnf/sample_vnf.py
index aec085057..743aed0ee 100644
--- a/yardstick/network_services/vnf_generic/vnf/sample_vnf.py
+++ b/yardstick/network_services/vnf_generic/vnf/sample_vnf.py
@@ -14,7 +14,7 @@
 
 import logging
 import decimal
-from multiprocessing import Queue, Value, Process
+from multiprocessing import Queue, Value, Process, JoinableQueue
 import os
 import posixpath
 import re
@@ -948,6 +948,39 @@ class SampleVNFTrafficGen(GenericTrafficGen):
         self.traffic_finished = False
         self._tg_process = None
         self._traffic_process = None
+        self._tasks_queue = JoinableQueue()
+        self._result_queue = Queue()
+
+    def _test_runner(self, traffic_profile, tasks, results):
+        self.resource_helper.run_test(traffic_profile, tasks, results)
+
+    def _init_traffic_process(self, traffic_profile):
+        name = '{}-{}-{}-{}'.format(self.name, self.APP_NAME,
+                                    traffic_profile.__class__.__name__,
+                                    os.getpid())
+        self._traffic_process = Process(name=name, target=self._test_runner,
+                                        args=(
+                                        traffic_profile, self._tasks_queue,
+                                        self._result_queue))
+
+        self._traffic_process.start()
+        while self.resource_helper.client_started.value == 0:
+            time.sleep(1)
+            if not self._traffic_process.is_alive():
+                break
+
+    def run_traffic_once(self, traffic_profile):
+        if self.resource_helper.client_started.value == 0:
+            self._init_traffic_process(traffic_profile)
+
+        # continue test - run next iteration
+        LOG.info("Run next iteration ...")
+        self._tasks_queue.put('RUN_TRAFFIC')
+
+    def wait_on_traffic(self):
+        self._tasks_queue.join()
+        result = self._result_queue.get()
+        return result
 
     def _start_server(self):
         # we can't share ssh paramiko objects to force new connection
diff --git a/yardstick/network_services/vnf_generic/vnf/tg_rfc2544_ixia.py b/yardstick/network_services/vnf_generic/vnf/tg_rfc2544_ixia.py
index d3a2d9443..2f9eb3f0d 100644
--- a/yardstick/network_services/vnf_generic/vnf/tg_rfc2544_ixia.py
+++ b/yardstick/network_services/vnf_generic/vnf/tg_rfc2544_ixia.py
@@ -16,13 +16,8 @@ import ipaddress
 import logging
 import six
 import collections
-import os
-import time
 
 from six import moves
-
-from multiprocessing import Queue, Process, JoinableQueue
-
 from yardstick.common import utils
 from yardstick.common import exceptions
 from yardstick.network_services.libs.ixia_libs.ixnet import ixnet_api
@@ -844,9 +839,6 @@ class IxiaTrafficGen(SampleVNFTrafficGen):
         self._ixia_traffic_gen = None
         self.ixia_file_name = ''
         self.vnf_port_pairs = []
-        self._traffic_process = None
-        self._tasks_queue = JoinableQueue()
-        self._result_queue = Queue()
 
     def _check_status(self):
         pass
@@ -854,34 +846,3 @@ class IxiaTrafficGen(SampleVNFTrafficGen):
     def terminate(self):
         self.resource_helper.stop_collect()
         super(IxiaTrafficGen, self).terminate()
-
-    def _test_runner(self, traffic_profile, tasks, results):
-        self.resource_helper.run_test(traffic_profile, tasks, results)
-
-    def _init_traffic_process(self, traffic_profile):
-        name = '{}-{}-{}-{}'.format(self.name, self.APP_NAME,
-                                    traffic_profile.__class__.__name__,
-                                    os.getpid())
-        self._traffic_process = Process(name=name, target=self._test_runner,
-                                        args=(
-                                        traffic_profile, self._tasks_queue,
-                                        self._result_queue))
-
-        self._traffic_process.start()
-        while self.resource_helper.client_started.value == 0:
-            time.sleep(1)
-            if not self._traffic_process.is_alive():
-                break
-
-    def run_traffic_once(self, traffic_profile):
-        if self.resource_helper.client_started.value == 0:
-            self._init_traffic_process(traffic_profile)
-
-        # continue test - run next iteration
-        LOG.info("Run next iteration ...")
-        self._tasks_queue.put('RUN_TRAFFIC')
-
-    def wait_on_traffic(self):
-        self._tasks_queue.join()
-        result = self._result_queue.get()
-        return result
diff --git a/yardstick/network_services/vnf_generic/vnf/tg_rfc2544_trex.py b/yardstick/network_services/vnf_generic/vnf/tg_rfc2544_trex.py
index 24ee313e2..a9c0222ac 100644
--- a/yardstick/network_services/vnf_generic/vnf/tg_rfc2544_trex.py
+++ b/yardstick/network_services/vnf_generic/vnf/tg_rfc2544_trex.py
@@ -15,12 +15,14 @@
 import logging
 import time
 
+from six import moves
 from yardstick.common import utils
 from yardstick.network_services.vnf_generic.vnf import sample_vnf
 from yardstick.network_services.vnf_generic.vnf import tg_trex
+from trex_stl_lib.trex_stl_exceptions import STLError
 
 
-LOGGING = logging.getLogger(__name__)
+LOG = logging.getLogger(__name__)
 
 
 class TrexRfcResourceHelper(tg_trex.TrexResourceHelper):
@@ -48,7 +50,8 @@ class TrexRfcResourceHelper(tg_trex.TrexResourceHelper):
         completed, output = traffic_profile.get_drop_percentage(
             samples, self.rfc2544_helper.tolerance_low,
             self.rfc2544_helper.tolerance_high,
-            self.rfc2544_helper.correlated_traffic)
+            self.rfc2544_helper.correlated_traffic,
+            self.rfc2544_helper.resolution)
         self._queue.put(output)
         return completed
 
@@ -58,6 +61,56 @@ class TrexRfcResourceHelper(tg_trex.TrexResourceHelper):
     def clear_client_stats(self, ports):
         self.client.clear_stats(ports=ports)
 
+    def run_test(self, traffic_profile, tasks_queue, results_queue, *args): # pragma: no cover
+        LOG.debug("Trex resource_helper run_test")
+        if self._terminated.value:
+            return
+        # if we don't do this we can hang waiting for the queue to drain
+        # have to do this in the subprocess
+        self._queue.cancel_join_thread()
+        try:
+            self._build_ports()
+            self.client = self._connect()
+            self.client.reset(ports=self.all_ports)
+            self.client.remove_all_streams(self.all_ports)  # remove all streams
+            traffic_profile.register_generator(self)
+
+            completed = False
+            self.rfc2544_helper.iteration.value = 0
+            self.client_started.value = 1
+            while completed is False and not self._terminated.value:
+                LOG.debug("Wait for task ...")
+                try:
+                    task = tasks_queue.get(True, 5)
+                except moves.queue.Empty:
+                    LOG.debug("Wait for task timeout, continue waiting...")
+                    continue
+                else:
+                    if task != 'RUN_TRAFFIC':
+                        continue
+                self.rfc2544_helper.iteration.value += 1
+                LOG.info("Got %s task, start iteration %d", task,
+                         self.rfc2544_helper.iteration.value)
+                completed = self._run_traffic_once(traffic_profile)
+                if completed:
+                    LOG.debug("%s::run_test - test completed",
+                              self.__class__.__name__)
+                    results_queue.put('COMPLETE')
+                else:
+                    results_queue.put('CONTINUE')
+                tasks_queue.task_done()
+
+            self.client.stop(self.all_ports)
+            self.client.disconnect()
+            self._terminated.value = 0
+        except STLError:
+            if self._terminated.value:
+                LOG.debug("traffic generator is stopped")
+                return  # return if trex/tg server is stopped.
+            raise
+
+        self.client_started.value = 0
+        LOG.debug("%s::run_test done", self.__class__.__name__)
 
 class TrexTrafficGenRFC(tg_trex.TrexTrafficGen):
     """