Merge "NSB sync: clean-up draft IPC implementation (part 1)"

21 files changed, 81 insertions, 195 deletions

diff --git a/yardstick/network_services/vnf_generic/vnf/acl_vnf.py b/yardstick/network_services/vnf_generic/vnf/acl_vnf.py
index 11a602472..e51d38748 100644
--- a/yardstick/network_services/vnf_generic/vnf/acl_vnf.py
+++ b/yardstick/network_services/vnf_generic/vnf/acl_vnf.py
@@ -246,9 +246,8 @@ class AclApproxVnf(SampleVNF):
         'packets_dropped': 2,
     }
 
-    def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
-                 resource_helper_type=None):
+    def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
         if setup_env_helper_type is None:
             setup_env_helper_type = AclApproxSetupEnvSetupEnvHelper
-        super(AclApproxVnf, self).__init__(
-            name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
+
+        super(AclApproxVnf, self).__init__(name, vnfd, setup_env_helper_type, resource_helper_type)
diff --git a/yardstick/network_services/vnf_generic/vnf/agnostic_vnf.py b/yardstick/network_services/vnf_generic/vnf/agnostic_vnf.py
index 115fddcf0..5856eb6e0 100644
--- a/yardstick/network_services/vnf_generic/vnf/agnostic_vnf.py
+++ b/yardstick/network_services/vnf_generic/vnf/agnostic_vnf.py
@@ -21,8 +21,8 @@ LOG = logging.getLogger(__name__)
 
 class AgnosticVnf(base.GenericVNF):
     """ AgnosticVnf implementation. """
-    def __init__(self, name, vnfd, task_id):
-        super(AgnosticVnf, self).__init__(name, vnfd, task_id)
+    def __init__(self, name, vnfd):
+        super(AgnosticVnf, self).__init__(name, vnfd)
 
     def instantiate(self, scenario_cfg, context_cfg):
         pass
diff --git a/yardstick/network_services/vnf_generic/vnf/base.py b/yardstick/network_services/vnf_generic/vnf/base.py
index 0fb310075..1f49b22d5 100644
--- a/yardstick/network_services/vnf_generic/vnf/base.py
+++ b/yardstick/network_services/vnf_generic/vnf/base.py
@@ -17,10 +17,6 @@ import abc
 import logging
 import six
 
-from yardstick.common import messaging
-from yardstick.common.messaging import consumer
-from yardstick.common.messaging import payloads
-from yardstick.common.messaging import producer
 from yardstick.network_services.helpers.samplevnf_helper import PortPairs
 
 
@@ -141,70 +137,6 @@ class VnfdHelper(dict):
             yield port_name, port_num
 
 
-class TrafficGeneratorProducer(producer.MessagingProducer):
-    """Class implementing the message producer for traffic generators
-
-    This message producer must be instantiated in the process created
-    "run_traffic" process.
-    """
-    def __init__(self, _id):
-        super(TrafficGeneratorProducer, self).__init__(messaging.TOPIC_TG,
-                                                       _id=_id)
-
-    def tg_method_started(self, version=1):
-        """Send a message to inform the traffic generation has started"""
-        self.send_message(
-            messaging.TG_METHOD_STARTED,
-            payloads.TrafficGeneratorPayload(version=version, iteration=0,
-                                             kpi={}))
-
-    def tg_method_finished(self, version=1):
-        """Send a message to inform the traffic generation has finished"""
-        self.send_message(
-            messaging.TG_METHOD_FINISHED,
-            payloads.TrafficGeneratorPayload(version=version, iteration=0,
-                                             kpi={}))
-
-    def tg_method_iteration(self, iteration, version=1, kpi=None):
-        """Send a message, with KPI, once an iteration has finished"""
-        kpi = {} if kpi is None else kpi
-        self.send_message(
-            messaging.TG_METHOD_ITERATION,
-            payloads.TrafficGeneratorPayload(version=version,
-                                             iteration=iteration, kpi=kpi))
-
-
-@six.add_metaclass(abc.ABCMeta)
-class GenericVNFEndpoint(consumer.NotificationHandler):
-    """Endpoint class for ``GenericVNFConsumer``"""
-
-    @abc.abstractmethod
-    def runner_method_start_iteration(self, ctxt, **kwargs):
-        """Endpoint when RUNNER_METHOD_START_ITERATION is received
-
-        :param ctxt: (dict) {'id': <Producer ID>}
-        :param kwargs: (dict) ``payloads.RunnerPayload`` context
-        """
-
-    @abc.abstractmethod
-    def runner_method_stop_iteration(self, ctxt, **kwargs):
-        """Endpoint when RUNNER_METHOD_STOP_ITERATION is received
-
-        :param ctxt: (dict) {'id': <Producer ID>}
-        :param kwargs: (dict) ``payloads.RunnerPayload`` context
-        """
-
-
-class GenericVNFConsumer(consumer.MessagingConsumer):
-    """MQ consumer for ``GenericVNF`` derived classes"""
-
-    def __init__(self, ctx_ids, endpoints):
-        if not isinstance(endpoints, list):
-            endpoints = [endpoints]
-        super(GenericVNFConsumer, self).__init__(messaging.TOPIC_RUNNER,
-                                                 ctx_ids, endpoints)
-
-
 @six.add_metaclass(abc.ABCMeta)
 class GenericVNF(object):
     """Class providing file-like API for generic VNF implementation
@@ -217,9 +149,8 @@ class GenericVNF(object):
     UPLINK = PortPairs.UPLINK
     DOWNLINK = PortPairs.DOWNLINK
 
-    def __init__(self, name, vnfd, task_id):
+    def __init__(self, name, vnfd):
         self.name = name
-        self._task_id = task_id
         self.vnfd_helper = VnfdHelper(vnfd)
         # List of statistics we can obtain from this VNF
         # - ETSI MANO 6.3.1.1 monitoring_parameter
@@ -280,11 +211,10 @@ class GenericVNF(object):
 class GenericTrafficGen(GenericVNF):
     """Class providing file-like API for generic traffic generator"""
 
-    def __init__(self, name, vnfd, task_id):
-        super(GenericTrafficGen, self).__init__(name, vnfd, task_id)
+    def __init__(self, name, vnfd):
+        super(GenericTrafficGen, self).__init__(name, vnfd)
         self.runs_traffic = True
         self.traffic_finished = False
-        self._mq_producer = None
 
     @abc.abstractmethod
     def run_traffic(self, traffic_profile):
@@ -355,16 +285,3 @@ class GenericTrafficGen(GenericVNF):
         :return: True/False
         """
         pass
-
-    @staticmethod
-    def _setup_mq_producer(id):
-        """Setup the TG MQ producer to send messages between processes
-
-        :return: (derived class from ``MessagingProducer``) MQ producer object
-        """
-        return TrafficGeneratorProducer(id)
-
-    def get_mq_producer_id(self):
-        """Return the MQ producer ID if initialized"""
-        if self._mq_producer:
-            return self._mq_producer.id
diff --git a/yardstick/network_services/vnf_generic/vnf/cgnapt_vnf.py b/yardstick/network_services/vnf_generic/vnf/cgnapt_vnf.py
index 14f1e2e97..bfe628f09 100644
--- a/yardstick/network_services/vnf_generic/vnf/cgnapt_vnf.py
+++ b/yardstick/network_services/vnf_generic/vnf/cgnapt_vnf.py
@@ -85,12 +85,12 @@ class CgnaptApproxVnf(SampleVNF):
         "packets_dropped": 4,
     }
 
