commit 4fe31c5fa8444fed4ccb2d5c47a5191ab0f1b730 Author: Damian Johnson atagar@torproject.org Date: Sun Jun 4 11:10:57 2017 -0700
Run asychronous tests in a subprocess
Fully parallelizing asynchronously run tests to avoid the GIL. This doesn't have a noticeable impact on our installation runtime (it causes the install test to be slower, but the overall runtime to be the same). That said, in theory should benefit others. I'll probably experiment back and forth between this and threads as we expand the tests which do this. --- run_tests.py | 10 +++------- stem/util/test_tools.py | 48 ++++++++++++++++++++++++++++++---------------- test/integ/installation.py | 9 ++++++--- 3 files changed, 40 insertions(+), 27 deletions(-)
diff --git a/run_tests.py b/run_tests.py index 49f8465..a174f65 100755 --- a/run_tests.py +++ b/run_tests.py @@ -232,6 +232,9 @@ def main():
skipped_tests = 0
+ if args.run_integ and (not args.specific_test or 'test.integ.installation'.startswith(args.specific_test)): + test.integ.installation.setup() + if args.run_unit: test.output.print_divider('UNIT TESTS', True) error_tracker.set_category('UNIT TEST') @@ -251,10 +254,6 @@ def main():
our_version = stem.version.get_system_tor_version(args.tor_path) skipped_targets = {} - integ_setup_thread = None - - if not args.specific_test or 'test.integ.installation'.startswith(args.specific_test): - integ_setup_thread = test.integ.installation.setup()
for target in args.run_targets: # check if we meet this target's tor version prerequisites @@ -286,9 +285,6 @@ def main(): # We should have joined on all threads. If not then that indicates a # leak that could both likely be a bug and disrupt further targets.
- if integ_setup_thread: - integ_setup_thread.join() - active_threads = threading.enumerate()
if len(active_threads) > 1: diff --git a/stem/util/test_tools.py b/stem/util/test_tools.py index c9912c0..47fc02a 100644 --- a/stem/util/test_tools.py +++ b/stem/util/test_tools.py @@ -30,6 +30,7 @@ to match just against the prefix or suffix. For instance...
import collections import linecache +import multiprocessing import os import re import time @@ -59,31 +60,44 @@ class AsyncTestResult(object): Test results that can be applied later. """
- def __init__(self, runner): - self._failure = None - self._skip_reason = None - - def _wrapper(): + def __init__(self, test_runner, *test_runner_args): + def _wrapper(conn, runner, args): try: - runner() + runner(*args) if args else runner() + conn.send(('success', None)) except AssertionError as exc: - self._failure = str(exc) + conn.send(('failure', str(exc))) except SkipTest as exc: - self._skip_reason = str(exc) + conn.send(('skipped', str(exc))) + finally: + conn.close() + + self._result_type, self._result_msg = None, None + self._result_lock = threading.RLock() + self._results_pipe, child_pipe = multiprocessing.Pipe() + self._test_process = multiprocessing.Process(target = _wrapper, args = (child_pipe, test_runner, test_runner_args)) + self._test_process.start()
- self._thread = threading.Thread(target = _wrapper) - self._thread.start() + def pid(self): + with self._result_lock: + return self._test_process.pid if self._test_process else None
def join(self): - self._thread.join() + self.result(None)
def result(self, test): - self.join() - - if self._failure: - test.fail(self._failure) - elif self._skip_reason: - test.skipTest(self._skip_reason) + with self._result_lock: + if self._test_process: + self._result_type, self._result_msg = self._results_pipe.recv() + self._test_process.join() + self._test_process = None + + if not test: + return + elif self._result_type == 'failure': + test.fail(self._result_msg) + elif self._result_type == 'skipped': + test.skipTest(self._result_msg)
class Issue(collections.namedtuple('Issue', ['line_number', 'message', 'line'])): diff --git a/test/integ/installation.py b/test/integ/installation.py index da918a4..8b39e9d 100644 --- a/test/integ/installation.py +++ b/test/integ/installation.py @@ -7,6 +7,7 @@ import os import shutil import sys import tarfile +import time import unittest
import stem @@ -29,7 +30,7 @@ def setup(): global TEST_INSTALL, TEST_SDIST
TEST_INSTALL = stem.util.test_tools.AsyncTestResult(_test_install) - TEST_SDIST = stem.util.test_tools.AsyncTestResult(_test_sdist) + TEST_SDIST = stem.util.test_tools.AsyncTestResult(_test_sdist, TEST_INSTALL.pid())
def _test_install(): @@ -56,8 +57,10 @@ def _test_install(): shutil.rmtree(BASE_INSTALL_PATH)
-def _test_sdist(): - TEST_INSTALL.join() # we need to run these tests serially +def _test_sdist(dependency_pid): + while stem.util.system.is_running(dependency_pid): + time.sleep(0.1) # we need to run these tests serially + git_dir = os.path.join(test.STEM_BASE, '.git')
if not stem.util.system.is_available('git'):