blob: 070da4fe7ab11d9c89510fa420283e9e7b517a36 [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#
6# This program is free software; you can redistribute it and/or modify
7# it under the terms of the GNU General Public License version 2 as
8# published by the Free Software Foundation.
9#
10# This program is distributed in the hope that it will be useful,
11# but WITHOUT ANY WARRANTY; without even the implied warranty of
12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13# GNU General Public License for more details.
14#
15# You should have received a copy of the GNU General Public License along
16# with this program; if not, write to the Free Software Foundation, Inc.,
17# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
18
19"""
20 This module implements a multiprocessing.Process based server for bitbake.
21"""
22
23import bb
24import bb.event
Patrick Williamsc124f4f2015-09-15 14:41:29 -050025import logging
26import multiprocessing
Brad Bishopd7bf8c12018-02-25 22:55:05 -050027import threading
28import array
Patrick Williamsc124f4f2015-09-15 14:41:29 -050029import os
Patrick Williamsc124f4f2015-09-15 14:41:29 -050030import sys
31import time
32import select
Brad Bishopd7bf8c12018-02-25 22:55:05 -050033import socket
34import subprocess
35import errno
36import re
37import datetime
38import bb.server.xmlrpcserver
39from bb import daemonize
40from multiprocessing import queues
Patrick Williamsc124f4f2015-09-15 14:41:29 -050041
42logger = logging.getLogger('BitBake')
43
Brad Bishopd7bf8c12018-02-25 22:55:05 -050044class ProcessTimeout(SystemExit):
45 pass
Patrick Williamsc124f4f2015-09-15 14:41:29 -050046
Brad Bishopd7bf8c12018-02-25 22:55:05 -050047class ProcessServer(multiprocessing.Process):
Patrick Williamsc124f4f2015-09-15 14:41:29 -050048 profile_filename = "profile.log"
49 profile_processed_filename = "profile.log.processed"
50
Brad Bishopd7bf8c12018-02-25 22:55:05 -050051 def __init__(self, lock, sock, sockname):
52 multiprocessing.Process.__init__(self)
53 self.command_channel = False
54 self.command_channel_reply = False
Patrick Williamsc124f4f2015-09-15 14:41:29 -050055 self.quit = False
Brad Bishop6e60e8b2018-02-01 10:27:11 -050056 self.heartbeat_seconds = 1 # default, BB_HEARTBEAT_EVENT will be checked once we have a datastore.
57 self.next_heartbeat = time.time()
Patrick Williamsc124f4f2015-09-15 14:41:29 -050058
Brad Bishopd7bf8c12018-02-25 22:55:05 -050059 self.event_handle = None
60 self.haveui = False
61 self.lastui = False
62 self.xmlrpc = False
63
64 self._idlefuns = {}
65
66 self.bitbake_lock = lock
67 self.sock = sock
68 self.sockname = sockname
69
70 def register_idle_function(self, function, data):
71 """Register a function to be called while the server is idle"""
72 assert hasattr(function, '__call__')
73 self._idlefuns[function] = data
Patrick Williamsc124f4f2015-09-15 14:41:29 -050074
75 def run(self):
Brad Bishopd7bf8c12018-02-25 22:55:05 -050076
77 if self.xmlrpcinterface[0]:
78 self.xmlrpc = bb.server.xmlrpcserver.BitBakeXMLRPCServer(self.xmlrpcinterface, self.cooker, self)
79
80 print("Bitbake XMLRPC server address: %s, server port: %s" % (self.xmlrpc.host, self.xmlrpc.port))
Patrick Williamsc124f4f2015-09-15 14:41:29 -050081
Brad Bishop6e60e8b2018-02-01 10:27:11 -050082 heartbeat_event = self.cooker.data.getVar('BB_HEARTBEAT_EVENT')
83 if heartbeat_event:
84 try:
85 self.heartbeat_seconds = float(heartbeat_event)
86 except:
Brad Bishop6e60e8b2018-02-01 10:27:11 -050087 bb.warn('Ignoring invalid BB_HEARTBEAT_EVENT=%s, must be a float specifying seconds.' % heartbeat_event)
Brad Bishopd7bf8c12018-02-25 22:55:05 -050088
89 self.timeout = self.server_timeout or self.cooker.data.getVar('BB_SERVER_TIMEOUT')
90 try:
91 if self.timeout:
92 self.timeout = float(self.timeout)
93 except:
94 bb.warn('Ignoring invalid BB_SERVER_TIMEOUT=%s, must be a float specifying seconds.' % self.timeout)
95
96
97 try:
98 self.bitbake_lock.seek(0)
99 self.bitbake_lock.truncate()
100 if self.xmlrpc:
101 self.bitbake_lock.write("%s %s:%s\n" % (os.getpid(), self.xmlrpc.host, self.xmlrpc.port))
102 else:
103 self.bitbake_lock.write("%s\n" % (os.getpid()))
104 self.bitbake_lock.flush()
105 except Exception as e:
106 print("Error writing to lock file: %s" % str(e))
107 pass
108
109 if self.cooker.configuration.profile:
110 try:
111 import cProfile as profile
112 except:
113 import profile
114 prof = profile.Profile()
115
116 ret = profile.Profile.runcall(prof, self.main)
117
118 prof.dump_stats("profile.log")
119 bb.utils.process_profilelog("profile.log")
120 print("Raw profiling information saved to profile.log and processed statistics to profile.log.processed")
121
122 else:
123 ret = self.main()
124
125 return ret
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500126
127 def main(self):
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500128 self.cooker.pre_serve()
129
Patrick Williamsd8c66bc2016-06-20 12:57:21 -0500130 bb.utils.set_process_name("Cooker")
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500131
132 ready = []
133
134 self.controllersock = False
135 fds = [self.sock]
136 if self.xmlrpc:
137 fds.append(self.xmlrpc)
138 print("Entering server connection loop")
139
140 def disconnect_client(self, fds):
141 if not self.haveui:
142 return
143 print("Disconnecting Client")
144 fds.remove(self.controllersock)
145 fds.remove(self.command_channel)
146 bb.event.unregister_UIHhandler(self.event_handle, True)
147 self.command_channel_reply.writer.close()
148 self.event_writer.writer.close()
149 del self.event_writer
150 self.controllersock.close()
151 self.controllersock = False
152 self.haveui = False
153 self.lastui = time.time()
154 self.cooker.clientComplete()
155 if self.timeout is None:
156 print("No timeout, exiting.")
157 self.quit = True
158
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500159 while not self.quit:
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500160 if self.sock in ready:
161 self.controllersock, address = self.sock.accept()
162 if self.haveui:
163 print("Dropping connection attempt as we have a UI %s" % (str(ready)))
164 self.controllersock.close()
165 else:
166 print("Accepting %s" % (str(ready)))
167 fds.append(self.controllersock)
168 if self.controllersock in ready:
169 try:
170 print("Connecting Client")
171 ui_fds = recvfds(self.controllersock, 3)
172
173 # Where to write events to
174 writer = ConnectionWriter(ui_fds[0])
175 self.event_handle = bb.event.register_UIHhandler(writer, True)
176 self.event_writer = writer
177
178 # Where to read commands from
179 reader = ConnectionReader(ui_fds[1])
180 fds.append(reader)
181 self.command_channel = reader
182
183 # Where to send command return values to
184 writer = ConnectionWriter(ui_fds[2])
185 self.command_channel_reply = writer
186
187 self.haveui = True
188
189 except (EOFError, OSError):
190 disconnect_client(self, fds)
191
192 if not self.timeout == -1.0 and not self.haveui and self.lastui and self.timeout and \
193 (self.lastui + self.timeout) < time.time():
194 print("Server timeout, exiting.")
195 self.quit = True
196
197 if self.command_channel in ready:
198 try:
199 command = self.command_channel.get()
200 except EOFError:
201 # Client connection shutting down
202 ready = []
203 disconnect_client(self, fds)
204 continue
205 if command[0] == "terminateServer":
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500206 self.quit = True
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500207 continue
208 try:
209 print("Running command %s" % command)
210 self.command_channel_reply.send(self.cooker.command.runCommand(command))
211 except Exception as e:
212 logger.exception('Exception in server main event loop running command %s (%s)' % (command, str(e)))
213
214 if self.xmlrpc in ready:
215 self.xmlrpc.handle_requests()
216
217 ready = self.idle_commands(.1, fds)
218
219 print("Exiting")
220 # Remove the socket file so we don't get any more connections to avoid races
221 os.unlink(self.sockname)
222 self.sock.close()
223
224 try:
225 self.cooker.shutdown(True)
Brad Bishop316dfdd2018-06-25 12:45:53 -0400226 self.cooker.notifier.stop()
227 self.cooker.confignotifier.stop()
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500228 except:
229 pass
230
231 self.cooker.post_serve()
232
233 # Finally release the lockfile but warn about other processes holding it open
234 lock = self.bitbake_lock
235 lockfile = lock.name
236 lock.close()
237 lock = None
238
239 while not lock:
240 with bb.utils.timeout(3):
241 lock = bb.utils.lockfile(lockfile, shared=False, retry=False, block=True)
Brad Bishopa5c52ff2018-11-23 10:55:50 +1300242 if lock:
243 # We hold the lock so we can remove the file (hide stale pid data)
244 bb.utils.remove(lockfile)
245 bb.utils.unlockfile(lock)
246 return
247
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500248 if not lock:
249 # Some systems may not have lsof available
250 procs = None
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500251 try:
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500252 procs = subprocess.check_output(["lsof", '-w', lockfile], stderr=subprocess.STDOUT)
253 except OSError as e:
254 if e.errno != errno.ENOENT:
255 raise
256 if procs is None:
257 # Fall back to fuser if lsof is unavailable
258 try:
259 procs = subprocess.check_output(["fuser", '-v', lockfile], stderr=subprocess.STDOUT)
260 except OSError as e:
261 if e.errno != errno.ENOENT:
262 raise
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500263
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500264 msg = "Delaying shutdown due to active processes which appear to be holding bitbake.lock"
265 if procs:
266 msg += ":\n%s" % str(procs)
267 print(msg)
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500268
269 def idle_commands(self, delay, fds=None):
270 nextsleep = delay
271 if not fds:
272 fds = []
273
Patrick Williamsc0f7c042017-02-23 20:41:17 -0600274 for function, data in list(self._idlefuns.items()):
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500275 try:
276 retval = function(self, data, False)
277 if retval is False:
278 del self._idlefuns[function]
279 nextsleep = None
280 elif retval is True:
281 nextsleep = None
Patrick Williamsc0f7c042017-02-23 20:41:17 -0600282 elif isinstance(retval, float) and nextsleep:
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500283 if (retval < nextsleep):
284 nextsleep = retval
285 elif nextsleep is None:
286 continue
287 else:
288 fds = fds + retval
289 except SystemExit:
290 raise
291 except Exception as exc:
292 if not isinstance(exc, bb.BBHandledException):
293 logger.exception('Running idle function')
294 del self._idlefuns[function]
295 self.quit = True
296
Brad Bishop6e60e8b2018-02-01 10:27:11 -0500297 # Create new heartbeat event?
298 now = time.time()
299 if now >= self.next_heartbeat:
300 # We might have missed heartbeats. Just trigger once in
301 # that case and continue after the usual delay.
302 self.next_heartbeat += self.heartbeat_seconds
303 if self.next_heartbeat <= now:
304 self.next_heartbeat = now + self.heartbeat_seconds
305 heartbeat = bb.event.HeartbeatEvent(now)
306 bb.event.fire(heartbeat, self.cooker.data)
307 if nextsleep and now + nextsleep > self.next_heartbeat:
308 # Shorten timeout so that we we wake up in time for
309 # the heartbeat.
310 nextsleep = self.next_heartbeat - now
311
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500312 if nextsleep is not None:
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500313 if self.xmlrpc:
314 nextsleep = self.xmlrpc.get_timeout(nextsleep)
315 try:
316 return select.select(fds,[],[],nextsleep)[0]
317 except InterruptedError:
318 # Ignore EINTR
319 return []
320 else:
321 return select.select(fds,[],[],0)[0]
322
323
324class ServerCommunicator():
325 def __init__(self, connection, recv):
326 self.connection = connection
327 self.recv = recv
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500328
329 def runCommand(self, command):
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500330 self.connection.send(command)
331 if not self.recv.poll(30):
332 raise ProcessTimeout("Timeout while waiting for a reply from the bitbake server")
333 return self.recv.get()
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500334
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500335 def updateFeatureSet(self, featureset):
336 _, error = self.runCommand(["setFeatures", featureset])
Patrick Williamsc124f4f2015-09-15 14:41:29 -0500337 if error:
338 logger.error("Unable to set the cooker to the correct featureset: %s" % error)
339 raise BaseException(error)
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500340
341 def getEventHandle(self):
342 handle, error = self.runCommand(["getUIHandlerNum"])
343 if error:
344 logger.error("Unable to get UI Handler Number: %s" % error)
345 raise BaseException(error)
346
347 return handle
348
349 def terminateServer(self):
350 self.connection.send(['terminateServer'])
351 return
352
353class BitBakeProcessServerConnection(object):
354 def __init__(self, ui_channel, recv, eq, sock):
355 self.connection = ServerCommunicator(ui_channel, recv)
356 self.events = eq
357 # Save sock so it doesn't get gc'd for the life of our connection
358 self.socket_connection = sock
359
360 def terminate(self):
361 self.socket_connection.close()
362 self.connection.connection.close()
363 self.connection.recv.close()
364 return
365
366class BitBakeServer(object):
367 start_log_format = '--- Starting bitbake server pid %s at %s ---'
368 start_log_datetime_format = '%Y-%m-%d %H:%M:%S.%f'
369
370 def __init__(self, lock, sockname, configuration, featureset):
371
372 self.configuration = configuration
373 self.featureset = featureset
374 self.sockname = sockname
375 self.bitbake_lock = lock
376 self.readypipe, self.readypipein = os.pipe()
377
378 # Create server control socket
379 if os.path.exists(sockname):
380 os.unlink(sockname)
381
Brad Bishop1a4b7ee2018-12-16 17:11:34 -0800382 # Place the log in the builddirectory alongside the lock file
383 logfile = os.path.join(os.path.dirname(self.bitbake_lock.name), "bitbake-cookerdaemon.log")
384
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500385 self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
386 # AF_UNIX has path length issues so chdir here to workaround
387 cwd = os.getcwd()
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500388 try:
389 os.chdir(os.path.dirname(sockname))
390 self.sock.bind(os.path.basename(sockname))
391 finally:
392 os.chdir(cwd)
393 self.sock.listen(1)
394
395 os.set_inheritable(self.sock.fileno(), True)
396 startdatetime = datetime.datetime.now()
397 bb.daemonize.createDaemon(self._startServer, logfile)
398 self.sock.close()
399 self.bitbake_lock.close()
Brad Bishop1a4b7ee2018-12-16 17:11:34 -0800400 os.close(self.readypipein)
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500401
402 ready = ConnectionReader(self.readypipe)
403 r = ready.poll(30)
404 if r:
Brad Bishop1a4b7ee2018-12-16 17:11:34 -0800405 try:
406 r = ready.get()
407 except EOFError:
408 # Trap the child exitting/closing the pipe and error out
409 r = None
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500410 if not r or r != "ready":
411 ready.close()
412 bb.error("Unable to start bitbake server")
413 if os.path.exists(logfile):
414 logstart_re = re.compile(self.start_log_format % ('([0-9]+)', '([0-9-]+ [0-9:.]+)'))
415 started = False
416 lines = []
Brad Bishopa5c52ff2018-11-23 10:55:50 +1300417 lastlines = []
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500418 with open(logfile, "r") as f:
419 for line in f:
420 if started:
421 lines.append(line)
422 else:
Brad Bishopa5c52ff2018-11-23 10:55:50 +1300423 lastlines.append(line)
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500424 res = logstart_re.match(line.rstrip())
425 if res:
426 ldatetime = datetime.datetime.strptime(res.group(2), self.start_log_datetime_format)
427 if ldatetime >= startdatetime:
428 started = True
429 lines.append(line)
Brad Bishopa5c52ff2018-11-23 10:55:50 +1300430 if len(lastlines) > 60:
431 lastlines = lastlines[-60:]
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500432 if lines:
Brad Bishopa5c52ff2018-11-23 10:55:50 +1300433 if len(lines) > 60:
434 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 -0500435 else:
436 bb.error("Server log for this session (%s):\n%s" % (logfile, "".join(lines)))
Brad Bishopa5c52ff2018-11-23 10:55:50 +1300437 elif lastlines:
438 bb.error("Server didn't start, last 60 loglines (%s):\n%s" % (logfile, "".join(lastlines)))
439 else:
440 bb.error("%s doesn't exist" % logfile)
441
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500442 raise SystemExit(1)
Brad Bishopa5c52ff2018-11-23 10:55:50 +1300443
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500444 ready.close()
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500445
446 def _startServer(self):
447 print(self.start_log_format % (os.getpid(), datetime.datetime.now().strftime(self.start_log_datetime_format)))
Brad Bishope2d5b612018-11-23 10:55:50 +1300448 sys.stdout.flush()
449
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500450 server = ProcessServer(self.bitbake_lock, self.sock, self.sockname)
451 self.configuration.setServerRegIdleCallback(server.register_idle_function)
Brad Bishop1a4b7ee2018-12-16 17:11:34 -0800452 os.close(self.readypipe)
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500453 writer = ConnectionWriter(self.readypipein)
Brad Bishop1a4b7ee2018-12-16 17:11:34 -0800454 self.cooker = bb.cooker.BBCooker(self.configuration, self.featureset)
455 writer.send("ready")
456 writer.close()
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500457 server.cooker = self.cooker
458 server.server_timeout = self.configuration.server_timeout
459 server.xmlrpcinterface = self.configuration.xmlrpcinterface
460 print("Started bitbake server pid %d" % os.getpid())
Brad Bishope2d5b612018-11-23 10:55:50 +1300461 sys.stdout.flush()
462
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500463 server.start()
464
465def connectProcessServer(sockname, featureset):
466 # Connect to socket
467 sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
468 # AF_UNIX has path length issues so chdir here to workaround
469 cwd = os.getcwd()
470
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500471 try:
Brad Bishope2d5b612018-11-23 10:55:50 +1300472 try:
473 os.chdir(os.path.dirname(sockname))
474 sock.connect(os.path.basename(sockname))
475 finally:
476 os.chdir(cwd)
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500477
Brad Bishop1a4b7ee2018-12-16 17:11:34 -0800478 readfd = writefd = readfd1 = writefd1 = readfd2 = writefd2 = None
479 eq = command_chan_recv = command_chan = None
480
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500481 # Send an fd for the remote to write events to
482 readfd, writefd = os.pipe()
483 eq = BBUIEventQueue(readfd)
484 # Send an fd for the remote to recieve commands from
485 readfd1, writefd1 = os.pipe()
486 command_chan = ConnectionWriter(writefd1)
487 # Send an fd for the remote to write commands results to
488 readfd2, writefd2 = os.pipe()
489 command_chan_recv = ConnectionReader(readfd2)
490
491 sendfds(sock, [writefd, readfd1, writefd2])
492
493 server_connection = BitBakeProcessServerConnection(command_chan, command_chan_recv, eq, sock)
494
495 # Close the ends of the pipes we won't use
496 for i in [writefd, readfd1, writefd2]:
497 os.close(i)
498
499 server_connection.connection.updateFeatureSet(featureset)
500
501 except (Exception, SystemExit) as e:
502 if command_chan_recv:
503 command_chan_recv.close()
504 if command_chan:
505 command_chan.close()
506 for i in [writefd, readfd1, writefd2]:
507 try:
Brad Bishope2d5b612018-11-23 10:55:50 +1300508 if i:
509 os.close(i)
Brad Bishopd7bf8c12018-02-25 22:55:05 -0500510 except OSError:
511 pass
512 sock.close()
513 raise
514
515 return server_connection
516
517def sendfds(sock, fds):
518 '''Send an array of fds over an AF_UNIX socket.'''
519 fds = array.array('i', fds)
520 msg = bytes([len(fds) % 256])
521 sock.sendmsg([msg], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, fds)])
522
523def recvfds(sock, size):
524 '''Receive an array of fds over an AF_UNIX socket.'''
525 a = array.array('i')
526 bytes_size = a.itemsize * size
527 msg, ancdata, flags, addr = sock.recvmsg(1, socket.CMSG_LEN(bytes_size))
528 if not msg and not ancdata:
529 raise EOFError
530 try:
531 if len(ancdata) != 1:
532 raise RuntimeError('received %d items of ancdata' %
533 len(ancdata))
534 cmsg_level, cmsg_type, cmsg_data = ancdata[0]
535 if (cmsg_level == socket.SOL_SOCKET and
536 cmsg_type == socket.SCM_RIGHTS):
537 if len(cmsg_data) % a.itemsize != 0:
538 raise ValueError
539 a.frombytes(cmsg_data)
540 assert len(a) % 256 == msg[0]
541 return list(a)
542 except (ValueError, IndexError):
543 pass
544 raise RuntimeError('Invalid data received')
545
546class BBUIEventQueue:
547 def __init__(self, readfd):
548
549 self.eventQueue = []
550 self.eventQueueLock = threading.Lock()
551 self.eventQueueNotify = threading.Event()
552
553 self.reader = ConnectionReader(readfd)
554
555 self.t = threading.Thread()
556 self.t.setDaemon(True)
557 self.t.run = self.startCallbackHandler
558 self.t.start()
559
560 def getEvent(self):
561 self.eventQueueLock.acquire()
562
563 if len(self.eventQueue) == 0:
564 self.eventQueueLock.release()
565 return None
566
567 item = self.eventQueue.pop(0)
568
569 if len(self.eventQueue) == 0:
570 self.eventQueueNotify.clear()
571
572 self.eventQueueLock.release()
573 return item
574
575 def waitEvent(self, delay):
576 self.eventQueueNotify.wait(delay)
577 return self.getEvent()
578
579 def queue_event(self, event):
580 self.eventQueueLock.acquire()
581 self.eventQueue.append(event)
582 self.eventQueueNotify.set()
583 self.eventQueueLock.release()
584
585 def send_event(self, event):
586 self.queue_event(pickle.loads(event))
587
588 def startCallbackHandler(self):
589 bb.utils.set_process_name("UIEventQueue")
590 while True:
591 try:
592 self.reader.wait()
593 event = self.reader.get()
594 self.queue_event(event)
595 except EOFError:
596 # Easiest way to exit is to close the file descriptor to cause an exit
597 break
598 self.reader.close()
599
600class ConnectionReader(object):
601
602 def __init__(self, fd):
603 self.reader = multiprocessing.connection.Connection(fd, writable=False)
604 self.rlock = multiprocessing.Lock()
605
606 def wait(self, timeout=None):
607 return multiprocessing.connection.wait([self.reader], timeout)
608
609 def poll(self, timeout=None):
610 return self.reader.poll(timeout)
611
612 def get(self):
613 with self.rlock:
614 res = self.reader.recv_bytes()
615 return multiprocessing.reduction.ForkingPickler.loads(res)
616
617 def fileno(self):
618 return self.reader.fileno()
619
620 def close(self):
621 return self.reader.close()
622
623
624class ConnectionWriter(object):
625
626 def __init__(self, fd):
627 self.writer = multiprocessing.connection.Connection(fd, readable=False)
628 self.wlock = multiprocessing.Lock()
629 # Why bb.event needs this I have no idea
630 self.event = self
631
632 def send(self, obj):
633 obj = multiprocessing.reduction.ForkingPickler.dumps(obj)
634 with self.wlock:
635 self.writer.send_bytes(obj)
636
637 def fileno(self):
638 return self.writer.fileno()
639
640 def close(self):
641 return self.writer.close()