blob: 69aae626ebbfd534d2b16fc90f087a7727a29672 [file] [log] [blame]
Patrick Williamsc124f4f2015-09-15 14:41:29 -05001#
2# BitBake Process based server.
3#
4# Copyright (C) 2010 Bob Foerster <robert@erafx.com>
5#
Brad Bishopc342db32019-05-15 21:57:59 -04006# SPDX-License-Identifier: GPL-2.0-only
Patrick Williamsc124f4f2015-09-15 14:41:29 -05007#
Patrick Williamsc124f4f2015-09-15 14:41:29 -05008
9"""
10 This module implements a multiprocessing.Process based server for bitbake.
11"""
12
13import bb
14import bb.event
Patrick Williamsc124f4f2015-09-15 14:41:29 -050015import logging
16import multiprocessing
Brad Bishopd7bf8c12018-02-25 22:55:05 -050017import threading
18import array
Patrick Williamsc124f4f2015-09-15 14:41:29 -050019import os
Patrick Williamsc124f4f2015-09-15 14:41:29 -050020import sys
21import time
22import select
Brad Bishopd7bf8c12018-02-25 22:55:05 -050023import socket
24import subprocess
25import errno
26import re
27import datetime
28import bb.server.xmlrpcserver
29from bb import daemonize
30from multiprocessing import queues
Patrick Williamsc124f4f2015-09-15 14:41:29 -050031
32logger = logging.getLogger('BitBake')
33
Brad Bishopd7bf8c12018-02-25 22:55:05 -050034class ProcessTimeout(SystemExit):
35 pass
Patrick Williamsc124f4f2015-09-15 14:41:29 -050036
Brad Bishopd7bf8c12018-02-25 22:55:05 -050037class ProcessServer(multiprocessing.Process):
Patrick Williamsc124f4f2015-09-15 14:41:29 -050038 profile_filename = "profile.log"
39 profile_processed_filename = "profile.log.processed"
40
Brad Bishopd7bf8c12018-02-25 22:55:05 -050041 def __init__(self, lock, sock, sockname):
42 multiprocessing.Process.__init__(self)
43 self.command_channel = False
44 self.command_channel_reply = False
Patrick Williamsc124f4f2015-09-15 14:41:29 -050045 self.quit = False
Brad Bishop6e60e8b2018-02-01 10:27:11 -050046 self.heartbeat_seconds = 1 # default, BB_HEARTBEAT_EVENT will be checked once we have a datastore.
47 self.next_heartbeat = time.time()
Patrick Williamsc124f4f2015-09-15 14:41:29 -050048
Brad Bishopd7bf8c12018-02-25 22:55:05 -050049 self.event_handle = None
50 self.haveui = False
51 self.lastui = False
52 self.xmlrpc = False
53
54 self._idlefuns = {}
55
56 self.bitbake_lock = lock
57 self.sock = sock
58 self.sockname = sockname
59
60 def register_idle_function(self, function, data):
61 """Register a function to be called while the server is idle"""
62 assert hasattr(function, '__call__')
63 self._idlefuns[function] = data
Patrick Williamsc124f4f2015-09-15 14:41:29 -050064
65 def run(self):
Brad Bishopd7bf8c12018-02-25 22:55:05 -050066
67 if self.xmlrpcinterface[0]:
68 self.xmlrpc = bb.server.xmlrpcserver.BitBakeXMLRPCServer(self.xmlrpcinterface, self.cooker, self)
69
70 print("Bitbake XMLRPC server address: %s, server port: %s" % (self.xmlrpc.host, self.xmlrpc.port))
Patrick Williamsc124f4f2015-09-15 14:41:29 -050071
Brad Bishop6e60e8b2018-02-01 10:27:11 -050072 heartbeat_event = self.cooker.data.getVar('BB_HEARTBEAT_EVENT')
73 if heartbeat_event:
74 try:
75 self.heartbeat_seconds = float(heartbeat_event)
76 except:
Brad Bishop6e60e8b2018-02-01 10:27:11 -050077 bb.warn('Ignoring invalid BB_HEARTBEAT_EVENT=%s, must be a float specifying seconds.' % heartbeat_event)
Brad Bishopd7bf8c12018-02-25 22:55:05 -050078
79 self.timeout = self.server_timeout or self.cooker.data.getVar('BB_SERVER_TIMEOUT')
80 try:
81 if self.timeout:
82 self.timeout = float(self.timeout)
83 except:
84 bb.warn('Ignoring invalid BB_SERVER_TIMEOUT=%s, must be a float specifying seconds.' % self.timeout)
85
86
87 try:
88 self.bitbake_lock.seek(0)
89 self.bitbake_lock.truncate()
90 if self.xmlrpc:
91 self.bitbake_lock.write("%s %s:%s\n" % (os.getpid(), self.xmlrpc.host, self.xmlrpc.port))
92 else:
93 self.bitbake_lock.write("%s\n" % (os.getpid()))
94 self.bitbake_lock.flush()
95 except Exception as e:
96 print("Error writing to lock file: %s" % str(e))
97 pass
98
99 if self.cooker.configuration.profile:
100 try:
101 import cProfile as profile
102 except:
103 import profile
104 prof = profile.Profile()
105
106 ret = profile.Profile.runcall(prof, self.main)
107
108 prof.dump_stats("profile.log")
109 bb.utils.process_profilelog("profile.log")
110 print("Raw profiling information saved to profile.log and processed statistics to profile.log.processed")
111
112 else:
113 ret = self.main()
114
115 return ret
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500116
117 def main(self):
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500118 self.cooker.pre_serve()
119
Patrick Williamsd8c66bc2016-06-20 12:57:21 -0500120 bb.utils.set_process_name("Cooker")
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500121
122 ready = []
Brad Bishopf058f492019-01-28 23:50:33 -0500123 newconnections = []
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500124
125 self.controllersock = False
126 fds = [self.sock]
127 if self.xmlrpc:
128 fds.append(self.xmlrpc)
129 print("Entering server connection loop")
130
131 def disconnect_client(self, fds):
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500132 print("Disconnecting Client")
Brad Bishopf058f492019-01-28 23:50:33 -0500133 if self.controllersock:
134 fds.remove(self.controllersock)
135 self.controllersock.close()
136 self.controllersock = False
137 if self.haveui:
138 fds.remove(self.command_channel)
139 bb.event.unregister_UIHhandler(self.event_handle, True)
140 self.command_channel_reply.writer.close()
141 self.event_writer.writer.close()
142 self.command_channel.close()
143 self.command_channel = False
144 del self.event_writer
145 self.lastui = time.time()
146 self.cooker.clientComplete()
147 self.haveui = False
148 ready = select.select(fds,[],[],0)[0]
149 if newconnections:
150 print("Starting new client")
151 conn = newconnections.pop(-1)
152 fds.append(conn)
153 self.controllersock = conn
154 elif self.timeout is None and not ready:
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500155 print("No timeout, exiting.")
156 self.quit = True
157
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500158 while not self.quit:
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500159 if self.sock in ready:
Brad Bishopf058f492019-01-28 23:50:33 -0500160 while select.select([self.sock],[],[],0)[0]:
161 controllersock, address = self.sock.accept()
162 if self.controllersock:
163 print("Queuing %s (%s)" % (str(ready), str(newconnections)))
164 newconnections.append(controllersock)
165 else:
166 print("Accepting %s (%s)" % (str(ready), str(newconnections)))
167 self.controllersock = controllersock
168 fds.append(controllersock)
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500169 if self.controllersock in ready:
170 try:
Brad Bishopf058f492019-01-28 23:50:33 -0500171 print("Processing Client")
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500172 ui_fds = recvfds(self.controllersock, 3)
Brad Bishopf058f492019-01-28 23:50:33 -0500173 print("Connecting Client")
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500174
175 # Where to write events to
176 writer = ConnectionWriter(ui_fds[0])
177 self.event_handle = bb.event.register_UIHhandler(writer, True)
178 self.event_writer = writer
179
180 # Where to read commands from
181 reader = ConnectionReader(ui_fds[1])
182 fds.append(reader)
183 self.command_channel = reader
184
185 # Where to send command return values to
186 writer = ConnectionWriter(ui_fds[2])
187 self.command_channel_reply = writer
188
189 self.haveui = True
190
191 except (EOFError, OSError):
192 disconnect_client(self, fds)
193
194 if not self.timeout == -1.0 and not self.haveui and self.lastui and self.timeout and \
195 (self.lastui + self.timeout) < time.time():
196 print("Server timeout, exiting.")
197 self.quit = True
198
199 if self.command_channel in ready:
200 try:
201 command = self.command_channel.get()
202 except EOFError:
203 # Client connection shutting down
204 ready = []
205 disconnect_client(self, fds)
206 continue
207 if command[0] == "terminateServer":
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500208 self.quit = True
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500209 continue
210 try:
211 print("Running command %s" % command)
212 self.command_channel_reply.send(self.cooker.command.runCommand(command))
213 except Exception as e:
214 logger.exception('Exception in server main event loop running command %s (%s)' % (command, str(e)))
215
216 if self.xmlrpc in ready:
217 self.xmlrpc.handle_requests()
218
219 ready = self.idle_commands(.1, fds)
220
221 print("Exiting")
222 # Remove the socket file so we don't get any more connections to avoid races
223 os.unlink(self.sockname)
224 self.sock.close()
225
226 try:
227 self.cooker.shutdown(True)
Brad Bishop316dfdd2018-06-25 12:45:53 -0400228 self.cooker.notifier.stop()
229 self.cooker.confignotifier.stop()
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500230 except:
231 pass
232
233 self.cooker.post_serve()
234
235 # Finally release the lockfile but warn about other processes holding it open
236 lock = self.bitbake_lock
237 lockfile = lock.name
238 lock.close()
239 lock = None
240
241 while not lock:
242 with bb.utils.timeout(3):
243 lock = bb.utils.lockfile(lockfile, shared=False, retry=False, block=True)
Brad Bishopa5c52ff2018-11-23 10:55:50 +1300244 if lock:
245 # We hold the lock so we can remove the file (hide stale pid data)
246 bb.utils.remove(lockfile)
247 bb.utils.unlockfile(lock)
248 return
249
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500250 if not lock:
251 # Some systems may not have lsof available
252 procs = None
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500253 try:
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500254 procs = subprocess.check_output(["lsof", '-w', lockfile], stderr=subprocess.STDOUT)
255 except OSError as e:
256 if e.errno != errno.ENOENT:
257 raise
258 if procs is None:
259 # Fall back to fuser if lsof is unavailable
260 try:
261 procs = subprocess.check_output(["fuser", '-v', lockfile], stderr=subprocess.STDOUT)
262 except OSError as e:
263 if e.errno != errno.ENOENT:
264 raise
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500265
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500266 msg = "Delaying shutdown due to active processes which appear to be holding bitbake.lock"
267 if procs:
268 msg += ":\n%s" % str(procs)
269 print(msg)
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500270
271 def idle_commands(self, delay, fds=None):
272 nextsleep = delay
273 if not fds:
274 fds = []
275
Patrick Williamsc0f7c042017-02-23 20:41:17 -0600276 for function, data in list(self._idlefuns.items()):
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500277 try:
278 retval = function(self, data, False)
279 if retval is False:
280 del self._idlefuns[function]
281 nextsleep = None
282 elif retval is True:
283 nextsleep = None
Patrick Williamsc0f7c042017-02-23 20:41:17 -0600284 elif isinstance(retval, float) and nextsleep:
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500285 if (retval < nextsleep):
286 nextsleep = retval
287 elif nextsleep is None:
288 continue
289 else:
290 fds = fds + retval
291 except SystemExit:
292 raise
293 except Exception as exc:
294 if not isinstance(exc, bb.BBHandledException):
295 logger.exception('Running idle function')
296 del self._idlefuns[function]
297 self.quit = True
298
Brad Bishop6e60e8b2018-02-01 10:27:11 -0500299 # Create new heartbeat event?
300 now = time.time()
301 if now >= self.next_heartbeat:
302 # We might have missed heartbeats. Just trigger once in
303 # that case and continue after the usual delay.
304 self.next_heartbeat += self.heartbeat_seconds
305 if self.next_heartbeat <= now:
306 self.next_heartbeat = now + self.heartbeat_seconds
307 heartbeat = bb.event.HeartbeatEvent(now)
308 bb.event.fire(heartbeat, self.cooker.data)
309 if nextsleep and now + nextsleep > self.next_heartbeat:
310 # Shorten timeout so that we we wake up in time for
311 # the heartbeat.
312 nextsleep = self.next_heartbeat - now
313
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500314 if nextsleep is not None:
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500315 if self.xmlrpc:
316 nextsleep = self.xmlrpc.get_timeout(nextsleep)
317 try:
318 return select.select(fds,[],[],nextsleep)[0]
319 except InterruptedError:
320 # Ignore EINTR
321 return []
322 else:
323 return select.select(fds,[],[],0)[0]
324
325
326class ServerCommunicator():
327 def __init__(self, connection, recv):
328 self.connection = connection
329 self.recv = recv
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500330
331 def runCommand(self, command):
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500332 self.connection.send(command)
333 if not self.recv.poll(30):
334 raise ProcessTimeout("Timeout while waiting for a reply from the bitbake server")
335 return self.recv.get()
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500336
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500337 def updateFeatureSet(self, featureset):
338 _, error = self.runCommand(["setFeatures", featureset])
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500339 if error:
340 logger.error("Unable to set the cooker to the correct featureset: %s" % error)
341 raise BaseException(error)
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500342
343 def getEventHandle(self):
344 handle, error = self.runCommand(["getUIHandlerNum"])
345 if error:
346 logger.error("Unable to get UI Handler Number: %s" % error)
347 raise BaseException(error)
348
349 return handle
350
351 def terminateServer(self):
352 self.connection.send(['terminateServer'])
353 return
354
355class BitBakeProcessServerConnection(object):
356 def __init__(self, ui_channel, recv, eq, sock):
357 self.connection = ServerCommunicator(ui_channel, recv)
358 self.events = eq
359 # Save sock so it doesn't get gc'd for the life of our connection
360 self.socket_connection = sock
361
362 def terminate(self):
363 self.socket_connection.close()
364 self.connection.connection.close()
365 self.connection.recv.close()
366 return
367
368class BitBakeServer(object):
369 start_log_format = '--- Starting bitbake server pid %s at %s ---'
370 start_log_datetime_format = '%Y-%m-%d %H:%M:%S.%f'
371
372 def __init__(self, lock, sockname, configuration, featureset):
373
374 self.configuration = configuration
375 self.featureset = featureset
376 self.sockname = sockname
377 self.bitbake_lock = lock
378 self.readypipe, self.readypipein = os.pipe()
379
380 # Create server control socket
381 if os.path.exists(sockname):
382 os.unlink(sockname)
383
Brad Bishop1a4b7ee2018-12-16 17:11:34 -0800384 # Place the log in the builddirectory alongside the lock file
385 logfile = os.path.join(os.path.dirname(self.bitbake_lock.name), "bitbake-cookerdaemon.log")
386
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500387 self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
388 # AF_UNIX has path length issues so chdir here to workaround
389 cwd = os.getcwd()
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500390 try:
391 os.chdir(os.path.dirname(sockname))
392 self.sock.bind(os.path.basename(sockname))
393 finally:
394 os.chdir(cwd)
395 self.sock.listen(1)
396
397 os.set_inheritable(self.sock.fileno(), True)
398 startdatetime = datetime.datetime.now()
399 bb.daemonize.createDaemon(self._startServer, logfile)
400 self.sock.close()
401 self.bitbake_lock.close()
Brad Bishop1a4b7ee2018-12-16 17:11:34 -0800402 os.close(self.readypipein)
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500403
404 ready = ConnectionReader(self.readypipe)
Brad Bishopf058f492019-01-28 23:50:33 -0500405 r = ready.poll(5)
406 if not r:
407 bb.note("Bitbake server didn't start within 5 seconds, waiting for 90")
408 r = ready.poll(90)
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500409 if r:
Brad Bishop1a4b7ee2018-12-16 17:11:34 -0800410 try:
411 r = ready.get()
412 except EOFError:
413 # Trap the child exitting/closing the pipe and error out
414 r = None
Brad Bishopf058f492019-01-28 23:50:33 -0500415 if not r or r[0] != "r":
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500416 ready.close()
Brad Bishopf058f492019-01-28 23:50:33 -0500417 bb.error("Unable to start bitbake server (%s)" % str(r))
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500418 if os.path.exists(logfile):
419 logstart_re = re.compile(self.start_log_format % ('([0-9]+)', '([0-9-]+ [0-9:.]+)'))
420 started = False
421 lines = []
Brad Bishopa5c52ff2018-11-23 10:55:50 +1300422 lastlines = []
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500423 with open(logfile, "r") as f:
424 for line in f:
425 if started:
426 lines.append(line)
427 else:
Brad Bishopa5c52ff2018-11-23 10:55:50 +1300428 lastlines.append(line)
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500429 res = logstart_re.match(line.rstrip())
430 if res:
431 ldatetime = datetime.datetime.strptime(res.group(2), self.start_log_datetime_format)
432 if ldatetime >= startdatetime:
433 started = True
434 lines.append(line)
Brad Bishopa5c52ff2018-11-23 10:55:50 +1300435 if len(lastlines) > 60:
436 lastlines = lastlines[-60:]
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500437 if lines:
Brad Bishopa5c52ff2018-11-23 10:55:50 +1300438 if len(lines) > 60:
439 bb.error("Last 60 lines of server log for this session (%s):\n%s" % (logfile, "".join(lines[-60:])))
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500440 else:
441 bb.error("Server log for this session (%s):\n%s" % (logfile, "".join(lines)))
Brad Bishopa5c52ff2018-11-23 10:55:50 +1300442 elif lastlines:
443 bb.error("Server didn't start, last 60 loglines (%s):\n%s" % (logfile, "".join(lastlines)))
444 else:
445 bb.error("%s doesn't exist" % logfile)
446
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500447 raise SystemExit(1)
Brad Bishopa5c52ff2018-11-23 10:55:50 +1300448
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500449 ready.close()
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500450
451 def _startServer(self):
452 print(self.start_log_format % (os.getpid(), datetime.datetime.now().strftime(self.start_log_datetime_format)))
Brad Bishope2d5b612018-11-23 10:55:50 +1300453 sys.stdout.flush()
454
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500455 server = ProcessServer(self.bitbake_lock, self.sock, self.sockname)
456 self.configuration.setServerRegIdleCallback(server.register_idle_function)
Brad Bishop1a4b7ee2018-12-16 17:11:34 -0800457 os.close(self.readypipe)
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500458 writer = ConnectionWriter(self.readypipein)
Brad Bishop08902b02019-08-20 09:16:51 -0400459 try:
460 self.cooker = bb.cooker.BBCooker(self.configuration, self.featureset)
461 except bb.BBHandledException:
462 return None
Brad Bishopf058f492019-01-28 23:50:33 -0500463 writer.send("r")
Brad Bishop1a4b7ee2018-12-16 17:11:34 -0800464 writer.close()
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500465 server.cooker = self.cooker
466 server.server_timeout = self.configuration.server_timeout
467 server.xmlrpcinterface = self.configuration.xmlrpcinterface
468 print("Started bitbake server pid %d" % os.getpid())
Brad Bishope2d5b612018-11-23 10:55:50 +1300469 sys.stdout.flush()
470
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500471 server.start()
472
473def connectProcessServer(sockname, featureset):
474 # Connect to socket
475 sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
476 # AF_UNIX has path length issues so chdir here to workaround
477 cwd = os.getcwd()
478
Brad Bishopf058f492019-01-28 23:50:33 -0500479 readfd = writefd = readfd1 = writefd1 = readfd2 = writefd2 = None
480 eq = command_chan_recv = command_chan = None
481
482 sock.settimeout(10)
483
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500484 try:
Brad Bishope2d5b612018-11-23 10:55:50 +1300485 try:
486 os.chdir(os.path.dirname(sockname))
Brad Bishopf058f492019-01-28 23:50:33 -0500487 finished = False
488 while not finished:
489 try:
490 sock.connect(os.path.basename(sockname))
491 finished = True
492 except IOError as e:
493 if e.errno == errno.EWOULDBLOCK:
494 pass
Richard Purdie3da11142019-02-05 21:34:37 +0000495 raise
Brad Bishope2d5b612018-11-23 10:55:50 +1300496 finally:
497 os.chdir(cwd)
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500498
499 # Send an fd for the remote to write events to
500 readfd, writefd = os.pipe()
501 eq = BBUIEventQueue(readfd)
502 # Send an fd for the remote to recieve commands from
503 readfd1, writefd1 = os.pipe()
504 command_chan = ConnectionWriter(writefd1)
505 # Send an fd for the remote to write commands results to
506 readfd2, writefd2 = os.pipe()
507 command_chan_recv = ConnectionReader(readfd2)
508
509 sendfds(sock, [writefd, readfd1, writefd2])
510
511 server_connection = BitBakeProcessServerConnection(command_chan, command_chan_recv, eq, sock)
512
513 # Close the ends of the pipes we won't use
514 for i in [writefd, readfd1, writefd2]:
515 os.close(i)
516
517 server_connection.connection.updateFeatureSet(featureset)
518
519 except (Exception, SystemExit) as e:
520 if command_chan_recv:
521 command_chan_recv.close()
522 if command_chan:
523 command_chan.close()
524 for i in [writefd, readfd1, writefd2]:
525 try:
Brad Bishope2d5b612018-11-23 10:55:50 +1300526 if i:
527 os.close(i)
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500528 except OSError:
529 pass
530 sock.close()
531 raise
532
533 return server_connection
534
535def sendfds(sock, fds):
536 '''Send an array of fds over an AF_UNIX socket.'''
537 fds = array.array('i', fds)
538 msg = bytes([len(fds) % 256])
539 sock.sendmsg([msg], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, fds)])
540
541def recvfds(sock, size):
542 '''Receive an array of fds over an AF_UNIX socket.'''
543 a = array.array('i')
544 bytes_size = a.itemsize * size
545 msg, ancdata, flags, addr = sock.recvmsg(1, socket.CMSG_LEN(bytes_size))
546 if not msg and not ancdata:
547 raise EOFError
548 try:
549 if len(ancdata) != 1:
550 raise RuntimeError('received %d items of ancdata' %
551 len(ancdata))
552 cmsg_level, cmsg_type, cmsg_data = ancdata[0]
553 if (cmsg_level == socket.SOL_SOCKET and
554 cmsg_type == socket.SCM_RIGHTS):
555 if len(cmsg_data) % a.itemsize != 0:
556 raise ValueError
557 a.frombytes(cmsg_data)
558 assert len(a) % 256 == msg[0]
559 return list(a)
560 except (ValueError, IndexError):
561 pass
562 raise RuntimeError('Invalid data received')
563
564class BBUIEventQueue:
565 def __init__(self, readfd):
566
567 self.eventQueue = []
568 self.eventQueueLock = threading.Lock()
569 self.eventQueueNotify = threading.Event()
570
571 self.reader = ConnectionReader(readfd)
572
573 self.t = threading.Thread()
574 self.t.setDaemon(True)
575 self.t.run = self.startCallbackHandler
576 self.t.start()
577
578 def getEvent(self):
579 self.eventQueueLock.acquire()
580
581 if len(self.eventQueue) == 0:
582 self.eventQueueLock.release()
583 return None
584
585 item = self.eventQueue.pop(0)
586
587 if len(self.eventQueue) == 0:
588 self.eventQueueNotify.clear()
589
590 self.eventQueueLock.release()
591 return item
592
593 def waitEvent(self, delay):
594 self.eventQueueNotify.wait(delay)
595 return self.getEvent()
596
597 def queue_event(self, event):
598 self.eventQueueLock.acquire()
599 self.eventQueue.append(event)
600 self.eventQueueNotify.set()
601 self.eventQueueLock.release()
602
603 def send_event(self, event):
604 self.queue_event(pickle.loads(event))
605
606 def startCallbackHandler(self):
607 bb.utils.set_process_name("UIEventQueue")
608 while True:
609 try:
610 self.reader.wait()
611 event = self.reader.get()
612 self.queue_event(event)
613 except EOFError:
614 # Easiest way to exit is to close the file descriptor to cause an exit
615 break
616 self.reader.close()
617
618class ConnectionReader(object):
619
620 def __init__(self, fd):
621 self.reader = multiprocessing.connection.Connection(fd, writable=False)
622 self.rlock = multiprocessing.Lock()
623
624 def wait(self, timeout=None):
625 return multiprocessing.connection.wait([self.reader], timeout)
626
627 def poll(self, timeout=None):
628 return self.reader.poll(timeout)
629
630 def get(self):
631 with self.rlock:
632 res = self.reader.recv_bytes()
633 return multiprocessing.reduction.ForkingPickler.loads(res)
634
635 def fileno(self):
636 return self.reader.fileno()
637
638 def close(self):
639 return self.reader.close()
640
641
642class ConnectionWriter(object):
643
644 def __init__(self, fd):
645 self.writer = multiprocessing.connection.Connection(fd, readable=False)
646 self.wlock = multiprocessing.Lock()
647 # Why bb.event needs this I have no idea
648 self.event = self
649
650 def send(self, obj):
651 obj = multiprocessing.reduction.ForkingPickler.dumps(obj)
652 with self.wlock:
653 self.writer.send_bytes(obj)
654
655 def fileno(self):
656 return self.writer.fileno()
657
658 def close(self):
659 return self.writer.close()