Execute pkg_pretend phases in a coroutine while parallel-fetch is running concurrently. When it's time to execute the pkg_pretend phase for a remote binary package, use a Scheduler _get_prefetcher method to get a running prefetcher if available, and otherwise start a new fetcher.
Bug: https://bugs.gentoo.org/710432 Signed-off-by: Zac Medico <zmed...@gentoo.org> --- lib/_emerge/Scheduler.py | 94 +++++++++++++++++++++++++--------------- 1 file changed, 58 insertions(+), 36 deletions(-) diff --git a/lib/_emerge/Scheduler.py b/lib/_emerge/Scheduler.py index a69421288..20884986f 100644 --- a/lib/_emerge/Scheduler.py +++ b/lib/_emerge/Scheduler.py @@ -25,6 +25,7 @@ from portage._sets import SETPREFIX from portage._sets.base import InternalPackageSet from portage.util import ensure_dirs, writemsg, writemsg_level from portage.util.futures import asyncio +from portage.util.futures.compat_coroutine import coroutine, coroutine_return from portage.util.SlotObject import SlotObject from portage.util._async.SchedulerInterface import SchedulerInterface from portage.package.ebuild.digestcheck import digestcheck @@ -766,7 +767,8 @@ class Scheduler(PollScheduler): return prefetcher - def _run_pkg_pretend(self): + @coroutine + def _run_pkg_pretend(self, loop=None): """ Since pkg_pretend output may be important, this method sends all output directly to stdout (regardless of options like --quiet or @@ -774,7 +776,7 @@ class Scheduler(PollScheduler): """ failures = 0 - sched_iface = self._sched_iface + sched_iface = loop = asyncio._wrap_loop(loop or self._sched_iface) for x in self._mergelist: if not isinstance(x, Package): @@ -795,12 +797,18 @@ class Scheduler(PollScheduler): root_config = x.root_config settings = self.pkgsettings[root_config.root] settings.setcpv(x) + if not x.built: + # Get required SRC_URI metadata (it's not cached in x.metadata + # because some packages have an extremely large SRC_URI value). + portdb = root_config.trees["porttree"].dbapi + settings.configdict["pkg"]["SRC_URI"], = (yield portdb.async_aux_get( + x.cpv, ["SRC_URI"], myrepo=x.repo, loop=loop)) # setcpv/package.env allows for per-package PORTAGE_TMPDIR so we # have to validate it for each package rval = _check_temp_dir(settings) if rval != os.EX_OK: - return rval + coroutine_return(rval) build_dir_path = os.path.join( os.path.realpath(settings["PORTAGE_TMPDIR"]), @@ -809,7 +817,7 @@ class Scheduler(PollScheduler): settings["PORTAGE_BUILDDIR"] = build_dir_path build_dir = EbuildBuildDir(scheduler=sched_iface, settings=settings) - sched_iface.run_until_complete(build_dir.async_lock()) + yield build_dir.async_lock() current_task = None try: @@ -835,7 +843,7 @@ class Scheduler(PollScheduler): phase='clean', scheduler=sched_iface, settings=settings) current_task = clean_phase clean_phase.start() - clean_phase.wait() + yield clean_phase.async_wait() if x.built: tree = "bintree" @@ -845,10 +853,11 @@ class Scheduler(PollScheduler): # Display fetch on stdout, so that it's always clear what # is consuming time here. if bintree.isremote(x.cpv): - fetcher = BinpkgFetcher(pkg=x, - scheduler=sched_iface) - fetcher.start() - if fetcher.wait() != os.EX_OK: + fetcher = self._get_prefetcher(x) + if fetcher is None: + fetcher = BinpkgFetcher(pkg=x, scheduler=loop) + fetcher.start() + if (yield fetcher.async_wait()) != os.EX_OK: failures += 1 continue fetched = fetcher.pkg_path @@ -861,7 +870,7 @@ class Scheduler(PollScheduler): scheduler=sched_iface, _pkg_path=filename) current_task = verifier verifier.start() - if verifier.wait() != os.EX_OK: + if (yield verifier.async_wait()) != os.EX_OK: failures += 1 continue @@ -870,8 +879,7 @@ class Scheduler(PollScheduler): infloc = os.path.join(build_dir_path, "build-info") ensure_dirs(infloc) - self._sched_iface.run_until_complete( - bintree.dbapi.unpack_metadata(settings, infloc, loop=self._sched_iface)) + yield bintree.dbapi.unpack_metadata(settings, infloc, loop=loop) ebuild_path = os.path.join(infloc, x.pf + ".ebuild") settings.configdict["pkg"]["EMERGE_FROM"] = "binary" settings.configdict["pkg"]["MERGE_TYPE"] = "binary" @@ -905,7 +913,7 @@ class Scheduler(PollScheduler): current_task = pretend_phase pretend_phase.start() - ret = pretend_phase.wait() + ret = (yield pretend_phase.async_wait()) if ret != os.EX_OK: failures += 1 portage.elog.elog_process(x.cpv, settings) @@ -920,13 +928,13 @@ class Scheduler(PollScheduler): phase='clean', scheduler=sched_iface, settings=settings) clean_phase.start() - clean_phase.wait() + yield clean_phase.async_wait() - sched_iface.run_until_complete(build_dir.async_unlock()) + yield build_dir.async_unlock() if failures: - return FAILURE - return os.EX_OK + return coroutine_return(FAILURE) + coroutine_return(os.EX_OK) def merge(self): if "--resume" in self.myopts: @@ -988,11 +996,6 @@ class Scheduler(PollScheduler): if rval != os.EX_OK and not keep_going: return rval - if not fetchonly: - rval = self._run_pkg_pretend() - if rval != os.EX_OK: - return rval - while True: received_signal = [] @@ -1390,6 +1393,21 @@ class Scheduler(PollScheduler): self._set_max_jobs(1) self._add_prefetchers() + + if not self._build_opts.fetchonly: + # Run pkg_pretend concurrently with parallel-fetch, and be careful + # to respond appropriately to termination, so that we don't start + # any new tasks after we've been terminated. + try: + rval = self._sched_iface.run_until_complete(self._run_pkg_pretend(loop=self._sched_iface)) + except asyncio.CancelledError: + rval = None + self._termination_check() + if self._terminated_tasks or rval is None: + rval = 128 + signal.SIGINT + if rval != os.EX_OK: + return rval + self._add_packages() failed_pkgs = self._failed_pkgs portage.locks._quiet = self._background @@ -1742,6 +1760,23 @@ class Scheduler(PollScheduler): return bool(state_change) + def _get_prefetcher(self, pkg): + try: + prefetcher = self._prefetchers.pop(pkg, None) + except KeyError: + # KeyError observed with PyPy 1.8, despite None given as default. + # Note that PyPy 1.8 has the same WeakValueDictionary code as + # CPython 2.7, so it may be possible for CPython to raise KeyError + # here as well. + prefetcher = None + if prefetcher is not None and not prefetcher.isAlive(): + try: + self._task_queues.fetch._task_queue.remove(prefetcher) + except ValueError: + pass + prefetcher = None + return prefetcher + def _task(self, pkg): pkg_to_replace = None @@ -1758,20 +1793,7 @@ class Scheduler(PollScheduler): "installed", pkg.root_config, installed=True, operation="uninstall") - try: - prefetcher = self._prefetchers.pop(pkg, None) - except KeyError: - # KeyError observed with PyPy 1.8, despite None given as default. - # Note that PyPy 1.8 has the same WeakValueDictionary code as - # CPython 2.7, so it may be possible for CPython to raise KeyError - # here as well. - prefetcher = None - if prefetcher is not None and not prefetcher.isAlive(): - try: - self._task_queues.fetch._task_queue.remove(prefetcher) - except ValueError: - pass - prefetcher = None + prefetcher = self._get_prefetcher(pkg) task = MergeListItem(args_set=self._args_set, background=self._background, binpkg_opts=self._binpkg_opts, -- 2.25.3
