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
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
|
# Copyright 2020 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Module for job counters, limiting the amount of concurrent executions."""
import fcntl
import functools
import logging
import multiprocessing
import os
import re
import select
import selectors
import shlex
import subprocess
import sys
import zmake
class JobHandle:
"""Small object to handle claim of a job."""
def __init__(self, release_func, *args, **kwargs):
self.release_func = release_func
self.args = args
self.kwargs = kwargs
def __enter__(self):
pass
def __exit__(self, exc_type, exc_value, traceback):
if self.release_func:
self.release_func(*self.args, **self.kwargs)
class JobClient:
"""Abstract base class for all job clients."""
def get_job(self):
"""Claim a job."""
raise NotImplementedError("Abstract method not implemented")
@staticmethod
def env():
"""Get the environment variables necessary to share the job server."""
return {}
@staticmethod
def is_sequential():
"""Returns True if the jobserver is using -j1."""
return False
@staticmethod
def pass_fds():
"""Returns the file descriptors that should be passed to subprocesses."""
return []
def popen(self, argv, **kwargs):
"""Start a process using subprocess.Popen
All other arguments are passed to subprocess.Popen.
Returns:
A Popen object.
"""
# By default, we scrub the environment for all commands we run, setting
# the bare minimum (PATH only). This prevents us from building obscure
# dependencies on the environment variables.
kwargs.setdefault("env", {"PATH": "/bin:/usr/bin"})
kwargs.setdefault("pass_fds", [])
kwargs["env"].update(self.env())
kwargs["pass_fds"] += self.pass_fds()
logger = logging.getLogger(self.__class__.__name__)
logger.debug(
"Running `env -i %s%s%s`",
" ".join(f"{k}={shlex.quote(v)}" for k, v in kwargs["env"].items()),
" " if kwargs["env"] else "",
zmake.util.repr_command(argv),
)
return subprocess.Popen( # pylint:disable=consider-using-with
argv, **kwargs
)
class GNUMakeJobClient(JobClient):
"""A job client for GNU make.
A client of jobserver is allowed to run 1 job without contacting the
jobserver, so maintain an optional self._internal_pipe to hold that
job.
"""
def __init__(self, inheritable_pipe, jobs, internal_jobs=0, makeflags=None):
self._makeflags = makeflags
self._inheritable_pipe = inheritable_pipe
self.jobs = jobs
self._selector = selectors.DefaultSelector()
if internal_jobs:
self._internal_pipe = os.pipe()
os.write(self._internal_pipe[1], b"+" * internal_jobs)
os.set_blocking(self._internal_pipe[0], False)
self._selector.register(
self._internal_pipe[0],
selectors.EVENT_READ,
self._internal_pipe[1],
)
else:
self._internal_pipe = None
if self._inheritable_pipe is not None:
os.set_blocking(self._inheritable_pipe[0], False)
self._selector.register(
self._inheritable_pipe[0],
selectors.EVENT_READ,
self._inheritable_pipe[1],
)
def __del__(self):
if self._inheritable_pipe:
os.close(self._inheritable_pipe[0])
os.close(self._inheritable_pipe[1])
if self._internal_pipe:
os.close(self._internal_pipe[0])
os.close(self._internal_pipe[1])
@classmethod
def from_environ(cls, env=None, jobs=0):
"""Create a job client from an environment with the MAKEFLAGS variable.
If we are started under a GNU Make Job Server, we can search
the environment for a string "--jobserver-auth=R,W", where R
and W will be the read and write file descriptors to the pipe
respectively. If we don't find this environment variable (or
the string inside of it), this will raise an OSError.
The specification for MAKEFLAGS is:
* If the first char is "n", this is a dry run, just exit.
* If the flags contains -j1, go to sequential mode.
* If the flags contains --jobserver-auth=R,W AND those file
descriptors are valid, use the jobserver. Otherwise output a
warning.
Args:
env: Optionally, the environment to search.
jobs: The number of jobs set by the user on the command line.
Returns:
A GNUMakeJobClient configured appropriately or None if there is
no MAKEFLAGS environment variable.
"""
if env is None:
env = os.environ
makeflags = env.get("MAKEFLAGS")
if not makeflags:
return None
match = re.search(r"--jobserver-auth=(\d+),(\d+)", makeflags)
if match:
pipe = [int(x) for x in match.groups()]
if jobs:
pipe = None
logging.warning(
"-jN forced on command line; ignoring GNU make jobserver"
)
else:
try:
# Use F_GETFD to see if file descriptors are valid
fcntl.fcntl(pipe[0], fcntl.F_GETFD)
fcntl.fcntl(pipe[1], fcntl.F_GETFD)
logging.info("using GNU make jobserver")
except OSError:
pipe = None
logging.warning(
"No file descriptors; ignoring GNU make jobserver"
)
else:
pipe = None
if not jobs:
match = re.search(r"-j(\d+)", makeflags)
if match:
jobs = int(match.group(1))
if jobs == 1:
logging.info("Running in sequential mode (-j1)")
if makeflags[0] == "n":
logging.info("MAKEFLAGS contained dry-run flag")
sys.exit(0)
return cls(pipe, jobs, internal_jobs=1, makeflags=makeflags)
def get_job(self):
"""Claim a job.
Returns:
A JobHandle object.
"""
while True:
ready_items = self._selector.select()
if len(ready_items) > 0:
read_fd = ready_items[0][0].fd
write_fd = ready_items[0][0].data
try:
byte = os.read(read_fd, 1)
return JobHandle(
functools.partial(os.write, write_fd, byte)
)
except BlockingIOError:
pass
def env(self):
"""Get the environment variables necessary to share the job server."""
if self._makeflags:
return {"MAKEFLAGS": self._makeflags}
flag = ""
if self.jobs:
flag += f" -j{self.jobs}"
if self.jobs != 1 and self._inheritable_pipe is not None:
flag += " --jobserver-auth={},{}".format(*self._inheritable_pipe)
return {"MAKEFLAGS": flag}
def is_sequential(self):
return self.jobs == 1
def pass_fds(self):
"""Returns the file descriptors that should be passed to subprocesses."""
if self.jobs != 1 and self._inheritable_pipe is not None:
return self._inheritable_pipe
return []
class GNUMakeJobServer(GNUMakeJobClient):
"""Implements a GNU Make POSIX Job Server.
See https://www.gnu.org/software/make/manual/html_node/POSIX-Jobserver.html
for specification.
"""
def __init__(self, jobs=0):
if not jobs:
jobs = multiprocessing.cpu_count()
elif jobs > select.PIPE_BUF:
jobs = select.PIPE_BUF
super().__init__(os.pipe(), jobs)
os.write(self._inheritable_pipe[1], b"+" * jobs)
|
