[PATCH] emerge: enable parallel-fetch during pkg_pretend (bug 710432)

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 <[email protected]>
---
 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