Merge packages asynchronously in Portage.

This allows for the scheduler to continue to run while packages are
being merged and installed, allowing for additional parallelism and
making better use of the CPUs.

Review URL: http://codereview.chromium.org/6713043

5 files changed, 61 insertions, 53 deletions

diff --git a/pym/_emerge/Binpkg.py b/pym/_emerge/Binpkg.py
index 00587451a..62d44c48f 100644
--- a/pym/_emerge/Binpkg.py
+++ b/pym/_emerge/Binpkg.py
@@ -307,7 +307,7 @@ class Binpkg(CompositeTask):
 		portage.elog.elog_process(self.pkg.cpv, self.settings)
 		self._build_dir.unlock()
 
-	def install(self):
+	def install(self, handler):
 
 		# This gives bashrc users an opportunity to do various things
 		# such as remove binary packages after they're installed.
@@ -320,19 +320,20 @@ class Binpkg(CompositeTask):
 			pkg=self.pkg, pkg_count=self.pkg_count,
 			pkg_path=self._pkg_path, scheduler=self.scheduler,
 			settings=settings, tree=self._tree, world_atom=self.world_atom)
+		task = merge.create_task()
+		task.addExitListener(self._install_exit)
+		self._start_task(task, handler)
 
-		try:
-			retval = merge.execute()
-		finally:
-			settings.pop("PORTAGE_BINPKG_FILE", None)
-			self._unlock_builddir()
+	def _install_exit(self, task):
+		self.settings.pop("PORTAGE_BINPKG_FILE", None)
+		self._unlock_builddir()
 
-		if retval == os.EX_OK and \
-			'binpkg-logs' not in self.settings.features and \
+		if self._default_final_exit(task) != os.EX_OK:
+			return
+
+		if 'binpkg-logs' not in self.settings.features and \
 			self.settings.get("PORTAGE_LOG_FILE"):
 			try:
 				os.unlink(self.settings["PORTAGE_LOG_FILE"])
 			except OSError:
 				pass
-		return retval
-
diff --git a/pym/_emerge/EbuildBuild.py b/pym/_emerge/EbuildBuild.py
index 98ab24522..c7a5f5cdb 100644
--- a/pym/_emerge/EbuildBuild.py
+++ b/pym/_emerge/EbuildBuild.py
@@ -314,7 +314,7 @@ class EbuildBuild(CompositeTask):
 			self._unlock_builddir()
 		self.wait()
 
