1 ##############################################################################
2 # Copyright (c) 2015 Ericsson AB and others.
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 ##############################################################################
12 import multiprocessing
17 log = logging.getLogger(__name__)
19 from oslo_config import cfg
21 import yardstick.common.utils as utils
22 from yardstick.benchmark.scenarios import base as base_scenario
23 from yardstick.dispatcher.base import Base as DispatcherBase
28 def _output_serializer_main(filename, queue):
29 '''entrypoint for the singleton subprocess writing to outfile
30 Use of this process enables multiple instances of a scenario without
31 messing up the output file.
34 config["type"] = CONF.dispatcher.capitalize()
35 config["file_path"] = filename
36 dispatcher = DispatcherBase.get(config)
39 # blocks until data becomes available
41 if record == '_TERMINATE_':
44 dispatcher.record_result_data(record)
47 def _execute_shell_command(command):
48 '''execute shell script with error handling'''
52 output = subprocess.check_output(command, shell=True)
55 output = traceback.format_exc()
56 log.error("exec command '%s' error:\n " % command)
57 log.error(traceback.format_exc())
59 return exitcode, output
62 def _single_action(seconds, command, queue):
63 '''entrypoint for the single action process'''
64 log.debug("single action, fires after %d seconds (from now)", seconds)
66 log.debug("single action: executing command: '%s'", command)
67 ret_code, data = _execute_shell_command(command)
69 log.error("single action error! command:%s" % command)
70 queue.put({'single-action-data': data})
72 log.debug("single action data: \n%s" % data)
73 queue.put({'single-action-data': data})
76 def _periodic_action(interval, command, queue):
77 '''entrypoint for the periodic action process'''
78 log.debug("periodic action, fires every: %d seconds", interval)
82 time_spent += interval
83 log.debug("periodic action, executing command: '%s'", command)
84 ret_code, data = _execute_shell_command(command)
86 log.error("periodic action error! command:%s", command)
87 queue.put({'periodic-action-data': data})
89 log.debug("periodic action data: \n%s" % data)
90 queue.put({'periodic-action-data': data})
99 def get_cls(runner_type):
100 '''return class of specified type'''
101 for runner in utils.itersubclasses(Runner):
102 if runner_type == runner.__execution_type__:
104 raise RuntimeError("No such runner_type %s" % runner_type)
108 '''return a list of known runner type (class) names'''
110 for runner in utils.itersubclasses(Runner):
116 """Returns instance of a scenario runner for execution type.
118 # if there is no runner, start the output serializer subprocess
119 if len(Runner.runners) == 0:
120 log.debug("Starting dump process file '%s'" %
121 config["output_filename"])
122 Runner.queue = multiprocessing.Queue()
123 Runner.dump_process = multiprocessing.Process(
124 target=_output_serializer_main,
126 args=(config["output_filename"], Runner.queue))
127 Runner.dump_process.start()
129 return Runner.get_cls(config["type"])(config, Runner.queue)
132 def release_dump_process():
133 '''Release the dumper process'''
134 log.debug("Stopping dump process")
135 if Runner.dump_process:
136 Runner.queue.put('_TERMINATE_')
137 Runner.dump_process.join()
138 Runner.dump_process = None
142 '''Release the runner'''
143 Runner.runners.remove(runner)
145 # if this was the last runner, stop the output serializer subprocess
146 if len(Runner.runners) == 0:
147 Runner.release_dump_process()
151 '''Terminate all runners (subprocesses)'''
152 log.debug("Terminating all runners")
154 # release dumper process as some errors before any runner is created
155 if len(Runner.runners) == 0:
156 Runner.release_dump_process()
159 for runner in Runner.runners:
160 log.debug("Terminating runner: %s", runner)
162 runner.process.terminate()
163 runner.process.join()
164 if runner.periodic_action_process:
165 log.debug("Terminating periodic action process")
166 runner.periodic_action_process.terminate()
167 runner.periodic_action_process = None
168 Runner.release(runner)
170 def __init__(self, config, queue):
173 self.periodic_action_process = None
174 self.result_queue = queue
176 Runner.runners.append(self)
178 def run_post_stop_action(self):
179 '''run a potentially configured post-stop action'''
180 if "post-stop-action" in self.config:
181 command = self.config["post-stop-action"]["command"]
182 log.debug("post stop action: command: '%s'" % command)
183 ret_code, data = _execute_shell_command(command)
185 log.error("post action error! command:%s", command)
186 self.result_queue.put({'post-stop-action-data': data})
188 log.debug("post-stop data: \n%s" % data)
189 self.result_queue.put({'post-stop-action-data': data})
191 def run(self, scenario_type, scenario_cfg):
192 class_name = base_scenario.Scenario.get(scenario_type)
193 path_split = class_name.split(".")
194 module_path = ".".join(path_split[:-1])
195 module = importlib.import_module(module_path)
196 cls = getattr(module, path_split[-1])
198 self.config['object'] = class_name
200 # run a potentially configured pre-start action
201 if "pre-start-action" in self.config:
202 command = self.config["pre-start-action"]["command"]
203 log.debug("pre start action: command: '%s'" % command)
204 ret_code, data = _execute_shell_command(command)
206 log.error("pre-start action error! command:%s", command)
207 self.result_queue.put({'pre-start-action-data': data})
209 log.debug("pre-start data: \n%s" % data)
210 self.result_queue.put({'pre-start-action-data': data})
212 if "single-shot-action" in self.config:
213 single_action_process = multiprocessing.Process(
214 target=_single_action,
215 name="single-shot-action",
216 args=(self.config["single-shot-action"]["after"],
217 self.config["single-shot-action"]["command"],
219 single_action_process.start()
221 if "periodic-action" in self.config:
222 self.periodic_action_process = multiprocessing.Process(
223 target=_periodic_action,
224 name="periodic-action",
225 args=(self.config["periodic-action"]["interval"],
226 self.config["periodic-action"]["command"],
228 self.periodic_action_process.start()
230 self._run_benchmark(cls, "run", scenario_cfg)
234 if self.periodic_action_process:
235 self.periodic_action_process.terminate()
236 self.periodic_action_process = None
238 self.run_post_stop_action()
239 return self.process.exitcode