import collections import faulthandler import json import os import queue import subprocess import sys import threading import time import traceback import types from test import support from test.libregrtest.runtest import ( runtest, INTERRUPTED, CHILD_ERROR, PROGRESS_MIN_TIME, format_test_result, TestResult, is_failed) from test.libregrtest.setup import setup_tests from test.libregrtest.utils import format_duration # Display the running tests if nothing happened last N seconds PROGRESS_UPDATE = 30.0 # seconds # Time to wait until a worker completes: should be immediate JOIN_TIMEOUT = 30.0 # seconds def must_stop(result, ns): if result.result == INTERRUPTED: return True if ns.failfast and is_failed(result, ns): return True return False def parse_worker_args(worker_args): ns_dict, test_name = json.loads(worker_args) ns = types.SimpleNamespace(**ns_dict) return (ns, test_name) def run_test_in_subprocess(testname, ns): ns_dict = vars(ns) worker_args = (ns_dict, testname) worker_args = json.dumps(worker_args) cmd = [sys.executable, *support.args_from_interpreter_flags(), '-u', # Unbuffered stdout and stderr '-m', 'test.regrtest', '--worker-args', worker_args] # Running the child from the same working directory as regrtest's original # invocation ensures that TEMPDIR for the child is the same when # sysconfig.is_python_build() is true. See issue 15300. return subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, close_fds=(os.name != 'nt'), cwd=support.SAVEDCWD) def run_tests_worker(ns, test_name): setup_tests(ns) result = runtest(ns, test_name) print() # Force a newline (just in case) # Serialize TestResult as list in JSON print(json.dumps(list(result)), flush=True) sys.exit(0) # We do not use a generator so multiple threads can call next(). class MultiprocessIterator: """A thread-safe iterator over tests for multiprocess mode.""" def __init__(self, tests_iter): self.lock = threading.Lock() self.tests_iter = tests_iter def __iter__(self): return self def __next__(self): with self.lock: if self.tests_iter is None: raise StopIteration return next(self.tests_iter) def stop(self): with self.lock: self.tests_iter = None MultiprocessResult = collections.namedtuple('MultiprocessResult', 'result stdout stderr error_msg') class ExitThread(Exception): pass class MultiprocessThread(threading.Thread): def __init__(self, pending, output, ns): super().__init__() self.pending = pending self.output = output self.ns = ns self.current_test_name = None self.start_time = None self._popen = None self._killed = False def __repr__(self): info = ['MultiprocessThread'] test = self.current_test_name if self.is_alive(): info.append('alive') if test: info.append(f'test={test}') popen = self._popen if popen: info.append(f'pid={popen.pid}') return '<%s>' % ' '.join(info) def kill(self): self._killed = True popen = self._popen if popen is None: return popen.kill() # stdout and stderr must be closed to ensure that communicate() # does not hang popen.stdout.close() popen.stderr.close() def _runtest(self, test_name): try: self.start_time = time.monotonic() self.current_test_name = test_name self._popen = run_test_in_subprocess(test_name, self.ns) popen = self._popen with popen: try: if self._killed: # If kill() has been called before self._popen is set, # self._popen is still running. Call again kill() # to ensure that the process is killed. self.kill() raise ExitThread try: stdout, stderr = popen.communicate() except OSError: if self._killed: # kill() has been called: communicate() fails # on reading closed stdout/stderr raise ExitThread raise except: self.kill() popen.wait() raise retcode = popen.wait() finally: self.current_test_name = None self._popen = None stdout = stdout.strip() stderr = stderr.rstrip() err_msg = None if retcode != 0: err_msg = "Exit code %s" % retcode else: stdout, _, result = stdout.rpartition("\n") stdout = stdout.rstrip() if not result: err_msg = "Failed to parse worker stdout" else: try: # deserialize run_tests_worker() output result = json.loads(result) result = TestResult(*result) except Exception as exc: err_msg = "Failed to parse worker JSON: %s" % exc if err_msg is not None: test_time = time.monotonic() - self.start_time result = TestResult(test_name, CHILD_ERROR, test_time, None) return MultiprocessResult(result, stdout, stderr, err_msg) def run(self): while not self._killed: try: try: test_name = next(self.pending) except StopIteration: break mp_result = self._runtest(test_name) self.output.put((False, mp_result)) if must_stop(mp_result.result, self.ns): break except ExitThread: break except BaseException: self.output.put((True, traceback.format_exc())) break def get_running(workers): running = [] for worker in workers: current_test_name = worker.current_test_name if not current_test_name: continue dt = time.monotonic() - worker.start_time if dt >= PROGRESS_MIN_TIME: text = '%s (%s)' % (current_test_name, format_duration(dt)) running.append(text) return running class MultiprocessRunner: def __init__(self, regrtest): self.regrtest = regrtest self.ns = regrtest.ns self.output = queue.Queue() self.pending = MultiprocessIterator(self.regrtest.tests) if self.ns.timeout is not None: self.test_timeout = self.ns.timeout * 1.5 else: self.test_timeout = None self.workers = None def start_workers(self): self.workers = [MultiprocessThread(self.pending, self.output, self.ns) for _ in range(self.ns.use_mp)] print("Run tests in parallel using %s child processes" % len(self.workers)) for worker in self.workers: worker.start() def wait_workers(self): start_time = time.monotonic() for worker in self.workers: worker.kill() for worker in self.workers: while True: worker.join(1.0) if not worker.is_alive(): break dt = time.monotonic() - start_time print("Wait for regrtest worker %r for %.1f sec" % (worker, dt)) if dt > JOIN_TIMEOUT: print("Warning -- failed to join a regrtest worker %s" % worker) break def _get_result(self): if not any(worker.is_alive() for worker in self.workers): # all worker threads are done: consume pending results try: return self.output.get(timeout=0) except queue.Empty: return None while True: if self.test_timeout is not None: faulthandler.dump_traceback_later(self.test_timeout, exit=True) # wait for a thread timeout = max(PROGRESS_UPDATE, PROGRESS_MIN_TIME) try: return self.output.get(timeout=timeout) except queue.Empty: pass # display progress running = get_running(self.workers) if running and not self.ns.pgo: print('running: %s' % ', '.join(running), flush=True) def display_result(self, mp_result): result = mp_result.result text = format_test_result(result) if mp_result.error_msg is not None: # CHILD_ERROR text += ' (%s)' % mp_result.error_msg elif (result.test_time >= PROGRESS_MIN_TIME and not self.ns.pgo): text += ' (%s)' % format_duration(result.test_time) running = get_running(self.workers) if running and not self.ns.pgo: text += ' -- running: %s' % ', '.join(running) self.regrtest.display_progress(self.test_index, text) def _process_result(self, item): if item[0]: # Thread got an exception format_exc = item[1] print(f"regrtest worker thread failed: {format_exc}", file=sys.stderr, flush=True) return True self.test_index += 1 mp_result = item[1] self.regrtest.accumulate_result(mp_result.result) self.display_result(mp_result) if mp_result.stdout: print(mp_result.stdout, flush=True) if mp_result.stderr and not self.ns.pgo: print(mp_result.stderr, file=sys.stderr, flush=True) if must_stop(mp_result.result, self.ns): return True return False def run_tests(self): self.start_workers() self.test_index = 0 try: while True: item = self._get_result() if item is None: break stop = self._process_result(item) if stop: break except KeyboardInterrupt: print() self.regrtest.interrupted = True finally: if self.test_timeout is not None: faulthandler.cancel_dump_traceback_later() # a test failed (and --failfast is set) or all tests completed self.pending.stop() self.wait_workers() def run_tests_multiprocess(regrtest): MultiprocessRunner(regrtest).run_tests()