bitbake-worker 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505
  1. #!/usr/bin/env python3
  2. #
  3. # SPDX-License-Identifier: GPL-2.0-only
  4. #
  5. import os
  6. import sys
  7. import warnings
  8. sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(sys.argv[0])), 'lib'))
  9. from bb import fetch2
  10. import logging
  11. import bb
  12. import select
  13. import errno
  14. import signal
  15. import pickle
  16. import traceback
  17. import queue
  18. from multiprocessing import Lock
  19. from threading import Thread
  20. if sys.getfilesystemencoding() != "utf-8":
  21. sys.exit("Please use a locale setting which supports UTF-8 (such as LANG=en_US.UTF-8).\nPython can't change the filesystem locale after loading so we need a UTF-8 when Python starts or things won't work.")
  22. # Users shouldn't be running this code directly
  23. if len(sys.argv) != 2 or not sys.argv[1].startswith("decafbad"):
  24. print("bitbake-worker is meant for internal execution by bitbake itself, please don't use it standalone.")
  25. sys.exit(1)
  26. profiling = False
  27. if sys.argv[1].startswith("decafbadbad"):
  28. profiling = True
  29. try:
  30. import cProfile as profile
  31. except:
  32. import profile
  33. # Unbuffer stdout to avoid log truncation in the event
  34. # of an unorderly exit as well as to provide timely
  35. # updates to log files for use with tail
  36. try:
  37. if sys.stdout.name == '<stdout>':
  38. import fcntl
  39. fl = fcntl.fcntl(sys.stdout.fileno(), fcntl.F_GETFL)
  40. fl |= os.O_SYNC
  41. fcntl.fcntl(sys.stdout.fileno(), fcntl.F_SETFL, fl)
  42. #sys.stdout = os.fdopen(sys.stdout.fileno(), 'w', 0)
  43. except:
  44. pass
  45. logger = logging.getLogger("BitBake")
  46. worker_pipe = sys.stdout.fileno()
  47. bb.utils.nonblockingfd(worker_pipe)
  48. # Need to guard against multiprocessing being used in child processes
  49. # and multiple processes trying to write to the parent at the same time
  50. worker_pipe_lock = None
  51. handler = bb.event.LogHandler()
  52. logger.addHandler(handler)
  53. if 0:
  54. # Code to write out a log file of all events passing through the worker
  55. logfilename = "/tmp/workerlogfile"
  56. format_str = "%(levelname)s: %(message)s"
  57. conlogformat = bb.msg.BBLogFormatter(format_str)
  58. consolelog = logging.FileHandler(logfilename)
  59. bb.msg.addDefaultlogFilter(consolelog)
  60. consolelog.setFormatter(conlogformat)
  61. logger.addHandler(consolelog)
  62. worker_queue = queue.Queue()
  63. def worker_fire(event, d):
  64. data = b"<event>" + pickle.dumps(event) + b"</event>"
  65. worker_fire_prepickled(data)
  66. def worker_fire_prepickled(event):
  67. global worker_queue
  68. worker_queue.put(event)
  69. #
  70. # We can end up with write contention with the cooker, it can be trying to send commands
  71. # and we can be trying to send event data back. Therefore use a separate thread for writing
  72. # back data to cooker.
  73. #
  74. worker_thread_exit = False
  75. def worker_flush(worker_queue):
  76. worker_queue_int = b""
  77. global worker_pipe, worker_thread_exit
  78. while True:
  79. try:
  80. worker_queue_int = worker_queue_int + worker_queue.get(True, 1)
  81. except queue.Empty:
  82. pass
  83. while (worker_queue_int or not worker_queue.empty()):
  84. try:
  85. (_, ready, _) = select.select([], [worker_pipe], [], 1)
  86. if not worker_queue.empty():
  87. worker_queue_int = worker_queue_int + worker_queue.get()
  88. written = os.write(worker_pipe, worker_queue_int)
  89. worker_queue_int = worker_queue_int[written:]
  90. except (IOError, OSError) as e:
  91. if e.errno != errno.EAGAIN and e.errno != errno.EPIPE:
  92. raise
  93. if worker_thread_exit and worker_queue.empty() and not worker_queue_int:
  94. return
  95. worker_thread = Thread(target=worker_flush, args=(worker_queue,))
  96. worker_thread.start()
  97. def worker_child_fire(event, d):
  98. global worker_pipe
  99. global worker_pipe_lock
  100. data = b"<event>" + pickle.dumps(event) + b"</event>"
  101. try:
  102. worker_pipe_lock.acquire()
  103. worker_pipe.write(data)
  104. worker_pipe_lock.release()
  105. except IOError:
  106. sigterm_handler(None, None)
  107. raise
  108. bb.event.worker_fire = worker_fire
  109. lf = None
  110. #lf = open("/tmp/workercommandlog", "w+")
  111. def workerlog_write(msg):
  112. if lf:
  113. lf.write(msg)
  114. lf.flush()
  115. def sigterm_handler(signum, frame):
  116. signal.signal(signal.SIGTERM, signal.SIG_DFL)
  117. os.killpg(0, signal.SIGTERM)
  118. sys.exit()
  119. def fork_off_task(cfg, data, databuilder, workerdata, fn, task, taskname, taskhash, unihash, appends, taskdepdata, extraconfigdata, quieterrors=False, dry_run_exec=False):
  120. # We need to setup the environment BEFORE the fork, since
  121. # a fork() or exec*() activates PSEUDO...
  122. envbackup = {}
  123. fakeenv = {}
  124. umask = None
  125. taskdep = workerdata["taskdeps"][fn]
  126. if 'umask' in taskdep and taskname in taskdep['umask']:
  127. # umask might come in as a number or text string..
  128. try:
  129. umask = int(taskdep['umask'][taskname],8)
  130. except TypeError:
  131. umask = taskdep['umask'][taskname]
  132. dry_run = cfg.dry_run or dry_run_exec
  133. # We can't use the fakeroot environment in a dry run as it possibly hasn't been built
  134. if 'fakeroot' in taskdep and taskname in taskdep['fakeroot'] and not dry_run:
  135. envvars = (workerdata["fakerootenv"][fn] or "").split()
  136. for key, value in (var.split('=') for var in envvars):
  137. envbackup[key] = os.environ.get(key)
  138. os.environ[key] = value
  139. fakeenv[key] = value
  140. fakedirs = (workerdata["fakerootdirs"][fn] or "").split()
  141. for p in fakedirs:
  142. bb.utils.mkdirhier(p)
  143. logger.debug(2, 'Running %s:%s under fakeroot, fakedirs: %s' %
  144. (fn, taskname, ', '.join(fakedirs)))
  145. else:
  146. envvars = (workerdata["fakerootnoenv"][fn] or "").split()
  147. for key, value in (var.split('=') for var in envvars):
  148. envbackup[key] = os.environ.get(key)
  149. os.environ[key] = value
  150. fakeenv[key] = value
  151. sys.stdout.flush()
  152. sys.stderr.flush()
  153. try:
  154. pipein, pipeout = os.pipe()
  155. pipein = os.fdopen(pipein, 'rb', 4096)
  156. pipeout = os.fdopen(pipeout, 'wb', 0)
  157. pid = os.fork()
  158. except OSError as e:
  159. logger.critical("fork failed: %d (%s)" % (e.errno, e.strerror))
  160. sys.exit(1)
  161. if pid == 0:
  162. def child():
  163. global worker_pipe
  164. global worker_pipe_lock
  165. pipein.close()
  166. signal.signal(signal.SIGTERM, sigterm_handler)
  167. # Let SIGHUP exit as SIGTERM
  168. signal.signal(signal.SIGHUP, sigterm_handler)
  169. bb.utils.signal_on_parent_exit("SIGTERM")
  170. # Save out the PID so that the event can include it the
  171. # events
  172. bb.event.worker_pid = os.getpid()
  173. bb.event.worker_fire = worker_child_fire
  174. worker_pipe = pipeout
  175. worker_pipe_lock = Lock()
  176. # Make the child the process group leader and ensure no
  177. # child process will be controlled by the current terminal
  178. # This ensures signals sent to the controlling terminal like Ctrl+C
  179. # don't stop the child processes.
  180. os.setsid()
  181. # No stdin
  182. newsi = os.open(os.devnull, os.O_RDWR)
  183. os.dup2(newsi, sys.stdin.fileno())
  184. if umask:
  185. os.umask(umask)
  186. try:
  187. bb_cache = bb.cache.NoCache(databuilder)
  188. (realfn, virtual, mc) = bb.cache.virtualfn2realfn(fn)
  189. the_data = databuilder.mcdata[mc]
  190. the_data.setVar("BB_WORKERCONTEXT", "1")
  191. the_data.setVar("BB_TASKDEPDATA", taskdepdata)
  192. if cfg.limited_deps:
  193. the_data.setVar("BB_LIMITEDDEPS", "1")
  194. the_data.setVar("BUILDNAME", workerdata["buildname"])
  195. the_data.setVar("DATE", workerdata["date"])
  196. the_data.setVar("TIME", workerdata["time"])
  197. for varname, value in extraconfigdata.items():
  198. the_data.setVar(varname, value)
  199. bb.parse.siggen.set_taskdata(workerdata["sigdata"])
  200. ret = 0
  201. the_data = bb_cache.loadDataFull(fn, appends)
  202. the_data.setVar('BB_TASKHASH', taskhash)
  203. the_data.setVar('BB_UNIHASH', unihash)
  204. bb.utils.set_process_name("%s:%s" % (the_data.getVar("PN"), taskname.replace("do_", "")))
  205. # exported_vars() returns a generator which *cannot* be passed to os.environ.update()
  206. # successfully. We also need to unset anything from the environment which shouldn't be there
  207. exports = bb.data.exported_vars(the_data)
  208. bb.utils.empty_environment()
  209. for e, v in exports:
  210. os.environ[e] = v
  211. for e in fakeenv:
  212. os.environ[e] = fakeenv[e]
  213. the_data.setVar(e, fakeenv[e])
  214. the_data.setVarFlag(e, 'export', "1")
  215. task_exports = the_data.getVarFlag(taskname, 'exports')
  216. if task_exports:
  217. for e in task_exports.split():
  218. the_data.setVarFlag(e, 'export', '1')
  219. v = the_data.getVar(e)
  220. if v is not None:
  221. os.environ[e] = v
  222. if quieterrors:
  223. the_data.setVarFlag(taskname, "quieterrors", "1")
  224. except Exception:
  225. if not quieterrors:
  226. logger.critical(traceback.format_exc())
  227. os._exit(1)
  228. try:
  229. if dry_run:
  230. return 0
  231. return bb.build.exec_task(fn, taskname, the_data, cfg.profile)
  232. except:
  233. os._exit(1)
  234. if not profiling:
  235. os._exit(child())
  236. else:
  237. profname = "profile-%s.log" % (fn.replace("/", "-") + "-" + taskname)
  238. prof = profile.Profile()
  239. try:
  240. ret = profile.Profile.runcall(prof, child)
  241. finally:
  242. prof.dump_stats(profname)
  243. bb.utils.process_profilelog(profname)
  244. os._exit(ret)
  245. else:
  246. for key, value in iter(envbackup.items()):
  247. if value is None:
  248. del os.environ[key]
  249. else:
  250. os.environ[key] = value
  251. return pid, pipein, pipeout
  252. class runQueueWorkerPipe():
  253. """
  254. Abstraction for a pipe between a worker thread and the worker server
  255. """
  256. def __init__(self, pipein, pipeout):
  257. self.input = pipein
  258. if pipeout:
  259. pipeout.close()
  260. bb.utils.nonblockingfd(self.input)
  261. self.queue = b""
  262. def read(self):
  263. start = len(self.queue)
  264. try:
  265. self.queue = self.queue + (self.input.read(102400) or b"")
  266. except (OSError, IOError) as e:
  267. if e.errno != errno.EAGAIN:
  268. raise
  269. end = len(self.queue)
  270. index = self.queue.find(b"</event>")
  271. while index != -1:
  272. worker_fire_prepickled(self.queue[:index+8])
  273. self.queue = self.queue[index+8:]
  274. index = self.queue.find(b"</event>")
  275. return (end > start)
  276. def close(self):
  277. while self.read():
  278. continue
  279. if len(self.queue) > 0:
  280. print("Warning, worker child left partial message: %s" % self.queue)
  281. self.input.close()
  282. normalexit = False
  283. class BitbakeWorker(object):
  284. def __init__(self, din):
  285. self.input = din
  286. bb.utils.nonblockingfd(self.input)
  287. self.queue = b""
  288. self.cookercfg = None
  289. self.databuilder = None
  290. self.data = None
  291. self.extraconfigdata = None
  292. self.build_pids = {}
  293. self.build_pipes = {}
  294. signal.signal(signal.SIGTERM, self.sigterm_exception)
  295. # Let SIGHUP exit as SIGTERM
  296. signal.signal(signal.SIGHUP, self.sigterm_exception)
  297. if "beef" in sys.argv[1]:
  298. bb.utils.set_process_name("Worker (Fakeroot)")
  299. else:
  300. bb.utils.set_process_name("Worker")
  301. def sigterm_exception(self, signum, stackframe):
  302. if signum == signal.SIGTERM:
  303. bb.warn("Worker received SIGTERM, shutting down...")
  304. elif signum == signal.SIGHUP:
  305. bb.warn("Worker received SIGHUP, shutting down...")
  306. self.handle_finishnow(None)
  307. signal.signal(signal.SIGTERM, signal.SIG_DFL)
  308. os.kill(os.getpid(), signal.SIGTERM)
  309. def serve(self):
  310. while True:
  311. (ready, _, _) = select.select([self.input] + [i.input for i in self.build_pipes.values()], [] , [], 1)
  312. if self.input in ready:
  313. try:
  314. r = self.input.read()
  315. if len(r) == 0:
  316. # EOF on pipe, server must have terminated
  317. self.sigterm_exception(signal.SIGTERM, None)
  318. self.queue = self.queue + r
  319. except (OSError, IOError):
  320. pass
  321. if len(self.queue):
  322. self.handle_item(b"cookerconfig", self.handle_cookercfg)
  323. self.handle_item(b"extraconfigdata", self.handle_extraconfigdata)
  324. self.handle_item(b"workerdata", self.handle_workerdata)
  325. self.handle_item(b"runtask", self.handle_runtask)
  326. self.handle_item(b"finishnow", self.handle_finishnow)
  327. self.handle_item(b"ping", self.handle_ping)
  328. self.handle_item(b"quit", self.handle_quit)
  329. for pipe in self.build_pipes:
  330. if self.build_pipes[pipe].input in ready:
  331. self.build_pipes[pipe].read()
  332. if len(self.build_pids):
  333. while self.process_waitpid():
  334. continue
  335. def handle_item(self, item, func):
  336. if self.queue.startswith(b"<" + item + b">"):
  337. index = self.queue.find(b"</" + item + b">")
  338. while index != -1:
  339. func(self.queue[(len(item) + 2):index])
  340. self.queue = self.queue[(index + len(item) + 3):]
  341. index = self.queue.find(b"</" + item + b">")
  342. def handle_cookercfg(self, data):
  343. self.cookercfg = pickle.loads(data)
  344. self.databuilder = bb.cookerdata.CookerDataBuilder(self.cookercfg, worker=True)
  345. self.databuilder.parseBaseConfiguration()
  346. self.data = self.databuilder.data
  347. def handle_extraconfigdata(self, data):
  348. self.extraconfigdata = pickle.loads(data)
  349. def handle_workerdata(self, data):
  350. self.workerdata = pickle.loads(data)
  351. bb.msg.loggerDefaultDebugLevel = self.workerdata["logdefaultdebug"]
  352. bb.msg.loggerDefaultVerbose = self.workerdata["logdefaultverbose"]
  353. bb.msg.loggerVerboseLogs = self.workerdata["logdefaultverboselogs"]
  354. bb.msg.loggerDefaultDomains = self.workerdata["logdefaultdomain"]
  355. for mc in self.databuilder.mcdata:
  356. self.databuilder.mcdata[mc].setVar("PRSERV_HOST", self.workerdata["prhost"])
  357. def handle_ping(self, _):
  358. workerlog_write("Handling ping\n")
  359. logger.warning("Pong from bitbake-worker!")
  360. def handle_quit(self, data):
  361. workerlog_write("Handling quit\n")
  362. global normalexit
  363. normalexit = True
  364. sys.exit(0)
  365. def handle_runtask(self, data):
  366. fn, task, taskname, taskhash, unihash, quieterrors, appends, taskdepdata, dry_run_exec = pickle.loads(data)
  367. workerlog_write("Handling runtask %s %s %s\n" % (task, fn, taskname))
  368. pid, pipein, pipeout = fork_off_task(self.cookercfg, self.data, self.databuilder, self.workerdata, fn, task, taskname, taskhash, unihash, appends, taskdepdata, self.extraconfigdata, quieterrors, dry_run_exec)
  369. self.build_pids[pid] = task
  370. self.build_pipes[pid] = runQueueWorkerPipe(pipein, pipeout)
  371. def process_waitpid(self):
  372. """
  373. Return none is there are no processes awaiting result collection, otherwise
  374. collect the process exit codes and close the information pipe.
  375. """
  376. try:
  377. pid, status = os.waitpid(-1, os.WNOHANG)
  378. if pid == 0 or os.WIFSTOPPED(status):
  379. return False
  380. except OSError:
  381. return False
  382. workerlog_write("Exit code of %s for pid %s\n" % (status, pid))
  383. if os.WIFEXITED(status):
  384. status = os.WEXITSTATUS(status)
  385. elif os.WIFSIGNALED(status):
  386. # Per shell conventions for $?, when a process exits due to
  387. # a signal, we return an exit code of 128 + SIGNUM
  388. status = 128 + os.WTERMSIG(status)
  389. task = self.build_pids[pid]
  390. del self.build_pids[pid]
  391. self.build_pipes[pid].close()
  392. del self.build_pipes[pid]
  393. worker_fire_prepickled(b"<exitcode>" + pickle.dumps((task, status)) + b"</exitcode>")
  394. return True
  395. def handle_finishnow(self, _):
  396. if self.build_pids:
  397. logger.info("Sending SIGTERM to remaining %s tasks", len(self.build_pids))
  398. for k, v in iter(self.build_pids.items()):
  399. try:
  400. os.kill(-k, signal.SIGTERM)
  401. os.waitpid(-1, 0)
  402. except:
  403. pass
  404. for pipe in self.build_pipes:
  405. self.build_pipes[pipe].read()
  406. try:
  407. worker = BitbakeWorker(os.fdopen(sys.stdin.fileno(), 'rb'))
  408. if not profiling:
  409. worker.serve()
  410. else:
  411. profname = "profile-worker.log"
  412. prof = profile.Profile()
  413. try:
  414. profile.Profile.runcall(prof, worker.serve)
  415. finally:
  416. prof.dump_stats(profname)
  417. bb.utils.process_profilelog(profname)
  418. except BaseException as e:
  419. if not normalexit:
  420. import traceback
  421. sys.stderr.write(traceback.format_exc())
  422. sys.stderr.write(str(e))
  423. worker_thread_exit = True
  424. worker_thread.join()
  425. workerlog_write("exitting")
  426. sys.exit(0)