diff options
| author |   Ross Brattain <ross.b.brattain@intel.com> | 2017-10-01 20:26:45 +0000 |
|---|---|---|
| committer |   Gerrit Code Review <gerrit@opnfv.org> | 2017-10-01 20:26:45 +0000 |
| commit | 6546e0467013abd5fff236b6e9d7137b292a4d4e (patch) | |
| tree | 4aa9eafe807b3d3d3d66aaf5cdfd7f4280c4c87e | |
| parent | 75c0e3a54b8f6e8fd77c7d9d95decab830159929 (diff) | |
| parent | 7eb6abb6931b24e085b139cc3500f4497cdde57d (diff) | |
Merge "duration runner: add teardown and cancel all queue join threads"
| -rw-r--r-- | yardstick/benchmark/runners/duration.py | 25 |
1 files changed, 20 insertions, 5 deletions
diff --git a/yardstick/benchmark/runners/duration.py b/yardstick/benchmark/runners/duration.py index c2c6a8f19..6a09131e1 100644 --- a/yardstick/benchmark/runners/duration.py +++ b/yardstick/benchmark/runners/duration.py @@ -36,6 +36,11 @@ def _worker_process(queue, cls, method_name, scenario_cfg, sequence = 1 + # if we don't do this we can hang waiting for the queue to drain + # have to do this in the subprocess + queue.cancel_join_thread() + output_queue.cancel_join_thread() + runner_cfg = scenario_cfg['runner'] interval = runner_cfg.get("interval", 1) @@ -54,6 +59,7 @@ def _worker_process(queue, cls, method_name, scenario_cfg, sla_action = scenario_cfg["sla"].get("action", "assert") start = time.time() + timeout = start + duration while True: LOG.debug("runner=%(runner)s seq=%(sequence)s START", @@ -71,9 +77,11 @@ def _worker_process(queue, cls, method_name, scenario_cfg, elif sla_action == "monitor": LOG.warning("SLA validation failed: %s", assertion.args) errors = assertion.args - except Exception as e: + # catch all exceptions because with multiprocessing we can have un-picklable exception + # problems https://bugs.python.org/issue9400 + except Exception: errors = traceback.format_exc() - LOG.exception(e) + LOG.exception("") else: if result: output_queue.put(result) @@ -94,12 +102,19 @@ def _worker_process(queue, cls, method_name, scenario_cfg, sequence += 1 - if (errors and sla_action is None) or \ - (time.time() - start > duration or aborted.is_set()): + if (errors and sla_action is None) or time.time() > timeout or aborted.is_set(): LOG.info("Worker END") break - benchmark.teardown() + try: + benchmark.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) class DurationRunner(base.Runner): |