| #!/usr/bin/env python3 | |
| # | |
| # Copyright BitBake Contributors | |
| # | |
| # SPDX-License-Identifier: GPL-2.0-only | |
| # | |
| import os | |
| import sys | |
| import warnings | |
| warnings.simplefilter("default") | |
| warnings.filterwarnings("ignore", category=DeprecationWarning, message=".*use.of.fork.*may.lead.to.deadlocks.in.the.child.*") | |
| sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(sys.argv[0])), 'lib')) | |
| from bb import fetch | |
| import logging | |
| import bb | |
| import select | |
| import errno | |
| import signal | |
| import pickle | |
| import traceback | |
| import queue | |
| import shlex | |
| import subprocess | |
| import fcntl | |
| from threading import Thread | |
| __version__ = "2.19.1" | |
| if __name__ == "__main__": | |
| if __version__ != bb.__version__: | |
| sys.exit("Bitbake core library version and program version mismatch!") | |
| Lock = bb.multiprocessing.Lock | |
| # Remove when we have a minimum of python 3.10 | |
| if not hasattr(fcntl, 'F_SETPIPE_SZ'): | |
| fcntl.F_SETPIPE_SZ = 1031 | |
| bb.utils.check_system_locale() | |
| # Users shouldn't be running this code directly | |
| if len(sys.argv) != 2 or not sys.argv[1].startswith("decafbad"): | |
| print("bitbake-worker is meant for internal execution by bitbake itself, please don't use it standalone.") | |
| sys.exit(1) | |
| profiling = False | |
| if sys.argv[1].startswith("decafbadbad"): | |
| profiling = True | |
| try: | |
| import cProfile as profile | |
| except: | |
| import profile | |
| # Unbuffer stdout to avoid log truncation in the event | |
| # of an unorderly exit as well as to provide timely | |
| # updates to log files for use with tail | |
| try: | |
| if sys.stdout.name == '<stdout>': | |
| fl = fcntl.fcntl(sys.stdout.fileno(), fcntl.F_GETFL) | |
| fl |= os.O_SYNC | |
| fcntl.fcntl(sys.stdout.fileno(), fcntl.F_SETFL, fl) | |
| #sys.stdout = os.fdopen(sys.stdout.fileno(), 'w', 0) | |
| except: | |
| pass | |
| logger = logging.getLogger("BitBake") | |
| worker_pipe = sys.stdout.fileno() | |
| bb.utils.nonblockingfd(worker_pipe) | |
| # Try to make the pipe buffers larger as it is much more efficient. If we can't | |
| # e.g. out of buffer space (/proc/sys/fs/pipe-user-pages-soft) then just pass over. | |
| try: | |
| fcntl.fcntl(worker_pipe, fcntl.F_SETPIPE_SZ, 512 * 1024) | |
| except: | |
| pass | |
| # Need to guard against multiprocessing being used in child processes | |
| # and multiple processes trying to write to the parent at the same time | |
| worker_pipe_lock = None | |
| handler = bb.event.LogHandler() | |
| logger.addHandler(handler) | |
| if 0: | |
| # Code to write out a log file of all events passing through the worker | |
| logfilename = "/tmp/workerlogfile" | |
| format_str = "%(levelname)s: %(message)s" | |
| conlogformat = bb.msg.BBLogFormatter(format_str) | |
| consolelog = logging.FileHandler(logfilename) | |
| consolelog.setFormatter(conlogformat) | |
| logger.addHandler(consolelog) | |
| worker_queue = queue.Queue() | |
| def worker_fire(event, d): | |
| data = b"<event>" + pickle.dumps(event) + b"</event>" | |
| worker_fire_prepickled(data) | |
| def worker_fire_prepickled(event): | |
| global worker_queue | |
| worker_queue.put(event) | |
| # | |
| # We can end up with write contention with the cooker, it can be trying to send commands | |
| # and we can be trying to send event data back. Therefore use a separate thread for writing | |
| # back data to cooker. | |
| # | |
| worker_thread_exit = False | |
| def worker_flush(worker_queue): | |
| worker_queue_int = bytearray() | |
| global worker_pipe, worker_thread_exit | |
| while True: | |
| try: | |
| worker_queue_int.extend(worker_queue.get(True, 1)) | |
| except queue.Empty: | |
| pass | |
| while (worker_queue_int or not worker_queue.empty()): | |
| try: | |
| (_, ready, _) = select.select([], [worker_pipe], [], 1) | |
| if not worker_queue.empty(): | |
| worker_queue_int.extend(worker_queue.get()) | |
| written = os.write(worker_pipe, worker_queue_int) | |
| del worker_queue_int[0:written] | |
| except (IOError, OSError) as e: | |
| if e.errno != errno.EAGAIN and e.errno != errno.EPIPE: | |
| raise | |
| if worker_thread_exit and worker_queue.empty() and not worker_queue_int: | |
| return | |
| worker_thread = Thread(target=worker_flush, args=(worker_queue,)) | |
| worker_thread.start() | |
| def worker_child_fire(event, d): | |
| global worker_pipe | |
| global worker_pipe_lock | |
| data = b"<event>" + pickle.dumps(event) + b"</event>" | |
| try: | |
| with bb.utils.lock_timeout(worker_pipe_lock): | |
| while(len(data)): | |
| written = worker_pipe.write(data) | |
| data = data[written:] | |
| except IOError: | |
| sigterm_handler(None, None) | |
| raise | |
| bb.event.worker_fire = worker_fire | |
| lf = None | |
| #lf = open("/tmp/workercommandlog", "w+") | |
| def workerlog_write(msg): | |
| if lf: | |
| lf.write(msg) | |
| lf.flush() | |
| def sigterm_handler(signum, frame): | |
| signal.signal(signal.SIGTERM, signal.SIG_DFL) | |
| os.killpg(0, signal.SIGTERM) | |
| sys.exit() | |
| def fork_off_task(cfg, data, databuilder, workerdata, extraconfigdata, runtask): | |
| fn = runtask['fn'] | |
| task = runtask['task'] | |
| taskname = runtask['taskname'] | |
| taskhash = runtask['taskhash'] | |
| unihash = runtask['unihash'] | |
| appends = runtask['appends'] | |
| layername = runtask['layername'] | |
| taskdepdata = runtask['taskdepdata'] | |
| quieterrors = runtask['quieterrors'] | |
| # We need to setup the environment BEFORE the fork, since | |
| # a fork() or exec*() activates PSEUDO... | |
| envbackup = {} | |
| fakeroot = False | |
| fakeenv = {} | |
| umask = None | |
| uid = os.getuid() | |
| gid = os.getgid() | |
| taskdep = runtask['taskdep'] | |
| if 'umask' in taskdep and taskname in taskdep['umask']: | |
| umask = taskdep['umask'][taskname] | |
| elif workerdata["umask"]: | |
| umask = workerdata["umask"] | |
| if umask: | |
| # Convert to a python numeric value as it could be a string | |
| umask = bb.utils.to_filemode(umask) | |
| dry_run = cfg.dry_run or runtask['dry_run'] | |
| # We can't use the fakeroot environment in a dry run as it possibly hasn't been built | |
| if 'fakeroot' in taskdep and taskname in taskdep['fakeroot'] and not dry_run: | |
| fakeroot = True | |
| envvars = (runtask['fakerootenv'] or "").split() | |
| for key, value in (var.split('=',1) for var in envvars): | |
| envbackup[key] = os.environ.get(key) | |
| os.environ[key] = value | |
| fakeenv[key] = value | |
| fakedirs = (runtask['fakerootdirs'] or "").split() | |
| for p in fakedirs: | |
| bb.utils.mkdirhier(p) | |
| logger.debug2('Running %s:%s under fakeroot, fakedirs: %s' % | |
| (fn, taskname, ', '.join(fakedirs))) | |
| else: | |
| envvars = (runtask['fakerootnoenv'] or "").split() | |
| for key, value in (var.split('=',1) for var in envvars): | |
| envbackup[key] = os.environ.get(key) | |
| os.environ[key] = value | |
| fakeenv[key] = value | |
| sys.stdout.flush() | |
| sys.stderr.flush() | |
| try: | |
| pipein, pipeout = os.pipe() | |
| pipein = os.fdopen(pipein, 'rb', 4096) | |
| pipeout = os.fdopen(pipeout, 'wb', 0) | |
| pid = os.fork() | |
| except OSError as e: | |
| logger.critical("fork failed: %d (%s)" % (e.errno, e.strerror)) | |
| sys.exit(1) | |
| if pid == 0: | |
| def child(): | |
| global worker_pipe | |
| global worker_pipe_lock | |
| pipein.close() | |
| bb.utils.signal_on_parent_exit("SIGTERM") | |
| # Save out the PID so that the event can include it the | |
| # events | |
| bb.event.worker_pid = os.getpid() | |
| bb.event.worker_fire = worker_child_fire | |
| worker_pipe = pipeout | |
| worker_pipe_lock = Lock() | |
| # Make the child the process group leader and ensure no | |
| # child process will be controlled by the current terminal | |
| # This ensures signals sent to the controlling terminal like Ctrl+C | |
| # don't stop the child processes. | |
| os.setsid() | |
| signal.signal(signal.SIGTERM, sigterm_handler) | |
| # Let SIGHUP exit as SIGTERM | |
| signal.signal(signal.SIGHUP, sigterm_handler) | |
| # No stdin & stdout | |
| # stdout is used as a status report channel and must not be used by child processes. | |
| dumbio = os.open(os.devnull, os.O_RDWR) | |
| os.dup2(dumbio, sys.stdin.fileno()) | |
| os.dup2(dumbio, sys.stdout.fileno()) | |
| if umask is not None: | |
| os.umask(umask) | |
| try: | |
| (realfn, virtual, mc) = bb.cache.virtualfn2realfn(fn) | |
| the_data = databuilder.mcdata[mc] | |
| the_data.setVar("BB_WORKERCONTEXT", "1") | |
| the_data.setVar("BB_TASKDEPDATA", taskdepdata) | |
| the_data.setVar('BB_CURRENTTASK', taskname.replace("do_", "")) | |
| if cfg.limited_deps: | |
| the_data.setVar("BB_LIMITEDDEPS", "1") | |
| the_data.setVar("BUILDNAME", workerdata["buildname"]) | |
| the_data.setVar("DATE", workerdata["date"]) | |
| the_data.setVar("TIME", workerdata["time"]) | |
| for varname, value in extraconfigdata.items(): | |
| the_data.setVar(varname, value) | |
| bb.parse.siggen.set_taskdata(workerdata["sigdata"]) | |
| if "newhashes" in workerdata: | |
| bb.parse.siggen.set_taskhashes(workerdata["newhashes"]) | |
| ret = 0 | |
| the_data = databuilder.parseRecipe(fn, appends, layername) | |
| the_data.setVar('BB_TASKHASH', taskhash) | |
| the_data.setVar('BB_UNIHASH', unihash) | |
| bb.parse.siggen.setup_datacache_from_datastore(fn, the_data) | |
| bb.utils.set_process_name("%s:%s" % (the_data.getVar("PN"), taskname.replace("do_", ""))) | |
| if not bb.utils.to_boolean(the_data.getVarFlag(taskname, 'network')): | |
| if bb.utils.is_local_uid(uid): | |
| logger.debug("Attempting to disable network for %s" % taskname) | |
| bb.utils.disable_network(uid, gid) | |
| else: | |
| logger.debug("Skipping disable network for %s since %s is not a local uid." % (taskname, uid)) | |
| # exported_vars() expands each value, so it must be called before | |
| # empty_environment() below: an expansion that shells out needs the | |
| # environment bitbake was started with. We also need to unset | |
| # anything from the environment which shouldn't be there | |
| exports = bb.data.exported_vars(the_data) | |
| bb.utils.empty_environment() | |
| for e, v in exports: | |
| os.environ[e] = v | |
| for e in fakeenv: | |
| os.environ[e] = fakeenv[e] | |
| the_data.setVar(e, fakeenv[e]) | |
| the_data.setVarFlag(e, 'export', "1") | |
| task_exports = the_data.getVarFlag(taskname, 'exports') | |
| if task_exports: | |
| for e in task_exports.split(): | |
| the_data.setVarFlag(e, 'export', '1') | |
| v = the_data.getVar(e) | |
| if v is not None: | |
| os.environ[e] = v | |
| if quieterrors: | |
| the_data.setVarFlag(taskname, "quieterrors", "1") | |
| except Exception: | |
| if not quieterrors: | |
| logger.critical(traceback.format_exc()) | |
| os._exit(1) | |
| sys.stdout.flush() | |
| sys.stderr.flush() | |
| try: | |
| if dry_run: | |
| return 0 | |
| try: | |
| ret = bb.build.exec_task(fn, taskname, the_data, cfg.profile) | |
| finally: | |
| if fakeroot: | |
| fakerootcmd = shlex.split(the_data.getVar("FAKEROOTCMD")) | |
| subprocess.run(fakerootcmd + ['-S'], check=True, stdout=subprocess.PIPE) | |
| return ret | |
| except: | |
| os._exit(1) | |
| if not profiling: | |
| os._exit(child()) | |
| else: | |
| profname = "profile-%s.log" % (fn.replace("/", "-") + "-" + taskname) | |
| prof = profile.Profile() | |
| try: | |
| ret = profile.Profile.runcall(prof, child) | |
| finally: | |
| prof.dump_stats(profname) | |
| bb.utils.process_profilelog(profname) | |
| os._exit(ret) | |
| else: | |
| for key, value in iter(envbackup.items()): | |
| if value is None: | |
| del os.environ[key] | |
| else: | |
| os.environ[key] = value | |
| return pid, pipein, pipeout | |
| class runQueueWorkerPipe(): | |
| """ | |
| Abstraction for a pipe between a worker thread and the worker server | |
| """ | |
| def __init__(self, pipein, pipeout): | |
| self.input = pipein | |
| if pipeout: | |
| pipeout.close() | |
| bb.utils.nonblockingfd(self.input) | |
| self.queue = bytearray() | |
| def read(self): | |
| start = len(self.queue) | |
| try: | |
| self.queue.extend(self.input.read(512*1024) or b"") | |
| except (OSError, IOError) as e: | |
| if e.errno != errno.EAGAIN: | |
| raise | |
| end = len(self.queue) | |
| index = self.queue.find(b"</event>") | |
| while index != -1: | |
| msg = self.queue[:index+8] | |
| assert msg.startswith(b"<event>") and msg.count(b"<event>") == 1 | |
| worker_fire_prepickled(msg) | |
| self.queue = self.queue[index+8:] | |
| index = self.queue.find(b"</event>") | |
| return (end > start) | |
| def close(self): | |
| while self.read(): | |
| continue | |
| if len(self.queue) > 0: | |
| print("Warning, worker child left partial message: %s" % self.queue) | |
| self.input.close() | |
| normalexit = False | |
| class BitbakeWorker(object): | |
| def __init__(self, din): | |
| self.input = din | |
| bb.utils.nonblockingfd(self.input) | |
| self.queue = bytearray() | |
| self.cookercfg = None | |
| self.databuilder = None | |
| self.data = None | |
| self.extraconfigdata = None | |
| self.build_pids = {} | |
| self.build_pipes = {} | |
| signal.signal(signal.SIGTERM, self.sigterm_exception) | |
| # Let SIGHUP exit as SIGTERM | |
| signal.signal(signal.SIGHUP, self.sigterm_exception) | |
| if "beef" in sys.argv[1]: | |
| bb.utils.set_process_name("Worker (Fakeroot)") | |
| else: | |
| bb.utils.set_process_name("Worker") | |
| def sigterm_exception(self, signum, stackframe): | |
| if signum == signal.SIGTERM: | |
| bb.warn("Worker received SIGTERM, shutting down...") | |
| elif signum == signal.SIGHUP: | |
| bb.warn("Worker received SIGHUP, shutting down...") | |
| self.handle_finishnow(None) | |
| signal.signal(signal.SIGTERM, signal.SIG_DFL) | |
| os.kill(os.getpid(), signal.SIGTERM) | |
| def serve(self): | |
| while True: | |
| (ready, _, _) = select.select([self.input] + [i.input for i in self.build_pipes.values()], [] , [], 1) | |
| if self.input in ready: | |
| try: | |
| r = self.input.read() | |
| if len(r) == 0: | |
| # EOF on pipe, server must have terminated | |
| self.sigterm_exception(signal.SIGTERM, None) | |
| self.queue.extend(r) | |
| except (OSError, IOError): | |
| pass | |
| if len(self.queue): | |
| self.handle_item(b"cookerconfig", self.handle_cookercfg) | |
| self.handle_item(b"extraconfigdata", self.handle_extraconfigdata) | |
| self.handle_item(b"workerdata", self.handle_workerdata) | |
| self.handle_item(b"newtaskhashes", self.handle_newtaskhashes) | |
| self.handle_item(b"runtask", self.handle_runtask) | |
| self.handle_item(b"finishnow", self.handle_finishnow) | |
| self.handle_item(b"ping", self.handle_ping) | |
| self.handle_item(b"quit", self.handle_quit) | |
| for pipe in self.build_pipes: | |
| if self.build_pipes[pipe].input in ready: | |
| self.build_pipes[pipe].read() | |
| if len(self.build_pids): | |
| while self.process_waitpid(): | |
| continue | |
| def handle_item(self, item, func): | |
| opening_tag = b"<" + item + b">" | |
| if not self.queue.startswith(opening_tag): | |
| return | |
| tag_len = len(opening_tag) | |
| if len(self.queue) < tag_len + 4: | |
| # we need to receive more data | |
| return | |
| header = self.queue[tag_len:tag_len + 4] | |
| payload_len = int.from_bytes(header, 'big') | |
| # closing tag has length (tag_len + 1) | |
| if len(self.queue) < tag_len * 2 + 1 + payload_len: | |
| # we need to receive more data | |
| return | |
| index = self.queue.find(b"</" + item + b">") | |
| if index != -1: | |
| try: | |
| func(self.queue[(tag_len + 4):index]) | |
| except pickle.UnpicklingError: | |
| workerlog_write("Unable to unpickle data: %s\n" % ":".join("{:02x}".format(c) for c in self.queue)) | |
| raise | |
| self.queue = self.queue[(index + len(b"</") + len(item) + len(b">")):] | |
| def handle_cookercfg(self, data): | |
| self.cookercfg = pickle.loads(data) | |
| self.databuilder = bb.cookerdata.CookerDataBuilder(self.cookercfg, worker=True) | |
| self.databuilder.parseBaseConfiguration(worker=True) | |
| self.data = self.databuilder.data | |
| def handle_extraconfigdata(self, data): | |
| self.extraconfigdata = pickle.loads(data) | |
| def handle_workerdata(self, data): | |
| self.workerdata = pickle.loads(data) | |
| bb.build.verboseShellLogging = self.workerdata["build_verbose_shell"] | |
| bb.build.verboseStdoutLogging = self.workerdata["build_verbose_stdout"] | |
| bb.msg.loggerDefaultLogLevel = self.workerdata["logdefaultlevel"] | |
| bb.msg.loggerDefaultDomains = self.workerdata["logdefaultdomain"] | |
| for mc in self.databuilder.mcdata: | |
| self.databuilder.mcdata[mc].setVar("PRSERV_HOST", self.workerdata["prhost"]) | |
| self.databuilder.mcdata[mc].setVar("BB_HASHSERVE", self.workerdata["hashservaddr"]) | |
| self.databuilder.mcdata[mc].setVar("__bbclasstype", "recipe") | |
| def handle_newtaskhashes(self, data): | |
| self.workerdata["newhashes"] = pickle.loads(data) | |
| def handle_ping(self, _): | |
| workerlog_write("Handling ping\n") | |
| logger.warning("Pong from bitbake-worker!") | |
| def handle_quit(self, data): | |
| workerlog_write("Handling quit\n") | |
| global normalexit | |
| normalexit = True | |
| sys.exit(0) | |
| def handle_runtask(self, data): | |
| runtask = pickle.loads(data) | |
| fn = runtask['fn'] | |
| task = runtask['task'] | |
| taskname = runtask['taskname'] | |
| workerlog_write("Handling runtask %s %s %s\n" % (task, fn, taskname)) | |
| pid, pipein, pipeout = fork_off_task(self.cookercfg, self.data, self.databuilder, self.workerdata, self.extraconfigdata, runtask) | |
| self.build_pids[pid] = task | |
| self.build_pipes[pid] = runQueueWorkerPipe(pipein, pipeout) | |
| def process_waitpid(self): | |
| """ | |
| Return none is there are no processes awaiting result collection, otherwise | |
| collect the process exit codes and close the information pipe. | |
| """ | |
| try: | |
| pid, status = os.waitpid(-1, os.WNOHANG) | |
| if pid == 0 or os.WIFSTOPPED(status): | |
| return False | |
| except OSError: | |
| return False | |
| workerlog_write("Exit code of %s for pid %s\n" % (status, pid)) | |
| if os.WIFEXITED(status): | |
| status = os.WEXITSTATUS(status) | |
| elif os.WIFSIGNALED(status): | |
| # Per shell conventions for $?, when a process exits due to | |
| # a signal, we return an exit code of 128 + SIGNUM | |
| status = 128 + os.WTERMSIG(status) | |
| task = self.build_pids[pid] | |
| del self.build_pids[pid] | |
| self.build_pipes[pid].close() | |
| del self.build_pipes[pid] | |
| worker_fire_prepickled(b"<exitcode>" + pickle.dumps((task, status)) + b"</exitcode>") | |
| return True | |
| def handle_finishnow(self, _): | |
| if self.build_pids: | |
| logger.info("Sending SIGTERM to remaining %s tasks", len(self.build_pids)) | |
| for k, v in iter(self.build_pids.items()): | |
| try: | |
| os.kill(-k, signal.SIGTERM) | |
| os.waitpid(-1, 0) | |
| except: | |
| pass | |
| for pipe in self.build_pipes: | |
| self.build_pipes[pipe].read() | |
| try: | |
| worker = BitbakeWorker(os.fdopen(sys.stdin.fileno(), 'rb')) | |
| if not profiling: | |
| worker.serve() | |
| else: | |
| profname = "profile-worker.log" | |
| prof = profile.Profile() | |
| try: | |
| profile.Profile.runcall(prof, worker.serve) | |
| finally: | |
| prof.dump_stats(profname) | |
| bb.utils.process_profilelog(profname) | |
| except BaseException as e: | |
| if not normalexit: | |
| import traceback | |
| sys.stderr.write(traceback.format_exc()) | |
| sys.stderr.write(str(e)) | |
| finally: | |
| worker_thread_exit = True | |
| worker_thread.join() | |
| workerlog_write("exiting") | |
| if not normalexit: | |
| sys.exit(1) | |
| sys.exit(0) |