bitbake-worker 16 KB

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