Python code coverage for Lib/test/libregrtest/runtest_mp.py

#	count	content
1	n/a	import faulthandler
2	n/a	import json
3	n/a	import os
4	n/a	import queue
5	n/a	import sys
6	n/a	import time
7	n/a	import traceback
8	n/a	import types
9	n/a	from test import support
10	n/a	try:
11	n/a	import threading
12	n/a	except ImportError:
13	n/a	print("Multiprocess option requires thread support")
14	n/a	sys.exit(2)
15	n/a
16	n/a	from test.libregrtest.runtest import (
17	n/a	runtest, INTERRUPTED, CHILD_ERROR, PROGRESS_MIN_TIME,
18	n/a	format_test_result)
19	n/a	from test.libregrtest.setup import setup_tests
20	n/a
21	n/a
22	n/a	# Display the running tests if nothing happened last N seconds
23	n/a	PROGRESS_UPDATE = 30.0 # seconds
24	n/a
25	n/a	# If interrupted, display the wait progress every N seconds
26	n/a	WAIT_PROGRESS = 2.0 # seconds
27	n/a
28	n/a
29	n/a	def run_test_in_subprocess(testname, ns):
30	n/a	"""Run the given test in a subprocess with --slaveargs.
31	n/a
32	n/a	ns is the option Namespace parsed from command-line arguments. regrtest
33	n/a	is invoked in a subprocess with the --slaveargs argument; when the
34	n/a	subprocess exits, its return code, stdout and stderr are returned as a
35	n/a	3-tuple.
36	n/a	"""
37	n/a	from subprocess import Popen, PIPE
38	n/a
39	n/a	ns_dict = vars(ns)
40	n/a	slaveargs = (ns_dict, testname)
41	n/a	slaveargs = json.dumps(slaveargs)
42	n/a
43	n/a	cmd = [sys.executable, *support.args_from_interpreter_flags(),
44	n/a	'-u', # Unbuffered stdout and stderr
45	n/a	'-m', 'test.regrtest',
46	n/a	'--slaveargs', slaveargs]
47	n/a	if ns.pgo:
48	n/a	cmd += ['--pgo']
49	n/a
50	n/a	# Running the child from the same working directory as regrtest's original
51	n/a	# invocation ensures that TEMPDIR for the child is the same when
52	n/a	# sysconfig.is_python_build() is true. See issue 15300.
53	n/a	popen = Popen(cmd,
54	n/a	stdout=PIPE, stderr=PIPE,
55	n/a	universal_newlines=True,
56	n/a	close_fds=(os.name != 'nt'),
57	n/a	cwd=support.SAVEDCWD)
58	n/a	with popen:
59	n/a	stdout, stderr = popen.communicate()
60	n/a	retcode = popen.wait()
61	n/a	return retcode, stdout, stderr
62	n/a
63	n/a
64	n/a	def run_tests_slave(slaveargs):
65	n/a	ns_dict, testname = json.loads(slaveargs)
66	n/a	ns = types.SimpleNamespace(**ns_dict)
67	n/a
68	n/a	setup_tests(ns)
69	n/a
70	n/a	try:
71	n/a	result = runtest(ns, testname)
72	n/a	except KeyboardInterrupt:
73	n/a	result = INTERRUPTED, ''
74	n/a	except BaseException as e:
75	n/a	traceback.print_exc()
76	n/a	result = CHILD_ERROR, str(e)
77	n/a
78	n/a	print() # Force a newline (just in case)
79	n/a	print(json.dumps(result), flush=True)
80	n/a	sys.exit(0)
81	n/a
82	n/a
83	n/a	# We do not use a generator so multiple threads can call next().
84	n/a	class MultiprocessIterator:
85	n/a
86	n/a	"""A thread-safe iterator over tests for multiprocess mode."""
87	n/a
88	n/a	def __init__(self, tests):
89	n/a	self.interrupted = False
90	n/a	self.lock = threading.Lock()
91	n/a	self.tests = tests
92	n/a
93	n/a	def __iter__(self):
94	n/a	return self
95	n/a
96	n/a	def __next__(self):
97	n/a	with self.lock:
98	n/a	if self.interrupted:
99	n/a	raise StopIteration('tests interrupted')
100	n/a	return next(self.tests)
101	n/a
102	n/a
103	n/a	class MultiprocessThread(threading.Thread):
104	n/a	def __init__(self, pending, output, ns):
105	n/a	super().__init__()
106	n/a	self.pending = pending
107	n/a	self.output = output
108	n/a	self.ns = ns
109	n/a	self.current_test = None
110	n/a	self.start_time = None
111	n/a
112	n/a	def _runtest(self):
113	n/a	try:
114	n/a	test = next(self.pending)
115	n/a	except StopIteration:
116	n/a	self.output.put((None, None, None, None))
117	n/a	return True
118	n/a
119	n/a	try:
120	n/a	self.start_time = time.monotonic()
121	n/a	self.current_test = test
122	n/a
123	n/a	retcode, stdout, stderr = run_test_in_subprocess(test, self.ns)
124	n/a	finally:
125	n/a	self.current_test = None
126	n/a
127	n/a	stdout, _, result = stdout.strip().rpartition("\n")
128	n/a	if retcode != 0:
129	n/a	result = (CHILD_ERROR, "Exit code %s" % retcode)
130	n/a	self.output.put((test, stdout.rstrip(), stderr.rstrip(),
131	n/a	result))
132	n/a	return False
133	n/a
134	n/a	if not result:
135	n/a	self.output.put((None, None, None, None))
136	n/a	return True
137	n/a
138	n/a	result = json.loads(result)
139	n/a	self.output.put((test, stdout.rstrip(), stderr.rstrip(),
140	n/a	result))
141	n/a	return False
142	n/a
143	n/a	def run(self):
144	n/a	try:
145	n/a	stop = False
146	n/a	while not stop:
147	n/a	stop = self._runtest()
148	n/a	except BaseException:
149	n/a	self.output.put((None, None, None, None))
150	n/a	raise
151	n/a
152	n/a
153	n/a	def run_tests_multiprocess(regrtest):
154	n/a	output = queue.Queue()
155	n/a	pending = MultiprocessIterator(regrtest.tests)
156	n/a	test_timeout = regrtest.ns.timeout
157	n/a	use_timeout = (test_timeout is not None)
158	n/a
159	n/a	workers = [MultiprocessThread(pending, output, regrtest.ns)
160	n/a	for i in range(regrtest.ns.use_mp)]
161	n/a	print("Run tests in parallel using %s child processes"
162	n/a	% len(workers))
163	n/a	for worker in workers:
164	n/a	worker.start()
165	n/a
166	n/a	def get_running(workers):
167	n/a	running = []
168	n/a	for worker in workers:
169	n/a	current_test = worker.current_test
170	n/a	if not current_test:
171	n/a	continue
172	n/a	dt = time.monotonic() - worker.start_time
173	n/a	if dt >= PROGRESS_MIN_TIME:
174	n/a	running.append('%s (%.0f sec)' % (current_test, dt))
175	n/a	return running
176	n/a
177	n/a	finished = 0
178	n/a	test_index = 1
179	n/a	get_timeout = max(PROGRESS_UPDATE, PROGRESS_MIN_TIME)
180	n/a	try:
181	n/a	while finished < regrtest.ns.use_mp:
182	n/a	if use_timeout:
183	n/a	faulthandler.dump_traceback_later(test_timeout, exit=True)
184	n/a
185	n/a	try:
186	n/a	item = output.get(timeout=get_timeout)
187	n/a	except queue.Empty:
188	n/a	running = get_running(workers)
189	n/a	if running and not regrtest.ns.pgo:
190	n/a	print('running: %s' % ', '.join(running))
191	n/a	continue
192	n/a
193	n/a	test, stdout, stderr, result = item
194	n/a	if test is None:
195	n/a	finished += 1
196	n/a	continue
197	n/a	regrtest.accumulate_result(test, result)
198	n/a
199	n/a	# Display progress
200	n/a	ok, test_time = result
201	n/a	text = format_test_result(test, ok)
202	n/a	if (ok not in (CHILD_ERROR, INTERRUPTED)
203	n/a	and test_time >= PROGRESS_MIN_TIME
204	n/a	and not regrtest.ns.pgo):
205	n/a	text += ' (%.0f sec)' % test_time
206	n/a	elif ok == CHILD_ERROR:
207	n/a	text = '%s (%s)' % (text, test_time)
208	n/a	running = get_running(workers)
209	n/a	if running and not regrtest.ns.pgo:
210	n/a	text += ' -- running: %s' % ', '.join(running)
211	n/a	regrtest.display_progress(test_index, text)
212	n/a
213	n/a	# Copy stdout and stderr from the child process
214	n/a	if stdout:
215	n/a	print(stdout, flush=True)
216	n/a	if stderr and not regrtest.ns.pgo:
217	n/a	print(stderr, file=sys.stderr, flush=True)
218	n/a
219	n/a	if result[0] == INTERRUPTED:
220	n/a	raise KeyboardInterrupt
221	n/a	test_index += 1
222	n/a	except KeyboardInterrupt:
223	n/a	regrtest.interrupted = True
224	n/a	pending.interrupted = True
225	n/a	print()
226	n/a	finally:
227	n/a	if use_timeout:
228	n/a	faulthandler.cancel_dump_traceback_later()
229	n/a
230	n/a	# If tests are interrupted, wait until tests complete
231	n/a	wait_start = time.monotonic()
232	n/a	while True:
233	n/a	running = [worker.current_test for worker in workers]
234	n/a	running = list(filter(bool, running))
235	n/a	if not running:
236	n/a	break
237	n/a
238	n/a	dt = time.monotonic() - wait_start
239	n/a	line = "Waiting for %s (%s tests)" % (', '.join(running), len(running))
240	n/a	if dt >= WAIT_PROGRESS:
241	n/a	line = "%s since %.0f sec" % (line, dt)
242	n/a	print(line)
243	n/a	for worker in workers:
244	n/a	worker.join(WAIT_PROGRESS)