diff options
Diffstat (limited to 'lib/portage/dbapi/_MergeProcess.py')
| -rw-r--r-- | lib/portage/dbapi/_MergeProcess.py | 142 |
1 files changed, 101 insertions, 41 deletions
diff --git a/lib/portage/dbapi/_MergeProcess.py b/lib/portage/dbapi/_MergeProcess.py index db3f3b105..d9ab2b47a 100644 --- a/lib/portage/dbapi/_MergeProcess.py +++ b/lib/portage/dbapi/_MergeProcess.py @@ -1,15 +1,19 @@ -# Copyright 2010-2020 Gentoo Authors +# Copyright 2010-2023 Gentoo Authors # Distributed under the terms of the GNU General Public License v2 +import functools import io +import multiprocessing import platform import fcntl import portage from portage import os, _unicode_decode -from portage.util._ctypes import find_library +from portage.package.ebuild._ipc.QueryCommand import QueryCommand +from portage.util._ctypes import load_libc import portage.elog.messages from portage.util._async.ForkProcess import ForkProcess +from portage.util import no_color class MergeProcess(ForkProcess): @@ -38,6 +42,7 @@ class MergeProcess(ForkProcess): "_dblink", "_elog_keys", "_locked_vdb", + "_mtime_reader", ) def _start(self): @@ -48,7 +53,7 @@ class MergeProcess(ForkProcess): # since closing of file descriptors in the subprocess # can prevent access to open database connections such # as that used by the sqlite metadata cache module. - cpv = "%s/%s" % (self.mycat, self.mypkg) + cpv = f"{self.mycat}/{self.mypkg}" settings = self.settings if cpv != settings.mycpv or "EAPI" not in settings.configdict["pkg"]: settings.reload() @@ -59,7 +64,7 @@ class MergeProcess(ForkProcess): # process, so that it's only done once rather than # for each child process. if platform.system() == "Linux" and "merge-sync" in settings.features: - find_library("c") + load_libc() # Inherit stdin by default, so that the pdb SIGUSR1 # handler is usable for the subprocess. @@ -70,7 +75,7 @@ class MergeProcess(ForkProcess): self.fd_pipes.setdefault(0, portage._get_stdin().fileno()) self.log_filter_file = self.settings.get("PORTAGE_LOG_FILTER_FILE_CMD") - super(MergeProcess, self)._start() + super()._start() def _lock_vdb(self): """ @@ -92,7 +97,7 @@ class MergeProcess(ForkProcess): self._locked_vdb = False def _elog_output_handler(self): - output = self._read_buf(self._elog_reader_fd) + output = self._read_buf(self._elog_reader_fd.fileno()) if output: lines = _unicode_decode(output).split("\n") if len(lines) == 1: @@ -108,25 +113,42 @@ class MergeProcess(ForkProcess): reporter(msg, phase=phase, key=key, out=out) elif output is not None: # EIO/POLLHUP - self.scheduler.remove_reader(self._elog_reader_fd) - os.close(self._elog_reader_fd) + self.scheduler.remove_reader(self._elog_reader_fd.fileno()) + self._elog_reader_fd.close() self._elog_reader_fd = None return False + def _mtime_handler(self): + if self._mtime_reader is not None: + try: + mtimes = self._mtime_reader.recv() + except EOFError: + self.scheduler.remove_reader(self._mtime_reader.fileno()) + self._mtime_reader.close() + self._mtime_reader = None + else: + if self.prev_mtimes is not None: + self.prev_mtimes.clear() + self.prev_mtimes.update(mtimes) + def _spawn(self, args, fd_pipes, **kwargs): """ Extend the superclass _spawn method to perform some pre-fork and post-fork actions. """ - elog_reader_fd, elog_writer_fd = os.pipe() + elog_reader_fd, elog_writer_fd = multiprocessing.Pipe(duplex=False) fcntl.fcntl( - elog_reader_fd, + elog_reader_fd.fileno(), fcntl.F_SETFL, - fcntl.fcntl(elog_reader_fd, fcntl.F_GETFL) | os.O_NONBLOCK, + fcntl.fcntl(elog_reader_fd.fileno(), fcntl.F_GETFL) | os.O_NONBLOCK, ) + mtime_reader, mtime_writer = multiprocessing.Pipe(duplex=False) + self.scheduler.add_reader(mtime_reader.fileno(), self._mtime_handler) + self._mtime_reader = mtime_reader + blockers = None if self.blockers is not None: # Query blockers in the main process, since closing @@ -142,9 +164,9 @@ class MergeProcess(ForkProcess): vartree=self.vartree, blockers=blockers, pipe=elog_writer_fd, + mtime_pipe=mtime_writer, ) - fd_pipes[elog_writer_fd] = elog_writer_fd - self.scheduler.add_reader(elog_reader_fd, self._elog_output_handler) + self.scheduler.add_reader(elog_reader_fd.fileno(), self._elog_output_handler) # If a concurrent emerge process tries to install a package # in the same SLOT as this one at the same time, there is an @@ -158,8 +180,34 @@ class MergeProcess(ForkProcess): self._dblink = mylink self._elog_reader_fd = elog_reader_fd - pids = super(MergeProcess, self)._spawn(args, fd_pipes, **kwargs) - os.close(elog_writer_fd) + + # Since the entire QueryCommand._db is not required, only pass + # in tree types that QueryCommand specifically requires. + child_db = {} + parent_db = portage.db if QueryCommand._db is None else QueryCommand._db + for root in parent_db: + child_db[root] = {} + for tree_type in ("vartree", "porttree"): + child_db[root][tree_type] = parent_db[root][tree_type] + + self.target = functools.partial( + self._target, + self._counter, + self._dblink, + self.infloc, + self.mydbapi, + self.myebuild, + self.pkgloc, + self.prev_mtimes, + self.settings, + self.unmerge, + self.vartree.dbapi, + child_db, + ) + + pids = super()._spawn(args, fd_pipes, **kwargs) + elog_writer_fd.close() + mtime_writer.close() self._buf = "" self._elog_keys = set() # Discard messages which will be collected by the subprocess, @@ -174,15 +222,27 @@ class MergeProcess(ForkProcess): return pids - def _run(self): - os.close(self._elog_reader_fd) - counter = self._counter - mylink = self._dblink - - portage.output.havecolor = self.settings.get("NOCOLOR") not in ("yes", "true") - + @staticmethod + def _target( + counter, + mylink, + infloc, + mydbapi, + myebuild, + pkgloc, + prev_mtimes, + settings, + unmerge, + vardb, + db, + ): + if QueryCommand._db is None: + # Initialize QueryCommand._db for AbstractEbuildProcess/EbuildIpcDaemon + # when not using the multiprocessing fork start method. + QueryCommand._db = db + portage.output.havecolor = not no_color(settings) # Avoid wastful updates of the vdb cache. - self.vartree.dbapi._flush_cache_enabled = False + vardb._flush_cache_enabled = False # In this subprocess we don't want PORTAGE_BACKGROUND to # suppress stdout/stderr output since they are pipes. We @@ -190,21 +250,21 @@ class MergeProcess(ForkProcess): # already be opened by the parent process, so we set the # "subprocess" value for use in conditional logging code # involving PORTAGE_LOG_FILE. - if not self.unmerge: + if not unmerge: # unmerge phases have separate logs - if self.settings.get("PORTAGE_BACKGROUND") == "1": - self.settings["PORTAGE_BACKGROUND_UNMERGE"] = "1" + if settings.get("PORTAGE_BACKGROUND") == "1": + settings["PORTAGE_BACKGROUND_UNMERGE"] = "1" else: - self.settings["PORTAGE_BACKGROUND_UNMERGE"] = "0" - self.settings.backup_changes("PORTAGE_BACKGROUND_UNMERGE") - self.settings["PORTAGE_BACKGROUND"] = "subprocess" - self.settings.backup_changes("PORTAGE_BACKGROUND") + settings["PORTAGE_BACKGROUND_UNMERGE"] = "0" + settings.backup_changes("PORTAGE_BACKGROUND_UNMERGE") + settings["PORTAGE_BACKGROUND"] = "subprocess" + settings.backup_changes("PORTAGE_BACKGROUND") rval = 1 - if self.unmerge: + if unmerge: if not mylink.exists(): rval = os.EX_OK - elif mylink.unmerge(ldpath_mtimes=self.prev_mtimes) == os.EX_OK: + elif mylink.unmerge(ldpath_mtimes=prev_mtimes) == os.EX_OK: mylink.lockdb() try: mylink.delete() @@ -213,11 +273,11 @@ class MergeProcess(ForkProcess): rval = os.EX_OK else: rval = mylink.merge( - self.pkgloc, - self.infloc, - myebuild=self.myebuild, - mydbapi=self.mydbapi, - prev_mtimes=self.prev_mtimes, + pkgloc, + infloc, + myebuild=myebuild, + mydbapi=mydbapi, + prev_mtimes=prev_mtimes, counter=counter, ) return rval @@ -232,7 +292,7 @@ class MergeProcess(ForkProcess): ): self.postinst_failure = True self.returncode = os.EX_OK - super(MergeProcess, self)._proc_join_done(proc, future) + super()._proc_join_done(proc, future) def _unregister(self): """ @@ -249,8 +309,8 @@ class MergeProcess(ForkProcess): self._unlock_vdb() if self._elog_reader_fd is not None: - self.scheduler.remove_reader(self._elog_reader_fd) - os.close(self._elog_reader_fd) + self.scheduler.remove_reader(self._elog_reader_fd.fileno()) + self._elog_reader_fd.close() self._elog_reader_fd = None if self._elog_keys is not None: for key in self._elog_keys: @@ -259,4 +319,4 @@ class MergeProcess(ForkProcess): ) self._elog_keys = None - super(MergeProcess, self)._unregister() + super()._unregister() |