bitbake-worker 18 KB

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