SpawnProcess: split out a PipeLogger class

The copyright dates for these classes begin in 2008, since SpawnProcess
code is derived from the EbuildFetcherAsync class which was added in
commit e4edadf5ae7063f375d76be151c6d0e949980ecf in 2008.

1 files changed, 34 insertions, 131 deletions

diff --git a/pym/_emerge/SpawnProcess.py b/pym/_emerge/SpawnProcess.py
index ab152c3c3..d18512b34 100644
--- a/pym/_emerge/SpawnProcess.py
+++ b/pym/_emerge/SpawnProcess.py
@@ -1,17 +1,12 @@
-# Copyright 1999-2012 Gentoo Foundation
+# Copyright 2008-2012 Gentoo Foundation
 # Distributed under the terms of the GNU General Public License v2
 
 from _emerge.SubProcess import SubProcess
 import sys
-from portage.cache.mappings import slot_dict_class
 import portage
-from portage import _encodings
-from portage import _unicode_encode
 from portage import os
 from portage.const import BASH_BINARY
-import fcntl
-import errno
-import gzip
+from portage.util._async.PipeLogger import PipeLogger
 
 class SpawnProcess(SubProcess):
 
@@ -26,10 +21,7 @@ class SpawnProcess(SubProcess):
 		"path_lookup", "pre_exec")
 
 	__slots__ = ("args",) + \
-		_spawn_kwarg_names + ("_log_file_real", "_selinux_type",)
-
-	_file_names = ("log", "process", "stdout")
-	_files_dict = slot_dict_class(_file_names, prefix="")
+		_spawn_kwarg_names + ("_pipe_logger", "_selinux_type",)
 
 	def _start(self):
 
@@ -37,17 +29,13 @@ class SpawnProcess(SubProcess):
 			self.fd_pipes = {}
 		fd_pipes = self.fd_pipes
 
-		self._files = self._files_dict()
-		files = self._files
-
 		master_fd, slave_fd = self._pipe(fd_pipes)
-		fcntl.fcntl(master_fd, fcntl.F_SETFL,
-			fcntl.fcntl(master_fd, fcntl.F_GETFL) | os.O_NONBLOCK)
-		files.process = master_fd
 
-		logfile = None
-		if self._can_log(slave_fd):
-			logfile = self.logfile
+		can_log = self._can_log(slave_fd)
+		if can_log:
+			log_file_path = self.logfile
+		else:
+			log_file_path = None
 
 		null_input = None
 		if not self.background or 0 in fd_pipes:
@@ -74,37 +62,19 @@ class SpawnProcess(SubProcess):
 				sys.__stderr__.flush()
 				break
 
-		if logfile is not None:
+		fd_pipes_orig = fd_pipes.copy()
 
-			fd_pipes_orig = fd_pipes.copy()
+		if log_file_path is not None:
 			fd_pipes[1] = slave_fd
 			fd_pipes[2] = slave_fd
 
-			files.log = open(_unicode_encode(logfile,
-				encoding=_encodings['fs'], errors='strict'), mode='ab')
-			if logfile.endswith('.gz'):
-				self._log_file_real = files.log
-				files.log = gzip.GzipFile(filename='', mode='ab',
-					fileobj=files.log)
-
-			portage.util.apply_secpass_permissions(logfile,
-				uid=portage.portage_uid, gid=portage.portage_gid,
-				mode=0o660)
-
-			if not self.background:
-				files.stdout = os.dup(fd_pipes_orig[1])
-
-			output_handler = self._output_handler
-
 		else:
-
 			# Create a dummy pipe so the scheduler can monitor
 			# the process from inside a poll() loop.
 			fd_pipes[self._dummy_pipe_fd] = slave_fd
 			if self.background:
 				fd_pipes[1] = slave_fd
 				fd_pipes[2] = slave_fd
-			output_handler = self._dummy_handler
 
 		kwargs = {}
 		for k in self._spawn_kwarg_names:
@@ -116,10 +86,6 @@ class SpawnProcess(SubProcess):
 		kwargs["returnpid"] = True
 		kwargs.pop("logfile", None)
 
