summaryrefslogtreecommitdiffstats
path: root/Lib/test/libregrtest/runtest_mp.py
blob: a732d705b1647cf4219d030745dd16133a0a0b82 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
import json
import os
import queue
import sys
import time
import traceback
import types
from test import support
try:
    import threading
except ImportError:
    print("Multiprocess option requires thread support")
    sys.exit(2)

from test.libregrtest.runtest import runtest, INTERRUPTED, CHILD_ERROR
from test.libregrtest.setup import setup_tests


# Minimum duration of a test to display its duration or to mention that
# the test is running in background
PROGRESS_MIN_TIME = 30.0   # seconds

# Display the running tests if nothing happened last N seconds
PROGRESS_UPDATE = 60.0   # seconds


def run_test_in_subprocess(testname, ns):
    """Run the given test in a subprocess with --slaveargs.

    ns is the option Namespace parsed from command-line arguments. regrtest
    is invoked in a subprocess with the --slaveargs argument; when the
    subprocess exits, its return code, stdout and stderr are returned as a
    3-tuple.
    """
    from subprocess import Popen, PIPE

    ns_dict = vars(ns)
    slaveargs = (ns_dict, testname)
    slaveargs = json.dumps(slaveargs)

    cmd = [sys.executable, *support.args_from_interpreter_flags(),
           '-X', 'faulthandler',
           '-m', 'test.regrtest',
           '--slaveargs', slaveargs]

    # 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.
    popen = Popen(cmd,
                  stdout=PIPE, stderr=PIPE,
                  universal_newlines=True,
                  close_fds=(os.name != 'nt'),
                  cwd=support.SAVEDCWD)
    with popen:
        stdout, stderr = popen.communicate()
        retcode = popen.wait()
    return retcode, stdout, stderr


def run_tests_slave(slaveargs):
    ns_dict, testname = json.loads(slaveargs)
    ns = types.SimpleNamespace(**ns_dict)

    setup_tests(ns)

    try:
        result = runtest(ns, testname)
    except KeyboardInterrupt:
        result = INTERRUPTED, ''
    except BaseException as e:
        traceback.print_exc()
        result = CHILD_ERROR, str(e)

    print()   # Force a newline (just in case)
    print(json.dumps(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):
        self.interrupted = False
        self.lock = threading.Lock()
        self.tests = tests

    def __iter__(self):
        return self

    def __next__(self):
        with self.lock:
            if self.interrupted:
                raise StopIteration('tests interrupted')
            return next(self.tests)


class MultiprocessThread(threading.Thread):
    def __init__(self, pending, output, ns):
        super().__init__()
        self.pending = pending
        self.output = output
        self.ns = ns
        self.current_test = None
        self.start_time = None

    def _runtest(self):
        try:
            test = next(self.pending)
        except StopIteration:
            self.output.put((None, None, None, None))
            return True

        try:
            self.start_time = time.monotonic()
            self.current_test = test

            retcode, stdout, stderr = run_test_in_subprocess(test, self.ns)
        finally:
            self.current_test = None

        stdout, _, result = stdout.strip().rpartition("\n")
        if retcode != 0:
            result = (CHILD_ERROR, "Exit code %s" % retcode)
            self.output.put((test, stdout.rstrip(), stderr.rstrip(),
                             result))
            return True

        if not result:
            self.output.put((None, None, None, None))
            return True

        result = json.loads(result)
        self.output.put((test, stdout.rstrip(), stderr.rstrip(),
                         result))
        return False

    def run(self):
        try:
            stop = False
            while not stop:
                stop = self._runtest()
        except BaseException:
            self.output.put((None, None, None, None))
            raise


def run_tests_multiprocess(regrtest):
    output = queue.Queue()
    pending = MultiprocessIterator(regrtest.tests)

    workers = [MultiprocessThread(pending, output, regrtest.ns)
               for i in range(regrtest.ns.use_mp)]
    for worker in workers:
        worker.start()

    def get_running(workers):
        running = []
        for worker in workers:
            current_test = worker.current_test
            if not current_test:
                continue
            dt = time.monotonic() - worker.start_time
            if dt >= PROGRESS_MIN_TIME:
                running.append('%s (%.0f sec)' % (current_test, dt))
        return running

    finished = 0
    test_index = 1
    timeout = max(PROGRESS_UPDATE, PROGRESS_MIN_TIME)
    try:
        while finished < regrtest.ns.use_mp:
            try:
                item = output.get(timeout=timeout)
            except queue.Empty:
                running = get_running(workers)
                if running:
                    print('running: %s' % ', '.join(running))
                continue

            test, stdout, stderr, result = item
            if test is None:
                finished += 1
                continue
            regrtest.accumulate_result(test, result)

            # Display progress
            text = test
            ok, test_time = result
            if (ok not in (CHILD_ERROR, INTERRUPTED)
                and test_time >= PROGRESS_MIN_TIME):
                text += ' (%.0f sec)' % test_time
            running = get_running(workers)
            if running:
                text += ' -- running: %s' % ', '.join(running)
            regrtest.display_progress(test_index, text)

            # Copy stdout and stderr from the child process
            if stdout:
                print(stdout, flush=True)
            if stderr:
                print(stderr, file=sys.stderr, flush=True)

            if result[0] == INTERRUPTED:
                raise KeyboardInterrupt
            if result[0] == CHILD_ERROR:
                msg = "Child error on {}: {}".format(test, result[1])
                raise Exception(msg)
            test_index += 1
    except KeyboardInterrupt:
        regrtest.interrupted = True
        pending.interrupted = True
        print()

    running = [worker.current_test for worker in workers]
    running = list(filter(bool, running))
    if running:
        print("Waiting for %s" % ', '.join(running))
    for worker in workers:
        worker.join()