Merge packages asynchronously in Portage.

pym/portage/dbapi/_MergeProcess.py

Index: pym/portage/dbapi/_MergeProcess.py
diff --git a/pym/portage/dbapi/_MergeProcess.py b/pym/portage/dbapi/_MergeProcess.py
index f717d12df81b5bc7ca513cafee26afcc5d63daa1..6e63f84fda3c78d4eec64b5820d17b3878478781 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()