-		self._reg_id = self.scheduler.io_add_watch(files.process,
-			self._registered_events, output_handler)
-		self._registered = True
-
 		retval = self._spawn(self.args, **kwargs)
 
 		os.close(slave_fd)
@@ -136,6 +102,18 @@ class SpawnProcess(SubProcess):
 		self.pid = retval[0]
 		portage.process.spawned_pids.remove(self.pid)
 
+		stdout_fd = None
+		if can_log and not self.background:
+			stdout_fd = os.dup(fd_pipes_orig[1])
+
+		self._pipe_logger = PipeLogger(background=self.background,
+			scheduler=self.scheduler, input_fd=master_fd,
+			log_file_path=log_file_path,
+			stdout_fd=stdout_fd)
+		self._pipe_logger.addExitListener(self._pipe_logger_exit)
+		self._pipe_logger.start()
+		self._registered = True
+
 	def _can_log(self, slave_fd):
 		return True
 
@@ -158,92 +136,17 @@ class SpawnProcess(SubProcess):
 
 		return spawn_func(args, **kwargs)
 
-	def _output_handler(self, fd, event):
-
-		files = self._files
-		while True:
-			buf = self._read_buf(fd, event)
-
-			if buf is None:
-				# not a POLLIN event, EAGAIN, etc...
-				break
+	def _pipe_logger_exit(self, pipe_logger):
+		self._pipe_logger = None
+		self._unregister()
+		self.wait()
 
-			if not buf:
-				# EOF
-				self._unregister()
-				self.wait()
-				break
-
-			else:
-				if not self.background:
-					write_successful = False
-					failures = 0
-					while True:
-						try:
-							if not write_successful:
-								os.write(files.stdout, buf)
-								write_successful = True
-							break
-						except OSError as e:
-							if e.errno != errno.EAGAIN:
-								raise
-							del e
-							failures += 1
-							if failures > 50:
-								# Avoid a potentially infinite loop. In
-								# most cases, the failure count is zero
-								# and it's unlikely to exceed 1.
-								raise
-
-							# This means that a subprocess has put an inherited
-							# stdio file descriptor (typically stdin) into
-							# O_NONBLOCK mode. This is not acceptable (see bug
-							# #264435), so revert it. We need to use a loop
-							# here since there's a race condition due to
-							# parallel processes being able to change the
-							# flags on the inherited file descriptor.
-							# TODO: When possible, avoid having child processes
-							# inherit stdio file descriptors from portage
-							# (maybe it can't be avoided with
-							# PROPERTIES=interactive).
-							fcntl.fcntl(files.stdout, fcntl.F_SETFL,
-								fcntl.fcntl(files.stdout,
-								fcntl.F_GETFL) ^ os.O_NONBLOCK)
-
-				files.log.write(buf)
-				files.log.flush()
-
-		self._unregister_if_appropriate(event)
-
-		return True
-
-	def _dummy_handler(self, fd, event):
-		"""
-		This method is mainly interested in detecting EOF, since
-		the only purpose of the pipe is to allow the scheduler to
-		monitor the process from inside a poll() loop.
-		"""
-
-		while True:
-			buf = self._read_buf(fd, event)
-
-			if buf is None:
-				# not a POLLIN event, EAGAIN, etc...
-				break
-
-			if not buf:
-				# EOF
-				self._unregister()
-				self.wait()
-				break
-
-		self._unregister_if_appropriate(event)
-
-		return True
+	def _waitpid_loop(self):
+		SubProcess._waitpid_loop(self)
 
-	def _unregister(self):
-		super(SpawnProcess, self)._unregister()
-		if self._log_file_real is not None:
-			# Avoid "ResourceWarning: unclosed file" since python 3.2.
-			self._log_file_real.close()
-			self._log_file_real = None
+		pipe_logger = self._pipe_logger
+		if pipe_logger is not None:
+			self._pipe_logger = None
+			pipe_logger.removeExitListener(self._pipe_logger_exit)
+			pipe_logger.cancel()
+			pipe_logger.wait()