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
222
223
|
# A unix-oriented process dispatcher. Uses a single thread with select and
# waitpid to dispatch tasks. This avoids several deadlocks that are possible
# with fork/exec + threads + Python.
import errno, os, select, sys
from datetime import datetime, timedelta
from progressbar import ProgressBar
from results import NullTestOutput, TestOutput, escape_cmdline
class Task(object):
def __init__(self, test, prefix, pid, stdout, stderr):
self.test = test
self.cmd = test.get_command(prefix)
self.pid = pid
self.stdout = stdout
self.stderr = stderr
self.start = datetime.utcnow()
self.out = []
self.err = []
def spawn_test(test, prefix, passthrough, run_skipped, show_cmd):
"""Spawn one child, return a task struct."""
if not test.enable and not run_skipped:
return None
cmd = test.get_command(prefix)
if show_cmd:
print(escape_cmdline(cmd))
if not passthrough:
(rout, wout) = os.pipe()
(rerr, werr) = os.pipe()
rv = os.fork()
# Parent.
if rv:
os.close(wout)
os.close(werr)
return Task(test, prefix, rv, rout, rerr)
# Child.
os.close(rout)
os.close(rerr)
os.dup2(wout, 1)
os.dup2(werr, 2)
os.execvp(cmd[0], cmd)
def get_max_wait(tasks, timeout):
"""
Return the maximum time we can wait before any task should time out.
"""
# If we have a progress-meter, we need to wake up to update it frequently.
wait = ProgressBar.update_granularity()
# If a timeout is supplied, we need to wake up for the first task to
# timeout if that is sooner.
if timeout:
now = datetime.utcnow()
timeout_delta = timedelta(seconds=timeout)
for task in tasks:
remaining = task.start + timeout_delta - now
if remaining < wait:
wait = remaining
# Return the wait time in seconds, clamped between zero and max_wait.
return max(wait.total_seconds(), 0)
def flush_input(fd, frags):
"""
Read any pages sitting in the file descriptor 'fd' into the list 'frags'.
"""
rv = os.read(fd, 4096)
frags.append(rv)
while len(rv) == 4096:
# If read() returns a full buffer, it may indicate there was 1 buffer
# worth of data, or that there is more data to read. Poll the socket
# before we read again to ensure that we will not block indefinitly.
readable, _, _ = select.select([fd], [], [], 0)
if not readable:
return
rv = os.read(fd, 4096)
frags.append(rv)
def read_input(tasks, timeout):
"""
Select on input or errors from the given task list for a max of timeout
seconds.
"""
rlist = []
exlist = []
outmap = {} # Fast access to fragment list given fd.
for t in tasks:
rlist.append(t.stdout)
rlist.append(t.stderr)
outmap[t.stdout] = t.out
outmap[t.stderr] = t.err
# This will trigger with a close event when the child dies, allowing
# us to respond immediately and not leave cores idle.
exlist.append(t.stdout)
readable = []
try:
readable, _, _ = select.select(rlist, [], exlist, timeout)
except OverflowError as e:
print >> sys.stderr, "timeout value", timeout
raise
for fd in readable:
flush_input(fd, outmap[fd])
def remove_task(tasks, pid):
"""
Return a pair with the removed task and the new, modified tasks list.
"""
index = None
for i, t in enumerate(tasks):
if t.pid == pid:
index = i
break
else:
raise KeyError("No such pid: {}".format(pid))
out = tasks[index]
tasks.pop(index)
return out
def timed_out(task, timeout):
"""
Return True if the given task has been running for longer than |timeout|.
|timeout| may be falsy, indicating an infinite timeout (in which case
timed_out always returns False).
"""
if timeout:
now = datetime.utcnow()
return (now - task.start) > timedelta(seconds=timeout)
return False
def reap_zombies(tasks, timeout):
"""
Search for children of this process that have finished. If they are tasks,
then this routine will clean up the child. This method returns a new task
list that has had the ended tasks removed, followed by the list of finished
tasks.
"""
finished = []
while True:
try:
pid, status = os.waitpid(0, os.WNOHANG)
if pid == 0:
break
except OSError as e:
if e.errno == errno.ECHILD:
break
raise e
ended = remove_task(tasks, pid)
flush_input(ended.stdout, ended.out)
flush_input(ended.stderr, ended.err)
os.close(ended.stdout)
os.close(ended.stderr)
returncode = os.WEXITSTATUS(status)
if os.WIFSIGNALED(status):
returncode = -os.WTERMSIG(status)
finished.append(
TestOutput(
ended.test,
ended.cmd,
''.join(ended.out),
''.join(ended.err),
returncode,
(datetime.utcnow() - ended.start).total_seconds(),
timed_out(ended, timeout)))
return tasks, finished
def kill_undead(tasks, timeout):
"""
Signal all children that are over the given timeout.
"""
for task in tasks:
if timed_out(task, timeout):
os.kill(task.pid, 9)
def run_all_tests(tests, prefix, pb, options):
# Copy and reverse for fast pop off end.
tests = list(tests)
tests = tests[:]
tests.reverse()
# The set of currently running tests.
tasks = []
while len(tests) or len(tasks):
while len(tests) and len(tasks) < options.worker_count:
test = tests.pop()
task = spawn_test(test, prefix,
options.passthrough, options.run_skipped, options.show_cmd)
if task:
tasks.append(task)
else:
yield NullTestOutput(test)
timeout = get_max_wait(tasks, options.timeout)
read_input(tasks, timeout)
kill_undead(tasks, options.timeout)
tasks, finished = reap_zombies(tasks, options.timeout)
# With Python3.4+ we could use yield from to remove this loop.
for out in finished:
yield out
# If we did not finish any tasks, poke the progress bar to show that
# the test harness is at least not frozen.
if len(finished) == 0:
pb.poke()
|