-    def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
-                 resource_helper_type=None):
+    def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
         if setup_env_helper_type is None:
             setup_env_helper_type = CgnaptApproxSetupEnvHelper
-        super(CgnaptApproxVnf, self).__init__(
-            name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
+
+        super(CgnaptApproxVnf, self).__init__(name, vnfd, setup_env_helper_type,
+                                              resource_helper_type)
 
     def _vnf_up_post(self):
         super(CgnaptApproxVnf, self)._vnf_up_post()
diff --git a/yardstick/network_services/vnf_generic/vnf/epc_vnf.py b/yardstick/network_services/vnf_generic/vnf/epc_vnf.py
index 66d16d07f..1b06c59c1 100644
--- a/yardstick/network_services/vnf_generic/vnf/epc_vnf.py
+++ b/yardstick/network_services/vnf_generic/vnf/epc_vnf.py
@@ -21,8 +21,8 @@ LOG = logging.getLogger(__name__)
 
 class EPCVnf(base.GenericVNF):
 
-    def __init__(self, name, vnfd, task_id):
-        super(EPCVnf, self).__init__(name, vnfd, task_id)
+    def __init__(self, name, vnfd):
+        super(EPCVnf, self).__init__(name, vnfd)
 
     def instantiate(self, scenario_cfg, context_cfg):
         """Prepare VNF for operation and start the VNF process/VM
diff --git a/yardstick/network_services/vnf_generic/vnf/prox_helpers.py b/yardstick/network_services/vnf_generic/vnf/prox_helpers.py
index cd3035ef8..aeb41858f 100644
--- a/yardstick/network_services/vnf_generic/vnf/prox_helpers.py
+++ b/yardstick/network_services/vnf_generic/vnf/prox_helpers.py
@@ -1146,7 +1146,7 @@ class ProxResourceHelper(ClientResourceHelper):
             self._test_type = self.setup_helper.find_in_section('global', 'name', None)
         return self._test_type
 
-    def run_traffic(self, traffic_profile, *args):
+    def run_traffic(self, traffic_profile):
         self._queue.cancel_join_thread()
         self.lower = 0.0
         self.upper = 100.0
diff --git a/yardstick/network_services/vnf_generic/vnf/prox_irq.py b/yardstick/network_services/vnf_generic/vnf/prox_irq.py
index dda26b0fe..4cc8923cb 100644
--- a/yardstick/network_services/vnf_generic/vnf/prox_irq.py
+++ b/yardstick/network_services/vnf_generic/vnf/prox_irq.py
@@ -28,13 +28,13 @@ LOG = logging.getLogger(__name__)
 
 class ProxIrq(SampleVNFTrafficGen):
 
-    def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+    def __init__(self, name, vnfd, setup_env_helper_type=None,
                  resource_helper_type=None):
         vnfd_cpy = copy.deepcopy(vnfd)
-        super(ProxIrq, self).__init__(name, vnfd_cpy, task_id)
+        super(ProxIrq, self).__init__(name, vnfd_cpy)
 
         self._vnf_wrapper = ProxApproxVnf(
-            name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
+            name, vnfd, setup_env_helper_type, resource_helper_type)
         self.bin_path = get_nsb_option('bin_path', '')
         self.name = self._vnf_wrapper.name
         self.ssh_helper = self._vnf_wrapper.ssh_helper
@@ -83,9 +83,9 @@ class ProxIrqVNF(ProxIrq, SampleVNFTrafficGen):
 
     APP_NAME = 'ProxIrqVNF'
 
-    def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+    def __init__(self, name, vnfd, setup_env_helper_type=None,
                  resource_helper_type=None):
-        ProxIrq.__init__(self, name, vnfd, task_id, setup_env_helper_type,
+        ProxIrq.__init__(self, name, vnfd, setup_env_helper_type,
                         resource_helper_type)
 
         self.start_test_time = None
@@ -150,9 +150,9 @@ class ProxIrqGen(ProxIrq, SampleVNFTrafficGen):
 
     APP_NAME = 'ProxIrqGen'
 
-    def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+    def __init__(self, name, vnfd, setup_env_helper_type=None,
                  resource_helper_type=None):
-        ProxIrq.__init__(self, name, vnfd, task_id, setup_env_helper_type,
+        ProxIrq.__init__(self, name, vnfd, setup_env_helper_type,
                                       resource_helper_type)
         self.start_test_time = None
         self.end_test_time = None
diff --git a/yardstick/network_services/vnf_generic/vnf/prox_vnf.py b/yardstick/network_services/vnf_generic/vnf/prox_vnf.py
index c3b50369b..0d1360dd0 100644
--- a/yardstick/network_services/vnf_generic/vnf/prox_vnf.py
+++ b/yardstick/network_services/vnf_generic/vnf/prox_vnf.py
@@ -35,8 +35,7 @@ class ProxApproxVnf(SampleVNF):
     VNF_PROMPT = "PROX started"
     LUA_PARAMETER_NAME = "sut"
 
-    def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
-                 resource_helper_type=None):
+    def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
         if setup_env_helper_type is None:
             setup_env_helper_type = ProxDpdkVnfSetupEnvHelper
 
@@ -47,8 +46,8 @@ class ProxApproxVnf(SampleVNF):
         self.prev_packets_sent = 0
         self.prev_tsc = 0
         self.tsc_hz = 0
-        super(ProxApproxVnf, self).__init__(
-            name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
+        super(ProxApproxVnf, self).__init__(name, vnfd, setup_env_helper_type,
+                                            resource_helper_type)
 
     def _vnf_up_post(self):
         self.resource_helper.up_post()
diff --git a/yardstick/network_services/vnf_generic/vnf/router_vnf.py b/yardstick/network_services/vnf_generic/vnf/router_vnf.py
index e99de9cb3..90b7b215e 100644
--- a/yardstick/network_services/vnf_generic/vnf/router_vnf.py
+++ b/yardstick/network_services/vnf_generic/vnf/router_vnf.py
@@ -34,8 +34,7 @@ class RouterVNF(SampleVNF):
 
     WAIT_TIME = 1
 
-    def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
-                 resource_helper_type=None):
+    def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
         if setup_env_helper_type is None:
             setup_env_helper_type = DpdkVnfSetupEnvHelper
 
@@ -43,8 +42,7 @@ class RouterVNF(SampleVNF):
         vnfd['mgmt-interface'].pop("pkey", "")
         vnfd['mgmt-interface']['password'] = 'password'
 
-        super(RouterVNF, self).__init__(
-            name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
+        super(RouterVNF, self).__init__(name, vnfd, setup_env_helper_type, resource_helper_type)
 
     def instantiate(self, scenario_cfg, context_cfg):
         self.scenario_helper.scenario_cfg = scenario_cfg
diff --git a/yardstick/network_services/vnf_generic/vnf/sample_vnf.py b/yardstick/network_services/vnf_generic/vnf/sample_vnf.py
index 8833b88f2..82a6eda33 100644
--- a/yardstick/network_services/vnf_generic/vnf/sample_vnf.py
+++ b/yardstick/network_services/vnf_generic/vnf/sample_vnf.py
@@ -18,7 +18,6 @@ from multiprocessing import Queue, Value, Process
 import os
 import posixpath
 import re
-import uuid
 import subprocess
 import time
 
@@ -397,13 +396,12 @@ class ClientResourceHelper(ResourceHelper):
         time.sleep(self.QUEUE_WAIT_TIME)
         self._queue.put(samples)
 
-    def run_traffic(self, traffic_profile, mq_producer):
+    def run_traffic(self, traffic_profile):
         # 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()
         # fixme: fix passing correct trex config file,
         # instead of searching the default path
-        mq_producer.tg_method_started()
         try:
             self._build_ports()
             self.client = self._connect()
@@ -411,12 +409,9 @@ class ClientResourceHelper(ResourceHelper):
             self.client.remove_all_streams(self.all_ports)  # remove all streams
             traffic_profile.register_generator(self)
 
-            iteration_index = 0
             while self._terminated.value == 0:
-                iteration_index += 1
                 if self._run_traffic_once(traffic_profile):
                     self._terminated.value = 1
-                mq_producer.tg_method_iteration(iteration_index)
 
             self.client.stop(self.all_ports)
             self.client.disconnect()
@@ -427,8 +422,6 @@ class ClientResourceHelper(ResourceHelper):
                 return  # return if trex/tg server is stopped.
             raise
 
-        mq_producer.tg_method_finished()
-
     def terminate(self):
         self._terminated.value = 1  # stop client
 
@@ -619,7 +612,6 @@ class ScenarioHelper(object):
         test_timeout = self.options.get('timeout', constants.DEFAULT_VNF_TIMEOUT)
         return test_duration if test_duration > test_timeout else test_timeout
 
-
 class SampleVNF(GenericVNF):
     """ Class providing file-like API for generic VNF implementation """
 
@@ -629,9 +621,8 @@ class SampleVNF(GenericVNF):
     APP_NAME = "SampleVNF"
     # we run the VNF interactively, so the ssh command will timeout after this long
 
-    def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
-                 resource_helper_type=None):
-        super(SampleVNF, self).__init__(name, vnfd, task_id)
+    def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
+        super(SampleVNF, self).__init__(name, vnfd)
         self.bin_path = get_nsb_option('bin_path', '')
 
         self.scenario_helper = ScenarioHelper(self.name)
@@ -862,9 +853,8 @@ class SampleVNFTrafficGen(GenericTrafficGen):
     APP_NAME = 'Sample'
     RUN_WAIT = 1
 
-    def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
-                 resource_helper_type=None):
-        super(SampleVNFTrafficGen, self).__init__(name, vnfd, task_id)
+    def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
+        super(SampleVNFTrafficGen, self).__init__(name, vnfd)
         self.bin_path = get_nsb_option('bin_path', '')
 
         self.scenario_helper = ScenarioHelper(self.name)
@@ -923,13 +913,12 @@ class SampleVNFTrafficGen(GenericTrafficGen):
                 LOG.info("%s TG Server is up and running.", self.APP_NAME)
                 return self._tg_process.exitcode
 
-    def _traffic_runner(self, traffic_profile, mq_id):
+    def _traffic_runner(self, traffic_profile):
         # always drop connections first thing in new processes
         # so we don't get paramiko errors
         self.ssh_helper.drop_connection()
         LOG.info("Starting %s client...", self.APP_NAME)
-        self._mq_producer = self._setup_mq_producer(mq_id)
-        self.resource_helper.run_traffic(traffic_profile, self._mq_producer)
+        self.resource_helper.run_traffic(traffic_profile)
 
     def run_traffic(self, traffic_profile):
         """ Generate traffic on the wire according to the given params.
@@ -939,12 +928,10 @@ class SampleVNFTrafficGen(GenericTrafficGen):
         :param traffic_profile:
         :return: True/False
         """
-        name = '{}-{}-{}-{}'.format(self.name, self.APP_NAME,
-                                    traffic_profile.__class__.__name__,
+        name = "{}-{}-{}-{}".format(self.name, self.APP_NAME, traffic_profile.__class__.__name__,
                                     os.getpid())
-        self._traffic_process = Process(
-            name=name, target=self._traffic_runner,
-            args=(traffic_profile, uuid.uuid1().int))
+        self._traffic_process = Process(name=name, target=self._traffic_runner,
+                                        args=(traffic_profile,))
         self._traffic_process.start()
         # Wait for traffic process to start
         while self.resource_helper.client_started.value == 0:
@@ -953,6 +940,8 @@ class SampleVNFTrafficGen(GenericTrafficGen):
             if not self._traffic_process.is_alive():
                 break
 
+        return self._traffic_process.is_alive()
+
     def collect_kpi(self):
         # check if the tg processes have exited
         physical_node = Context.get_physical_node_from_server(
diff --git a/yardstick/network_services/vnf_generic/vnf/tg_ixload.py b/yardstick/network_services/vnf_generic/vnf/tg_ixload.py
index d25402740..97b36d708 100644
--- a/yardstick/network_services/vnf_generic/vnf/tg_ixload.py
+++ b/yardstick/network_services/vnf_generic/vnf/tg_ixload.py
@@ -126,13 +126,12 @@ class IxLoadResourceHelper(sample_vnf.ClientResourceHelper):
 
 class IxLoadTrafficGen(sample_vnf.SampleVNFTrafficGen):
 
-    def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
-                 resource_helper_type=None):
+    def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
         if resource_helper_type is None:
             resource_helper_type = IxLoadResourceHelper
 
-        super(IxLoadTrafficGen, self).__init__(
-            name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
+        super(IxLoadTrafficGen, self).__init__(name, vnfd, setup_env_helper_type,
+                                               resource_helper_type)
         self._result = {}
 
     def update_gateways(self, links):
diff --git a/yardstick/network_services/vnf_generic/vnf/tg_landslide.py b/yardstick/network_services/vnf_generic/vnf/tg_landslide.py
index 2fba89b22..fb8c24752 100644
--- a/yardstick/network_services/vnf_generic/vnf/tg_landslide.py
+++ b/yardstick/network_services/vnf_generic/vnf/tg_landslide.py
@@ -35,11 +35,11 @@ LOG = logging.getLogger(__name__)
 class LandslideTrafficGen(sample_vnf.SampleVNFTrafficGen):
     APP_NAME = 'LandslideTG'
 
-    def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+    def __init__(self, name, vnfd, setup_env_helper_type=None,
                  resource_helper_type=None):
         if resource_helper_type is None:
             resource_helper_type = LandslideResourceHelper
-        super(LandslideTrafficGen, self).__init__(name, vnfd, task_id,
+        super(LandslideTrafficGen, self).__init__(name, vnfd,
                                                   setup_env_helper_type,
                                                   resource_helper_type)
 
diff --git a/yardstick/network_services/vnf_generic/vnf/tg_ping.py b/yardstick/network_services/vnf_generic/vnf/tg_ping.py
index a3b5afa39..a989543f5 100644
--- a/yardstick/network_services/vnf_generic/vnf/tg_ping.py
+++ b/yardstick/network_services/vnf_generic/vnf/tg_ping.py
@@ -71,7 +71,7 @@ class PingResourceHelper(ClientResourceHelper):
         self._queue = Queue()
         self._parser = PingParser(self._queue)
 
-    def run_traffic(self, traffic_profile, *args):
+    def run_traffic(self, traffic_profile):
         # drop the connection in order to force a new one
         self.ssh_helper.drop_connection()
 
@@ -103,14 +103,14 @@ class PingTrafficGen(SampleVNFTrafficGen):
     APP_NAME = 'Ping'
     RUN_WAIT = 4
 
-    def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
-                 resource_helper_type=None):
+    def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
         if setup_env_helper_type is None:
             setup_env_helper_type = PingSetupEnvHelper
         if resource_helper_type is None:
             resource_helper_type = PingResourceHelper
-        super(PingTrafficGen, self).__init__(
-            name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
+
+        super(PingTrafficGen, self).__init__(name, vnfd, setup_env_helper_type,
+                                             resource_helper_type)
         self._result = {}
 
     def _check_status(self):
diff --git a/yardstick/network_services/vnf_generic/vnf/tg_pktgen.py b/yardstick/network_services/vnf_generic/vnf/tg_pktgen.py
index 9d452213f..e73cebc40 100644
--- a/yardstick/network_services/vnf_generic/vnf/tg_pktgen.py
+++ b/yardstick/network_services/vnf_generic/vnf/tg_pktgen.py
@@ -13,9 +13,7 @@
 # limitations under the License.
 
 import logging
-import multiprocessing
 import time
-import uuid
 
 from yardstick.common import constants
 from yardstick.common import exceptions
@@ -26,8 +24,7 @@ from yardstick.network_services.vnf_generic.vnf import base as vnf_base
 LOG = logging.getLogger(__name__)
 
 
-class PktgenTrafficGen(vnf_base.GenericTrafficGen,
-                       vnf_base.GenericVNFEndpoint):
+class PktgenTrafficGen(vnf_base.GenericTrafficGen):
     """DPDK Pktgen traffic generator
 
     Website: http://pktgen-dpdk.readthedocs.io/en/latest/index.html
@@ -35,15 +32,8 @@ class PktgenTrafficGen(vnf_base.GenericTrafficGen,
 
     TIMEOUT = 30
 
-    def __init__(self, name, vnfd, task_id):
-        vnf_base.GenericTrafficGen.__init__(self, name, vnfd, task_id)
-        self.queue = multiprocessing.Queue()
-        self._id = uuid.uuid1().int
-        self._mq_producer = self._setup_mq_producer(self._id)
-        vnf_base.GenericVNFEndpoint.__init__(self, self._id, [task_id],
-                                             self.queue)
-        self._consumer = vnf_base.GenericVNFConsumer([task_id], self)
-        self._consumer.start_rpc_server()
+    def __init__(self, name, vnfd):
+        vnf_base.GenericTrafficGen.__init__(self, name, vnfd)
         self._traffic_profile = None
         self._node_ip = vnfd['mgmt-interface'].get('ip')
         self._lua_node_port = self._get_lua_node_port(
@@ -71,7 +61,7 @@ class PktgenTrafficGen(vnf_base.GenericTrafficGen,
     def wait_for_instantiate(self):  # pragma: no cover
         pass
 
-    def runner_method_start_iteration(self, ctxt, **kwargs):
+    def runner_method_start_iteration(self):
         # pragma: no cover
         LOG.debug('Start method')
         # NOTE(ralonsoh): 'rate' should be modified between iterations. The
@@ -81,11 +71,6 @@ class PktgenTrafficGen(vnf_base.GenericTrafficGen,
         self._traffic_profile.rate(self._rate)
         time.sleep(4)
         self._traffic_profile.stop()
-        self._mq_producer.tg_method_iteration(1, 1, {})
-
-    def runner_method_stop_iteration(self, ctxt, **kwargs):  # pragma: no cover
-        # pragma: no cover
-        LOG.debug('Stop method')
 
     @staticmethod
     def _get_lua_node_port(service_ports):
diff --git a/yardstick/network_services/vnf_generic/vnf/tg_prox.py b/yardstick/network_services/vnf_generic/vnf/tg_prox.py
index d12c42ec8..61ff2e036 100644
--- a/yardstick/network_services/vnf_generic/vnf/tg_prox.py
+++ b/yardstick/network_services/vnf_generic/vnf/tg_prox.py
@@ -29,13 +29,13 @@ class ProxTrafficGen(SampleVNFTrafficGen):
     LUA_PARAMETER_NAME = "gen"
     WAIT_TIME = 1
 
-    def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+    def __init__(self, name, vnfd, setup_env_helper_type=None,
                  resource_helper_type=None):
         vnfd_cpy = copy.deepcopy(vnfd)
-        super(ProxTrafficGen, self).__init__(name, vnfd_cpy, task_id)
+        super(ProxTrafficGen, self).__init__(name, vnfd_cpy)
 
         self._vnf_wrapper = ProxApproxVnf(
-            name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
+            name, vnfd, setup_env_helper_type, resource_helper_type)
         self.bin_path = get_nsb_option('bin_path', '')
         self.name = self._vnf_wrapper.name
         self.ssh_helper = self._vnf_wrapper.ssh_helper
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 ae95843a5..8f3698cb9 100644
--- a/yardstick/network_services/vnf_generic/vnf/tg_rfc2544_ixia.py
+++ b/yardstick/network_services/vnf_generic/vnf/tg_rfc2544_ixia.py
@@ -530,7 +530,7 @@ class IxiaResourceHelper(ClientResourceHelper):
         self._ix_scenario.apply_config()
         self._ix_scenario.create_traffic_model(traffic_profile)
 
-    def run_traffic(self, traffic_profile, *args):
+    def run_traffic(self, traffic_profile):
         if self._terminated.value:
             return
 
@@ -588,12 +588,12 @@ class IxiaTrafficGen(SampleVNFTrafficGen):
 
     APP_NAME = 'Ixia'
 
-    def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
-                 resource_helper_type=None):
+    def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
         if resource_helper_type is None:
             resource_helper_type = IxiaResourceHelper
-        super(IxiaTrafficGen, self).__init__(
-            name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
+
+        super(IxiaTrafficGen, self).__init__(name, vnfd, setup_env_helper_type,
+                                             resource_helper_type)
         self._ixia_traffic_gen = None
         self.ixia_file_name = ''
         self.vnf_port_pairs = []
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 7ecb12478..7c2079101 100644
--- a/yardstick/network_services/vnf_generic/vnf/tg_rfc2544_trex.py
+++ b/yardstick/network_services/vnf_generic/vnf/tg_rfc2544_trex.py
@@ -65,9 +65,9 @@ class TrexTrafficGenRFC(tg_trex.TrexTrafficGen):
     traffic for rfc2544 testcase.
     """
 
-    def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
-                 resource_helper_type=None):
+    def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
         if resource_helper_type is None:
             resource_helper_type = TrexRfcResourceHelper
-        super(TrexTrafficGenRFC, self).__init__(
-            name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
+
+        super(TrexTrafficGenRFC, self).__init__(name, vnfd, setup_env_helper_type,
+                                                resource_helper_type)
diff --git a/yardstick/network_services/vnf_generic/vnf/tg_trex.py b/yardstick/network_services/vnf_generic/vnf/tg_trex.py
index 4296da84c..43ee82629 100644
--- a/yardstick/network_services/vnf_generic/vnf/tg_trex.py
+++ b/yardstick/network_services/vnf_generic/vnf/tg_trex.py
@@ -200,14 +200,15 @@ class TrexTrafficGen(SampleVNFTrafficGen):
 
     APP_NAME = 'TRex'
 
-    def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
-                 resource_helper_type=None):
+    def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
         if resource_helper_type is None:
             resource_helper_type = TrexResourceHelper
+
         if setup_env_helper_type is None:
             setup_env_helper_type = TrexDpdkVnfSetupEnvHelper
-        super(TrexTrafficGen, self).__init__(
-            name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
+
+        super(TrexTrafficGen, self).__init__(name, vnfd, setup_env_helper_type,
+                                             resource_helper_type)
 
     def _check_status(self):
         return self.resource_helper.check_status()
diff --git a/yardstick/network_services/vnf_generic/vnf/udp_replay.py b/yardstick/network_services/vnf_generic/vnf/udp_replay.py
index e3fde1a79..fa92744d8 100644
--- a/yardstick/network_services/vnf_generic/vnf/udp_replay.py
+++ b/yardstick/network_services/vnf_generic/vnf/udp_replay.py
@@ -60,14 +60,15 @@ class UdpReplayApproxVnf(SampleVNF):
 
     PIPELINE_COMMAND = REPLAY_PIPELINE_COMMAND
 
-    def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
-                 resource_helper_type=None):
+    def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
         if resource_helper_type is None:
             resource_helper_type = UdpReplayResourceHelper
+
         if setup_env_helper_type is None:
             setup_env_helper_type = UdpReplaySetupEnvHelper
-        super(UdpReplayApproxVnf, self).__init__(
-            name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
+
+        super(UdpReplayApproxVnf, self).__init__(name, vnfd, setup_env_helper_type,
+                                                 resource_helper_type)
 
     def _build_pipeline_kwargs(self):
         ports = self.vnfd_helper.port_pairs.all_ports
diff --git a/yardstick/network_services/vnf_generic/vnf/vfw_vnf.py b/yardstick/network_services/vnf_generic/vnf/vfw_vnf.py
index a1523dee3..432f30a0c 100644
--- a/yardstick/network_services/vnf_generic/vnf/vfw_vnf.py
+++ b/yardstick/network_services/vnf_generic/vnf/vfw_vnf.py
@@ -52,9 +52,8 @@ class FWApproxVnf(SampleVNF):
         'packets_dropped': 3,
     }
 
-    def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
-                 resource_helper_type=None):
+    def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
         if setup_env_helper_type is None:
             setup_env_helper_type = FWApproxSetupEnvHelper
-        super(FWApproxVnf, self).__init__(
-            name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
+
+        super(FWApproxVnf, self).__init__(name, vnfd, setup_env_helper_type, resource_helper_type)
diff --git a/yardstick/network_services/vnf_generic/vnf/vpe_vnf.py b/yardstick/network_services/vnf_generic/vnf/vpe_vnf.py
index dd3221386..08fcb5e14 100644
--- a/yardstick/network_services/vnf_generic/vnf/vpe_vnf.py
+++ b/yardstick/network_services/vnf_generic/vnf/vpe_vnf.py
@@ -158,12 +158,11 @@ class VpeApproxVnf(SampleVNF):
     COLLECT_KPI = VPE_COLLECT_KPI
     WAIT_TIME = 20
 
-    def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
-                 resource_helper_type=None):
+    def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
         if setup_env_helper_type is None:
             setup_env_helper_type = VpeApproxSetupEnvHelper
-        super(VpeApproxVnf, self).__init__(
-            name, vnfd, task_id, setup_env_helper_type, resource_helper_type)
+
+        super(VpeApproxVnf, self).__init__(name, vnfd, setup_env_helper_type, resource_helper_type)
 
     def get_stats(self, *args, **kwargs):
         raise NotImplementedError