diff options
Diffstat (limited to 'vnftest/runners/dynamictp.py')
| -rwxr-xr-x | vnftest/runners/dynamictp.py | 179 |
1 files changed, 179 insertions, 0 deletions
diff --git a/vnftest/runners/dynamictp.py b/vnftest/runners/dynamictp.py new file mode 100755 index 0000000..e394567 --- /dev/null +++ b/vnftest/runners/dynamictp.py @@ -0,0 +1,179 @@ +############################################################################## +# Copyright 2018 EuropeanSoftwareMarketingLtd. +# =================================================================== +# Licensed under the ApacheLicense, Version2.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 +# +# 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 +############################################################################## +# vnftest comment: this is a modified copy of +# rally/rally/step/runners/dynamictp.py + +"""A runner that searches for the max throughput with binary search +""" + +import logging +import multiprocessing +import time +import traceback + +import os + +from vnftest.runners import base + +LOG = logging.getLogger(__name__) + + +def _worker_process(queue, cls, method_name, step_cfg, + context_cfg, aborted): # pragma: no cover + + runner_cfg = step_cfg['runner'] + iterations = runner_cfg.get("iterations", 1) + interval = runner_cfg.get("interval", 1) + run_step = runner_cfg.get("run_step", "setup,run,teardown") + delta = runner_cfg.get("delta", 1000) + options_cfg = step_cfg['options'] + initial_rate = options_cfg.get("pps", 1000000) + LOG.info("worker START, class %s", cls) + + runner_cfg['runner_id'] = os.getpid() + + step = cls(step_cfg, context_cfg) + if "setup" in run_step: + step.setup() + + method = getattr(step, method_name) + + queue.put({'runner_id': runner_cfg['runner_id'], + 'step_cfg': step_cfg, + 'context_cfg': context_cfg}) + + if "run" in run_step: + iterator = 0 + search_max = initial_rate + search_min = 0 + while iterator < iterations: + search_min = int(search_min / 2) + step_cfg['options']['pps'] = search_max + search_max_found = False + max_throuput_found = False + sequence = 0 + + last_min_data = {'packets_per_second': 0} + + while True: + sequence += 1 + + data = {} + errors = "" + too_high = False + + LOG.debug("sequence: %s search_min: %s search_max: %s", + sequence, search_min, search_max) + + try: + method(data) + except AssertionError as assertion: + LOG.warning("SLA validation failed: %s" % assertion.args) + too_high = True + except Exception as e: + errors = traceback.format_exc() + LOG.exception(e) + + actual_pps = data['packets_per_second'] + + if too_high: + search_max = actual_pps + + if not search_max_found: + search_max_found = True + else: + last_min_data = data + search_min = actual_pps + + # Check if the actual rate is well below the asked rate + if step_cfg['options']['pps'] > actual_pps * 1.5: + search_max = actual_pps + LOG.debug("Sender reached max tput: %s", search_max) + elif not search_max_found: + search_max = int(actual_pps * 1.5) + + if ((search_max - search_min) < delta) or \ + (search_max <= search_min) or (10 <= sequence): + if last_min_data['packets_per_second'] > 0: + data = last_min_data + + step_output = { + 'timestamp': time.time(), + 'sequence': sequence, + 'data': data, + 'errors': errors + } + + record = { + 'runner_id': runner_cfg['runner_id'], + 'step': step_output + } + + queue.put(record) + max_throuput_found = True + + if errors or aborted.is_set() or max_throuput_found: + LOG.info("worker END") + break + + if not search_max_found: + step_cfg['options']['pps'] = search_max + else: + step_cfg['options']['pps'] = \ + (search_max - search_min) / 2 + search_min + + time.sleep(interval) + + iterator += 1 + LOG.debug("iterator: %s iterations: %s", iterator, iterations) + + if "teardown" in run_step: + try: + step.teardown() + except Exception: + # catch any exception in teardown and convert to simple exception + # never pass exceptions back to multiprocessing, because some exceptions can + # be unpicklable + # https://bugs.python.org/issue9400 + LOG.exception("") + raise SystemExit(1) + + LOG.debug("queue.qsize() = %s", queue.qsize()) + + +class IterationRunner(base.Runner): + """Run a step to find the max throughput + +If the step ends before the time has elapsed, it will be started again. + + Parameters + interval - time to wait between each step invocation + type: int + unit: seconds + default: 1 sec + delta - stop condition for the search. + type: int + unit: pps + default: 1000 pps + """ + __execution_type__ = 'Dynamictp' + + def _run_step(self, cls, method, step_cfg, context_cfg): + name = "{}-{}-{}".format(self.__execution_type__, step_cfg.get("type"), os.getpid()) + self.process = multiprocessing.Process( + name=name, + target=_worker_process, + args=(self.result_queue, cls, method, step_cfg, + context_cfg, self.aborted)) + self.process.start() |