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

1 files changed, 69 insertions, 13 deletions

diff --git a/pym/portage/dbapi/_MergeProcess.py b/pym/portage/dbapi/_MergeProcess.py
index f717d12df..6e63f84fd 100644
--- a/pym/portage/dbapi/_MergeProcess.py
+++ b/pym/portage/dbapi/_MergeProcess.py
@@ -4,30 +4,72 @@
 import signal
 import traceback
 
+import errno
+import fcntl
 import portage
-from portage import os
+from portage import os, StringIO
+import portage.elog.messages
+from _emerge.PollConstants import PollConstants
 from _emerge.SpawnProcess import SpawnProcess
 
 class MergeProcess(SpawnProcess):
 	"""
-	Merge package files in a subprocess, so the Scheduler can run in the
-	main thread while files are moved or copied asynchronously.
+	Merge packages in a subprocess, so the Scheduler can run in the main
+	thread while files are moved or copied asynchronously.
 	"""
 
-	__slots__ = ('cfgfiledict', 'conf_mem_file', \
-		'destroot', 'dblink', 'srcroot',)
+	__slots__ = ('dblink', 'mycat', 'mypkg', 'settings', 'treetype',
+		'vartree', 'scheduler', 'blockers', 'pkgloc', 'infloc', 'myebuild',
+		'mydbapi', 'prev_mtimes', '_elog_reader_fd', '_elog_reg_id',
+		'_buf')
 
-	def _spawn(self, args, fd_pipes=None, **kwargs):
+	def _elog_output_handler(self, fd, event):
+		output = None
+		if event & PollConstants.POLLIN:
+			try:
+				output = os.read(fd, self._bufsize)
+			except OSError as e:
+				if e.errno not in (errno.EAGAIN, errno.EINTR):
+					raise
+		if output:
+			lines = output.split('\n')
+			if len(lines) == 1:
+				self._buf += lines[0]
+			else:
+				lines[0] = self._buf + lines[0]
+				self._buf = lines.pop()
+				out = StringIO()
+				for line in lines:
+					funcname, phase, key, msg = line.split(' ', 3)
+					reporter = getattr(portage.elog.messages, funcname)
+					reporter(msg, phase=phase, key=key, out=out)
+
+	def _spawn(self, args, fd_pipes, **kwargs):
 		"""
 		Fork a subprocess, apply local settings, and call
-		dblink._merge_process().
+		dblink.merge().
 		"""
 
+		files = self._files
+		elog_reader_fd, elog_writer_fd = os.pipe()
+		fcntl.fcntl(elog_reader_fd, fcntl.F_SETFL,
+			fcntl.fcntl(elog_reader_fd, fcntl.F_GETFL) | os.O_NONBLOCK)
+		mylink = self.dblink(self.mycat, self.mypkg, settings=self.settings,
+			treetype=self.treetype, vartree=self.vartree,
+			blockers=self.blockers, scheduler=self.scheduler,
+			pipe=elog_writer_fd)
+		fd_pipes[elog_writer_fd] = elog_writer_fd
+		self._elog_reg_id = self.scheduler.register(elog_reader_fd,
+			self._registered_events, self._elog_output_handler)
+
 		pid = os.fork()
 		if pid != 0:
+			self._elog_reader_fd = elog_reader_fd
+			self._buf = ""
 			portage.process.spawned_pids.append(pid)
 			return [pid]
 
+		os.close(elog_reader_fd)
 		portage.process._setup_pipes(fd_pipes)
 
 		# Use default signal handlers since the ones inherited
@@ -35,18 +77,19 @@ class MergeProcess(SpawnProcess):
 		signal.signal(signal.SIGINT, signal.SIG_DFL)
 		signal.signal(signal.SIGTERM, signal.SIG_DFL)
 
-		portage.output.havecolor = self.dblink.settings.get('NOCOLOR') \
+		portage.output.havecolor = self.settings.get('NOCOLOR') \
 			not in ('yes', 'true')
 
-		# In this subprocess we want dblink._display_merge() to use
+		# In this subprocess we want mylink._display_merge() to use
 		# stdout/stderr directly since they are pipes. This behavior
-		# is triggered when dblink._scheduler is None.
-		self.dblink._scheduler = None
+		# is triggered when mylink._scheduler is None.
+		mylink._scheduler = None
 
 		rval = 1
 		try:
-			rval = self.dblink._merge_process(self.srcroot, self.destroot,
-				self.cfgfiledict, self.conf_mem_file)
+			rval = mylink.merge(self.pkgloc, self.infloc,
+				myebuild=self.myebuild, mydbapi=self.mydbapi,
+				prev_mtimes=self.prev_mtimes)
 		except SystemExit:
 			raise
 		except:
@@ -55,3 +98,16 @@ class MergeProcess(SpawnProcess):
 			# Call os._exit() from finally block, in order to suppress any
 			# finally blocks from earlier in the call stack. See bug #345289.
 			os._exit(rval)
+
+	def _unregister(self):
+		"""
+		Unregister from the scheduler and close open files.
+		"""
+		if self._elog_reg_id is not None:
+			self.scheduler.unregister(self._elog_reg_id)
+			self._elog_reg_id = None
+		if self._elog_reader_fd:
+			os.close(self._elog_reader_fd)
+			self._elog_reader_fd = None
+
+		super(MergeProcess, self)._unregister()