bitbake-worker 14 KB

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