bitbake-worker 19 KB

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