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
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
|
#! /usr/bin/env python
# encoding: utf-8
# Thomas Nagy, 2015 (ita)
#
# prefer the waf 1.8 version
"""
The full samba build can be faster by ~10%, but there are a few limitations:
* only one build process should be run at a time as the servers would use the same ports
* only one build command is going to be called ("waf build configure build" would not work)
def build(bld):
mod = Utils.load_tool('prefork')
mod.build(bld)
...
(build declarations after)
"""
import os, re, socket, threading, sys, subprocess, time, atexit, traceback
try:
import SocketServer
except ImportError:
import socketserver as SocketServer
try:
from queue import Queue
except ImportError:
from Queue import Queue
try:
import cPickle
except ImportError:
import pickle as cPickle
DEFAULT_PORT = 51200
HEADER_SIZE = 128
REQ = 'REQ'
RES = 'RES'
BYE = 'BYE'
def make_header(params):
header = ','.join(params)
if sys.hexversion > 0x3000000:
header = header.encode('iso8859-1')
header = header.ljust(HEADER_SIZE)
assert(len(header) == HEADER_SIZE)
return header
re_valid_query = re.compile('^[a-zA-Z0-9_, ]+$')
class req(SocketServer.StreamRequestHandler):
def handle(self):
while 1:
try:
self.process_command()
except Exception as e:
print(e)
break
def process_command(self):
query = self.rfile.read(HEADER_SIZE)
if not query:
return
#print(len(query))
assert(len(query) == HEADER_SIZE)
if sys.hexversion > 0x3000000:
query = query.decode('iso8859-1')
#print "%r" % query
if not re_valid_query.match(query):
raise ValueError('Invalid query %r' % query)
query = query.strip().split(',')
if query[0] == REQ:
self.run_command(query[1:])
elif query[0] == BYE:
raise ValueError('Exit')
else:
raise ValueError('Invalid query %r' % query)
def run_command(self, query):
size = int(query[0])
data = self.rfile.read(size)
assert(len(data) == size)
kw = cPickle.loads(data)
# run command
ret = out = err = exc = None
cmd = kw['cmd']
del kw['cmd']
#print(cmd)
try:
if kw['stdout'] or kw['stderr']:
p = subprocess.Popen(cmd, **kw)
(out, err) = p.communicate()
ret = p.returncode
else:
ret = subprocess.Popen(cmd, **kw).wait()
except Exception as e:
ret = -1
exc = str(e) + traceback.format_exc()
# write the results
if out or err or exc:
data = (out, err, exc)
data = cPickle.dumps(data, -1)
else:
data = ''
params = [RES, str(ret), str(len(data))]
self.wfile.write(make_header(params))
if data:
self.wfile.write(data)
def create_server(conn, cls):
#SocketServer.ThreadingTCPServer.allow_reuse_address = True
#server = SocketServer.ThreadingTCPServer(conn, req)
SocketServer.TCPServer.allow_reuse_address = True
server = SocketServer.TCPServer(conn, req)
#server.timeout = 6000 # seconds
server.serve_forever(poll_interval=0.001)
if __name__ == '__main__':
if len(sys.argv) > 1:
port = int(sys.argv[1])
else:
port = DEFAULT_PORT
#conn = (socket.gethostname(), port)
conn = ("127.0.0.1", port)
#print("listening - %r %r\n" % conn)
create_server(conn, req)
else:
import Runner, Utils
def init_task_pool(self):
# lazy creation, and set a common pool for all task consumers
pool = self.pool = []
for i in range(self.numjobs):
consumer = Runner.get_pool()
pool.append(consumer)
consumer.idx = i
self.ready = Queue(0)
def setq(consumer):
consumer.ready = self.ready
try:
threading.current_thread().idx = consumer.idx
except Exception as e:
print(e)
for x in pool:
x.ready.put(setq)
return pool
Runner.Parallel.init_task_pool = init_task_pool
PORT = 51200
def make_server(idx):
port = PORT + idx
cmd = [sys.executable, os.path.abspath(__file__), str(port)]
proc = subprocess.Popen(cmd)
proc.port = port
return proc
def make_conn(srv):
#port = PORT + idx
port = srv.port
conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
conn.connect(('127.0.0.1', port))
return conn
SERVERS = []
CONNS = []
def close_all():
while CONNS:
conn = CONNS.pop()
try:
conn.close()
except:
pass
while SERVERS:
srv = SERVERS.pop()
try:
srv.kill()
except:
pass
atexit.register(close_all)
def put_data(conn, data):
conn.send(data)
def read_data(conn, siz):
ret = conn.recv(siz)
if not ret:
print("closed connection?")
assert(len(ret) == siz)
return ret
def exec_command(cmd, **kw):
if 'log' in kw:
log = kw['log']
kw['stdout'] = kw['stderr'] = subprocess.PIPE
del(kw['log'])
else:
kw['stdout'] = kw['stderr'] = None
kw['shell'] = isinstance(cmd, str)
idx = threading.current_thread().idx
kw['cmd'] = cmd
data = cPickle.dumps(kw, -1)
params = [REQ, str(len(data))]
header = make_header(params)
conn = CONNS[idx]
put_data(conn, header)
put_data(conn, data)
data = read_data(conn, HEADER_SIZE)
if sys.hexversion > 0x3000000:
data = data.decode('iso8859-1')
lst = data.split(',')
ret = int(lst[1])
dlen = int(lst[2])
out = err = None
if dlen:
data = read_data(conn, dlen)
(out, err, exc) = cPickle.loads(data)
if exc:
raise Utils.WafError('Execution failure: %s' % exc)
if out:
log.write(out)
if err:
log.write(err)
return ret
def __init__(self):
threading.Thread.__init__(self)
# identifier of the current thread
self.idx = len(SERVERS)
# create a server and wait for the connection
srv = make_server(self.idx)
SERVERS.append(srv)
conn = None
for x in range(30):
try:
conn = make_conn(srv)
break
except socket.error:
time.sleep(0.01)
if not conn:
raise ValueError('Could not start the server!')
CONNS.append(conn)
self.setDaemon(1)
self.start()
Runner.TaskConsumer.__init__ = __init__
def build(bld):
# dangerous, there is no other command hopefully
Utils.exec_command = exec_command
|