process.py 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668
  1. #
  2. # BitBake Process based server.
  3. #
  4. # Copyright (C) 2010 Bob Foerster <robert@erafx.com>
  5. #
  6. # SPDX-License-Identifier: GPL-2.0-only
  7. #
  8. # This program is free software; you can redistribute it and/or modify
  9. # it under the terms of the GNU General Public License version 2 as
  10. # published by the Free Software Foundation.
  11. #
  12. # This program is distributed in the hope that it will be useful,
  13. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  15. # GNU General Public License for more details.
  16. #
  17. # You should have received a copy of the GNU General Public License along
  18. # with this program; if not, write to the Free Software Foundation, Inc.,
  19. # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  20. """
  21. This module implements a multiprocessing.Process based server for bitbake.
  22. """
  23. import bb
  24. import bb.event
  25. import logging
  26. import multiprocessing
  27. import threading
  28. import array
  29. import os
  30. import sys
  31. import time
  32. import select
  33. import socket
  34. import subprocess
  35. import errno
  36. import re
  37. import datetime
  38. import bb.server.xmlrpcserver
  39. from bb import daemonize
  40. from multiprocessing import queues
  41. logger = logging.getLogger('BitBake')
  42. class ProcessTimeout(SystemExit):
  43. pass
  44. class ProcessServer(multiprocessing.Process):
  45. profile_filename = "profile.log"
  46. profile_processed_filename = "profile.log.processed"
  47. def __init__(self, lock, sock, sockname):
  48. multiprocessing.Process.__init__(self)
  49. self.command_channel = False
  50. self.command_channel_reply = False
  51. self.quit = False
  52. self.heartbeat_seconds = 1 # default, BB_HEARTBEAT_EVENT will be checked once we have a datastore.
  53. self.next_heartbeat = time.time()
  54. self.event_handle = None
  55. self.haveui = False
  56. self.lastui = False
  57. self.xmlrpc = False
  58. self._idlefuns = {}
  59. self.bitbake_lock = lock
  60. self.sock = sock
  61. self.sockname = sockname
  62. def register_idle_function(self, function, data):
  63. """Register a function to be called while the server is idle"""
  64. assert hasattr(function, '__call__')
  65. self._idlefuns[function] = data
  66. def run(self):
  67. if self.xmlrpcinterface[0]:
  68. self.xmlrpc = bb.server.xmlrpcserver.BitBakeXMLRPCServer(self.xmlrpcinterface, self.cooker, self)
  69. print("Bitbake XMLRPC server address: %s, server port: %s" % (self.xmlrpc.host, self.xmlrpc.port))
  70. heartbeat_event = self.cooker.data.getVar('BB_HEARTBEAT_EVENT')
  71. if heartbeat_event:
  72. try:
  73. self.heartbeat_seconds = float(heartbeat_event)
  74. except:
  75. bb.warn('Ignoring invalid BB_HEARTBEAT_EVENT=%s, must be a float specifying seconds.' % heartbeat_event)
  76. self.timeout = self.server_timeout or self.cooker.data.getVar('BB_SERVER_TIMEOUT')
  77. try:
  78. if self.timeout:
  79. self.timeout = float(self.timeout)
  80. except:
  81. bb.warn('Ignoring invalid BB_SERVER_TIMEOUT=%s, must be a float specifying seconds.' % self.timeout)
  82. try:
  83. self.bitbake_lock.seek(0)
  84. self.bitbake_lock.truncate()
  85. if self.xmlrpc:
  86. self.bitbake_lock.write("%s %s:%s\n" % (os.getpid(), self.xmlrpc.host, self.xmlrpc.port))
  87. else:
  88. self.bitbake_lock.write("%s\n" % (os.getpid()))
  89. self.bitbake_lock.flush()
  90. except Exception as e:
  91. print("Error writing to lock file: %s" % str(e))
  92. pass
  93. if self.cooker.configuration.profile:
  94. try:
  95. import cProfile as profile
  96. except:
  97. import profile
  98. prof = profile.Profile()
  99. ret = profile.Profile.runcall(prof, self.main)
  100. prof.dump_stats("profile.log")
  101. bb.utils.process_profilelog("profile.log")
  102. print("Raw profiling information saved to profile.log and processed statistics to profile.log.processed")
  103. else:
  104. ret = self.main()
  105. return ret
  106. def main(self):
  107. self.cooker.pre_serve()
  108. bb.utils.set_process_name("Cooker")
  109. ready = []
  110. newconnections = []
  111. self.controllersock = False
  112. fds = [self.sock]
  113. if self.xmlrpc:
  114. fds.append(self.xmlrpc)
  115. print("Entering server connection loop")
  116. def disconnect_client(self, fds):
  117. print("Disconnecting Client")
  118. if self.controllersock:
  119. fds.remove(self.controllersock)
  120. self.controllersock.close()
  121. self.controllersock = False
  122. if self.haveui:
  123. fds.remove(self.command_channel)
  124. bb.event.unregister_UIHhandler(self.event_handle, True)
  125. self.command_channel_reply.writer.close()
  126. self.event_writer.writer.close()
  127. self.command_channel.close()
  128. self.command_channel = False
  129. del self.event_writer
  130. self.lastui = time.time()
  131. self.cooker.clientComplete()
  132. self.haveui = False
  133. ready = select.select(fds,[],[],0)[0]
  134. if newconnections:
  135. print("Starting new client")
  136. conn = newconnections.pop(-1)
  137. fds.append(conn)
  138. self.controllersock = conn
  139. elif self.timeout is None and not ready:
  140. print("No timeout, exiting.")
  141. self.quit = True
  142. while not self.quit:
  143. if self.sock in ready:
  144. while select.select([self.sock],[],[],0)[0]:
  145. controllersock, address = self.sock.accept()
  146. if self.controllersock:
  147. print("Queuing %s (%s)" % (str(ready), str(newconnections)))
  148. newconnections.append(controllersock)
  149. else:
  150. print("Accepting %s (%s)" % (str(ready), str(newconnections)))
  151. self.controllersock = controllersock
  152. fds.append(controllersock)
  153. if self.controllersock in ready:
  154. try:
  155. print("Processing Client")
  156. ui_fds = recvfds(self.controllersock, 3)
  157. print("Connecting Client")
  158. # Where to write events to
  159. writer = ConnectionWriter(ui_fds[0])
  160. self.event_handle = bb.event.register_UIHhandler(writer, True)
  161. self.event_writer = writer
  162. # Where to read commands from
  163. reader = ConnectionReader(ui_fds[1])
  164. fds.append(reader)
  165. self.command_channel = reader
  166. # Where to send command return values to
  167. writer = ConnectionWriter(ui_fds[2])
  168. self.command_channel_reply = writer
  169. self.haveui = True
  170. except (EOFError, OSError):
  171. disconnect_client(self, fds)
  172. if not self.timeout == -1.0 and not self.haveui and self.lastui and self.timeout and \
  173. (self.lastui + self.timeout) < time.time():
  174. print("Server timeout, exiting.")
  175. self.quit = True
  176. if self.command_channel in ready:
  177. try:
  178. command = self.command_channel.get()
  179. except EOFError:
  180. # Client connection shutting down
  181. ready = []
  182. disconnect_client(self, fds)
  183. continue
  184. if command[0] == "terminateServer":
  185. self.quit = True
  186. continue
  187. try:
  188. print("Running command %s" % command)
  189. self.command_channel_reply.send(self.cooker.command.runCommand(command))
  190. except Exception as e:
  191. logger.exception('Exception in server main event loop running command %s (%s)' % (command, str(e)))
  192. if self.xmlrpc in ready:
  193. self.xmlrpc.handle_requests()
  194. ready = self.idle_commands(.1, fds)
  195. print("Exiting")
  196. # Remove the socket file so we don't get any more connections to avoid races
  197. os.unlink(self.sockname)
  198. self.sock.close()
  199. try:
  200. self.cooker.shutdown(True)
  201. self.cooker.notifier.stop()
  202. self.cooker.confignotifier.stop()
  203. except:
  204. pass
  205. self.cooker.post_serve()
  206. # Finally release the lockfile but warn about other processes holding it open
  207. lock = self.bitbake_lock
  208. lockfile = lock.name
  209. lock.close()
  210. lock = None
  211. while not lock:
  212. with bb.utils.timeout(3):
  213. lock = bb.utils.lockfile(lockfile, shared=False, retry=False, block=True)
  214. if lock:
  215. # We hold the lock so we can remove the file (hide stale pid data)
  216. bb.utils.remove(lockfile)
  217. bb.utils.unlockfile(lock)
  218. return
  219. if not lock:
  220. # Some systems may not have lsof available
  221. procs = None
  222. try:
  223. procs = subprocess.check_output(["lsof", '-w', lockfile], stderr=subprocess.STDOUT)
  224. except OSError as e:
  225. if e.errno != errno.ENOENT:
  226. raise
  227. if procs is None:
  228. # Fall back to fuser if lsof is unavailable
  229. try:
  230. procs = subprocess.check_output(["fuser", '-v', lockfile], stderr=subprocess.STDOUT)
  231. except OSError as e:
  232. if e.errno != errno.ENOENT:
  233. raise
  234. msg = "Delaying shutdown due to active processes which appear to be holding bitbake.lock"
  235. if procs:
  236. msg += ":\n%s" % str(procs)
  237. print(msg)
  238. def idle_commands(self, delay, fds=None):
  239. nextsleep = delay
  240. if not fds:
  241. fds = []
  242. for function, data in list(self._idlefuns.items()):
  243. try:
  244. retval = function(self, data, False)
  245. if retval is False:
  246. del self._idlefuns[function]
  247. nextsleep = None
  248. elif retval is True:
  249. nextsleep = None
  250. elif isinstance(retval, float) and nextsleep:
  251. if (retval < nextsleep):
  252. nextsleep = retval
  253. elif nextsleep is None:
  254. continue
  255. else:
  256. fds = fds + retval
  257. except SystemExit:
  258. raise
  259. except Exception as exc:
  260. if not isinstance(exc, bb.BBHandledException):
  261. logger.exception('Running idle function')
  262. del self._idlefuns[function]
  263. self.quit = True
  264. # Create new heartbeat event?
  265. now = time.time()
  266. if now >= self.next_heartbeat:
  267. # We might have missed heartbeats. Just trigger once in
  268. # that case and continue after the usual delay.
  269. self.next_heartbeat += self.heartbeat_seconds
  270. if self.next_heartbeat <= now:
  271. self.next_heartbeat = now + self.heartbeat_seconds
  272. heartbeat = bb.event.HeartbeatEvent(now)
  273. bb.event.fire(heartbeat, self.cooker.data)
  274. if nextsleep and now + nextsleep > self.next_heartbeat:
  275. # Shorten timeout so that we we wake up in time for
  276. # the heartbeat.
  277. nextsleep = self.next_heartbeat - now
  278. if nextsleep is not None:
  279. if self.xmlrpc:
  280. nextsleep = self.xmlrpc.get_timeout(nextsleep)
  281. try:
  282. return select.select(fds,[],[],nextsleep)[0]
  283. except InterruptedError:
  284. # Ignore EINTR
  285. return []
  286. else:
  287. return select.select(fds,[],[],0)[0]
  288. class ServerCommunicator():
  289. def __init__(self, connection, recv):
  290. self.connection = connection
  291. self.recv = recv
  292. def runCommand(self, command):
  293. self.connection.send(command)
  294. if not self.recv.poll(30):
  295. raise ProcessTimeout("Timeout while waiting for a reply from the bitbake server")
  296. return self.recv.get()
  297. def updateFeatureSet(self, featureset):
  298. _, error = self.runCommand(["setFeatures", featureset])
  299. if error:
  300. logger.error("Unable to set the cooker to the correct featureset: %s" % error)
  301. raise BaseException(error)
  302. def getEventHandle(self):
  303. handle, error = self.runCommand(["getUIHandlerNum"])
  304. if error:
  305. logger.error("Unable to get UI Handler Number: %s" % error)
  306. raise BaseException(error)
  307. return handle
  308. def terminateServer(self):
  309. self.connection.send(['terminateServer'])
  310. return
  311. class BitBakeProcessServerConnection(object):
  312. def __init__(self, ui_channel, recv, eq, sock):
  313. self.connection = ServerCommunicator(ui_channel, recv)
  314. self.events = eq
  315. # Save sock so it doesn't get gc'd for the life of our connection
  316. self.socket_connection = sock
  317. def terminate(self):
  318. self.socket_connection.close()
  319. self.connection.connection.close()
  320. self.connection.recv.close()
  321. return
  322. class BitBakeServer(object):
  323. start_log_format = '--- Starting bitbake server pid %s at %s ---'
  324. start_log_datetime_format = '%Y-%m-%d %H:%M:%S.%f'
  325. def __init__(self, lock, sockname, configuration, featureset):
  326. self.configuration = configuration
  327. self.featureset = featureset
  328. self.sockname = sockname
  329. self.bitbake_lock = lock
  330. self.readypipe, self.readypipein = os.pipe()
  331. # Create server control socket
  332. if os.path.exists(sockname):
  333. os.unlink(sockname)
  334. # Place the log in the builddirectory alongside the lock file
  335. logfile = os.path.join(os.path.dirname(self.bitbake_lock.name), "bitbake-cookerdaemon.log")
  336. self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
  337. # AF_UNIX has path length issues so chdir here to workaround
  338. cwd = os.getcwd()
  339. try:
  340. os.chdir(os.path.dirname(sockname))
  341. self.sock.bind(os.path.basename(sockname))
  342. finally:
  343. os.chdir(cwd)
  344. self.sock.listen(1)
  345. os.set_inheritable(self.sock.fileno(), True)
  346. startdatetime = datetime.datetime.now()
  347. bb.daemonize.createDaemon(self._startServer, logfile)
  348. self.sock.close()
  349. self.bitbake_lock.close()
  350. os.close(self.readypipein)
  351. ready = ConnectionReader(self.readypipe)
  352. r = ready.poll(5)
  353. if not r:
  354. bb.note("Bitbake server didn't start within 5 seconds, waiting for 90")
  355. r = ready.poll(90)
  356. if r:
  357. try:
  358. r = ready.get()
  359. except EOFError:
  360. # Trap the child exitting/closing the pipe and error out
  361. r = None
  362. if not r or r[0] != "r":
  363. ready.close()
  364. bb.error("Unable to start bitbake server (%s)" % str(r))
  365. if os.path.exists(logfile):
  366. logstart_re = re.compile(self.start_log_format % ('([0-9]+)', '([0-9-]+ [0-9:.]+)'))
  367. started = False
  368. lines = []
  369. lastlines = []
  370. with open(logfile, "r") as f:
  371. for line in f:
  372. if started:
  373. lines.append(line)
  374. else:
  375. lastlines.append(line)
  376. res = logstart_re.match(line.rstrip())
  377. if res:
  378. ldatetime = datetime.datetime.strptime(res.group(2), self.start_log_datetime_format)
  379. if ldatetime >= startdatetime:
  380. started = True
  381. lines.append(line)
  382. if len(lastlines) > 60:
  383. lastlines = lastlines[-60:]
  384. if lines:
  385. if len(lines) > 60:
  386. bb.error("Last 60 lines of server log for this session (%s):\n%s" % (logfile, "".join(lines[-60:])))
  387. else:
  388. bb.error("Server log for this session (%s):\n%s" % (logfile, "".join(lines)))
  389. elif lastlines:
  390. bb.error("Server didn't start, last 60 loglines (%s):\n%s" % (logfile, "".join(lastlines)))
  391. else:
  392. bb.error("%s doesn't exist" % logfile)
  393. raise SystemExit(1)
  394. ready.close()
  395. def _startServer(self):
  396. print(self.start_log_format % (os.getpid(), datetime.datetime.now().strftime(self.start_log_datetime_format)))
  397. sys.stdout.flush()
  398. server = ProcessServer(self.bitbake_lock, self.sock, self.sockname)
  399. self.configuration.setServerRegIdleCallback(server.register_idle_function)
  400. os.close(self.readypipe)
  401. writer = ConnectionWriter(self.readypipein)
  402. self.cooker = bb.cooker.BBCooker(self.configuration, self.featureset)
  403. writer.send("r")
  404. writer.close()
  405. server.cooker = self.cooker
  406. server.server_timeout = self.configuration.server_timeout
  407. server.xmlrpcinterface = self.configuration.xmlrpcinterface
  408. print("Started bitbake server pid %d" % os.getpid())
  409. sys.stdout.flush()
  410. server.start()
  411. def connectProcessServer(sockname, featureset):
  412. # Connect to socket
  413. sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
  414. # AF_UNIX has path length issues so chdir here to workaround
  415. cwd = os.getcwd()
  416. readfd = writefd = readfd1 = writefd1 = readfd2 = writefd2 = None
  417. eq = command_chan_recv = command_chan = None
  418. sock.settimeout(10)
  419. try:
  420. try:
  421. os.chdir(os.path.dirname(sockname))
  422. finished = False
  423. while not finished:
  424. try:
  425. sock.connect(os.path.basename(sockname))
  426. finished = True
  427. except IOError as e:
  428. if e.errno == errno.EWOULDBLOCK:
  429. pass
  430. raise
  431. finally:
  432. os.chdir(cwd)
  433. # Send an fd for the remote to write events to
  434. readfd, writefd = os.pipe()
  435. eq = BBUIEventQueue(readfd)
  436. # Send an fd for the remote to recieve commands from
  437. readfd1, writefd1 = os.pipe()
  438. command_chan = ConnectionWriter(writefd1)
  439. # Send an fd for the remote to write commands results to
  440. readfd2, writefd2 = os.pipe()
  441. command_chan_recv = ConnectionReader(readfd2)
  442. sendfds(sock, [writefd, readfd1, writefd2])
  443. server_connection = BitBakeProcessServerConnection(command_chan, command_chan_recv, eq, sock)
  444. # Close the ends of the pipes we won't use
  445. for i in [writefd, readfd1, writefd2]:
  446. os.close(i)
  447. server_connection.connection.updateFeatureSet(featureset)
  448. except (Exception, SystemExit) as e:
  449. if command_chan_recv:
  450. command_chan_recv.close()
  451. if command_chan:
  452. command_chan.close()
  453. for i in [writefd, readfd1, writefd2]:
  454. try:
  455. if i:
  456. os.close(i)
  457. except OSError:
  458. pass
  459. sock.close()
  460. raise
  461. return server_connection
  462. def sendfds(sock, fds):
  463. '''Send an array of fds over an AF_UNIX socket.'''
  464. fds = array.array('i', fds)
  465. msg = bytes([len(fds) % 256])
  466. sock.sendmsg([msg], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, fds)])
  467. def recvfds(sock, size):
  468. '''Receive an array of fds over an AF_UNIX socket.'''
  469. a = array.array('i')
  470. bytes_size = a.itemsize * size
  471. msg, ancdata, flags, addr = sock.recvmsg(1, socket.CMSG_LEN(bytes_size))
  472. if not msg and not ancdata:
  473. raise EOFError
  474. try:
  475. if len(ancdata) != 1:
  476. raise RuntimeError('received %d items of ancdata' %
  477. len(ancdata))
  478. cmsg_level, cmsg_type, cmsg_data = ancdata[0]
  479. if (cmsg_level == socket.SOL_SOCKET and
  480. cmsg_type == socket.SCM_RIGHTS):
  481. if len(cmsg_data) % a.itemsize != 0:
  482. raise ValueError
  483. a.frombytes(cmsg_data)
  484. assert len(a) % 256 == msg[0]
  485. return list(a)
  486. except (ValueError, IndexError):
  487. pass
  488. raise RuntimeError('Invalid data received')
  489. class BBUIEventQueue:
  490. def __init__(self, readfd):
  491. self.eventQueue = []
  492. self.eventQueueLock = threading.Lock()
  493. self.eventQueueNotify = threading.Event()
  494. self.reader = ConnectionReader(readfd)
  495. self.t = threading.Thread()
  496. self.t.setDaemon(True)
  497. self.t.run = self.startCallbackHandler
  498. self.t.start()
  499. def getEvent(self):
  500. self.eventQueueLock.acquire()
  501. if len(self.eventQueue) == 0:
  502. self.eventQueueLock.release()
  503. return None
  504. item = self.eventQueue.pop(0)
  505. if len(self.eventQueue) == 0:
  506. self.eventQueueNotify.clear()
  507. self.eventQueueLock.release()
  508. return item
  509. def waitEvent(self, delay):
  510. self.eventQueueNotify.wait(delay)
  511. return self.getEvent()
  512. def queue_event(self, event):
  513. self.eventQueueLock.acquire()
  514. self.eventQueue.append(event)
  515. self.eventQueueNotify.set()
  516. self.eventQueueLock.release()
  517. def send_event(self, event):
  518. self.queue_event(pickle.loads(event))
  519. def startCallbackHandler(self):
  520. bb.utils.set_process_name("UIEventQueue")
  521. while True:
  522. try:
  523. self.reader.wait()
  524. event = self.reader.get()
  525. self.queue_event(event)
  526. except EOFError:
  527. # Easiest way to exit is to close the file descriptor to cause an exit
  528. break
  529. self.reader.close()
  530. class ConnectionReader(object):
  531. def __init__(self, fd):
  532. self.reader = multiprocessing.connection.Connection(fd, writable=False)
  533. self.rlock = multiprocessing.Lock()
  534. def wait(self, timeout=None):
  535. return multiprocessing.connection.wait([self.reader], timeout)
  536. def poll(self, timeout=None):
  537. return self.reader.poll(timeout)
  538. def get(self):
  539. with self.rlock:
  540. res = self.reader.recv_bytes()
  541. return multiprocessing.reduction.ForkingPickler.loads(res)
  542. def fileno(self):
  543. return self.reader.fileno()
  544. def close(self):
  545. return self.reader.close()
  546. class ConnectionWriter(object):
  547. def __init__(self, fd):
  548. self.writer = multiprocessing.connection.Connection(fd, readable=False)
  549. self.wlock = multiprocessing.Lock()
  550. # Why bb.event needs this I have no idea
  551. self.event = self
  552. def send(self, obj):
  553. obj = multiprocessing.reduction.ForkingPickler.dumps(obj)
  554. with self.wlock:
  555. self.writer.send_bytes(obj)
  556. def fileno(self):
  557. return self.writer.fileno()
  558. def close(self):
  559. return self.writer.close()