-	def install(self):
+	def install(self, exit_handler):
 		"""
 		Install the package and then clean up and release locks.
 		Only call this after the build has completed successfully
@@ -343,10 +343,11 @@ class EbuildBuild(CompositeTask):
 			(pkg_count.curval, pkg_count.maxval, pkg.cpv)
 		logger.log(msg, short_msg=short_msg)
 
-		try:
-			rval = merge.execute()
-		finally:
-			self._unlock_builddir()
+		task = merge.create_task()
+		task.addExitListener(self._install_exit)
+		self._start_task(task, exit_handler)
 
-		return rval
+	def _install_exit(self, task):
+		self._unlock_builddir()
+		self._default_final_exit(task)
 
diff --git a/pym/_emerge/EbuildMerge.py b/pym/_emerge/EbuildMerge.py
index d73a262b3..6a5869270 100644
--- a/pym/_emerge/EbuildMerge.py
+++ b/pym/_emerge/EbuildMerge.py
@@ -4,6 +4,8 @@
 from _emerge.SlotObject import SlotObject
 import portage
 from portage import os
+from portage.dbapi._MergeProcess import MergeProcess
+from portage.dbapi.vartree import dblink
 
 class EbuildMerge(SlotObject):
 
@@ -11,28 +13,35 @@ class EbuildMerge(SlotObject):
 		"pkg", "pkg_count", "pkg_path", "pretend",
 		"scheduler", "settings", "tree", "world_atom")
 
-	def execute(self):
+	def create_task(self):
 		root_config = self.pkg.root_config
 		settings = self.settings
-		retval = portage.merge(settings["CATEGORY"],
-			settings["PF"], settings["D"],
-			os.path.join(settings["PORTAGE_BUILDDIR"],
-			"build-info"), root_config.root, settings,
-			myebuild=settings["EBUILD"],
-			mytree=self.tree, mydbapi=root_config.trees[self.tree].dbapi,
-			vartree=root_config.trees["vartree"],
-			prev_mtimes=self.ldpath_mtimes,
-			scheduler=self.scheduler,
-			blockers=self.find_blockers)
-
-		if retval == os.EX_OK:
-			self.world_atom(self.pkg)
-			self._log_success()
-
-		return retval
-
-	def _log_success(self):
+		mycat = settings["CATEGORY"]
+		mypkg = settings["PF"]
+		pkgloc = settings["D"]
+		infloc = os.path.join(settings["PORTAGE_BUILDDIR"], "build-info")
+		myroot = root_config.root
+		myebuild = settings["EBUILD"]
+		mydbapi = root_config.trees[self.tree].dbapi
+		vartree = root_config.trees["vartree"]
+		background = (settings.get('PORTAGE_BACKGROUND') == '1')
+		logfile = settings.get('PORTAGE_LOG_FILE')
+
+		merge_task = MergeProcess(
+			dblink=dblink, mycat=mycat, mypkg=mypkg, settings=settings,
+			treetype=self.tree, vartree=vartree, scheduler=self.scheduler,
+			background=background, blockers=self.find_blockers, pkgloc=pkgloc,
+			infloc=infloc, myebuild=myebuild, mydbapi=mydbapi,
+			prev_mtimes=self.ldpath_mtimes, logfile=logfile)
+		merge_task.addExitListener(self._log_exit)
+		return merge_task
+
+	def _log_exit(self, task):
+		if task.returncode != os.EX_OK:
+			return
+
 		pkg = self.pkg
+		self.world_atom(pkg)
 		pkg_count = self.pkg_count
 		pkg_path = self.pkg_path
 		logger = self.logger
diff --git a/pym/_emerge/MergeListItem.py b/pym/_emerge/MergeListItem.py
index 1dcc1780a..768865e6b 100644
--- a/pym/_emerge/MergeListItem.py
+++ b/pym/_emerge/MergeListItem.py
@@ -111,7 +111,7 @@ class MergeListItem(CompositeTask):
 		self._install_task.wait()
 		return self.returncode
 
-	def merge(self):
+	def merge(self, exit_handler):
 
 		pkg = self.pkg
 		build_opts = self.build_opts
@@ -135,15 +135,14 @@ class MergeListItem(CompositeTask):
 					world_atom=world_atom)
 
 				uninstall.start()
-				retval = uninstall.wait()
-				if retval != os.EX_OK:
-					return retval
-			return os.EX_OK
-
-		if build_opts.fetchonly or \
+				self.returncode = uninstall.wait()
+			else:
+				self.returncode = os.EX_OK
+			exit_handler(self)
+		elif build_opts.fetchonly or \
 			build_opts.buildpkgonly:
-			return self.returncode
-
-		retval = self._install_task.install()
-		return retval
+			exit_handler(self)
+		else:
+			self._current_task = self._install_task
+			self._install_task.install(exit_handler)
 
diff --git a/pym/_emerge/PackageMerge.py b/pym/_emerge/PackageMerge.py
index 4aecf8adb..45d2e7dc6 100644
--- a/pym/_emerge/PackageMerge.py
+++ b/pym/_emerge/PackageMerge.py
@@ -4,11 +4,6 @@
 from _emerge.AsynchronousTask import AsynchronousTask
 from portage.output import colorize
 class PackageMerge(AsynchronousTask):
-	"""
-	TODO: Implement asynchronous merge so that the scheduler can
-	run while a merge is executing.
-	"""
-
 	__slots__ = ("merge",)
 
 	def _start(self):
@@ -40,6 +35,9 @@ class PackageMerge(AsynchronousTask):
 			not self.merge.build_opts.buildpkgonly:
 			self.merge.statusMessage(msg)
 
-		self.returncode = self.merge.merge()
-		self.wait()
+		self.merge.merge(self.exit_handler)
+
+	def exit_handler(self, task):
+		self.returncode = task.returncode
+		self._wait_hook()