yardstick/benchmark/runners/base.py

   1 ##############################################################################
   2 # Copyright (c) 2015 Ericsson AB and others.
   3 #
   4 # All rights reserved. This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 ##############################################################################
   9
  10 import importlib
  11 import multiprocessing
  12 import json
  13 import logging
  14
  15 log = logging.getLogger(__name__)
  16
  17 import yardstick.common.utils as utils
  18 from yardstick.benchmark.scenarios import base as base_scenario
  19
  20
  21 def _output_serializer_main(filename, queue):
  22     '''entrypoint for the singleton subprocess writing to outfile
  23     Use of this process enables multiple instances of a scenario without
  24     messing up the output file.
  25     '''
  26     with open(filename, 'w') as outfile:
  27         while True:
  28             # blocks until data becomes available
  29             record = queue.get()
  30             if record == '_TERMINATE_':
  31                 outfile.close()
  32                 break
  33             else:
  34                 json.dump(record, outfile)
  35                 outfile.write('\n')
  36
  37
  38 class Runner(object):
  39     queue = None
  40     dump_process = None
  41     runners = []
  42
  43     @staticmethod
  44     def _get_cls(runner_type):
  45         for runner in utils.itersubclasses(Runner):
  46             if runner_type == runner.__execution_type__:
  47                 return runner
  48         raise RuntimeError("No such runner_type %s" % runner_type)
  49
  50     @staticmethod
  51     def get(config):
  52         """Returns instance of a scenario runner for execution type.
  53         """
  54         # if there is no runner, start the output serializer subprocess
  55         if len(Runner.runners) == 0:
  56             log.debug("Starting dump process file '%s'" %
  57                       config["output_filename"])
  58             Runner.queue = multiprocessing.Queue()
  59             Runner.dump_process = multiprocessing.Process(
  60                 target=_output_serializer_main,
  61                 name="Dumper",
  62                 args=(config["output_filename"], Runner.queue))
  63             Runner.dump_process.start()
  64
  65         return Runner._get_cls(config["type"])(config, Runner.queue)
  66
  67     @staticmethod
  68     def release(runner):
  69         '''Release the runner'''
  70         Runner.runners.remove(runner)
  71         # if this was the last runner, stop the output serializer subprocess
  72         if len(Runner.runners) == 0:
  73             log.debug("Stopping dump process")
  74             Runner.queue.put('_TERMINATE_')
  75             Runner.dump_process.join()
  76
  77     def __init__(self, config, queue):
  78         self.context = {}
  79         self.config = config
  80         self.result_queue = queue
  81         Runner.runners.append(self)
  82
  83     def run(self, scenario_type, scenario_args):
  84         class_name = base_scenario.Scenario.get(scenario_type)
  85         path_split = class_name.split(".")
  86         module_path = ".".join(path_split[:-1])
  87         module = importlib.import_module(module_path)
  88         cls = getattr(module, path_split[-1])
  89
  90         self.config['object'] = class_name
  91         self._run_benchmark(cls, "run", scenario_args)
  92
  93     def join(self):
  94         self.process.join()