SpawnProcess: use async_start method (bug 709746)

Convert SpawnProcess to use the async_start method, since
eventually this method will need to be a coroutine in order to write
messages to the build log as discussed in bug 709746.

Also update AbstractEbuildProcess, AsynchronousLock,
BinpkgExtractorAsync, BinpkgFetcher, EbuildFetcher,
IterCompletedTestCase, AsyncFunction, and FileDigester for
compatibility with SpawnProcess async_start.

Bug: https://bugs.gentoo.org/709746
Signed-off-by: Zac Medico <zmedico@gentoo.org>

lib/_emerge/AbstractEbuildProcess.py

@@ -182,7 +182,7 @@ def _async_start(self):
 			self.fd_pipes[0] = null_fd
 
 		try:
-			SpawnProcess._start(self)
+			yield SpawnProcess._async_start(self)
 		finally:
 			if null_fd is not None:
 				os.close(null_fd)

lib/_emerge/AsynchronousLock.py

@@ -1,4 +1,4 @@
-# Copyright 2010-2018 Gentoo Foundation
+# Copyright 2010-2020 Gentoo Authors
 # Distributed under the terms of the GNU General Public License v2
 
 import fcntl
@@ -21,6 +21,7 @@
 from portage.localization import _
 from portage.locks import lockfile, unlockfile
 from portage.util import writemsg_level
+from portage.util.futures.compat_coroutine import coroutine
 from _emerge.AbstractPollTask import AbstractPollTask
 from _emerge.AsynchronousTask import AsynchronousTask
 from _emerge.SpawnProcess import SpawnProcess
@@ -43,6 +44,10 @@ class AsynchronousLock(AsynchronousTask):
 	_use_process_by_default = True
 
 	def _start(self):
+		self.scheduler.run_until_complete(self._async_start())
+
+	@coroutine
+	def _async_start(self):
 
 		if not self._force_async:
 			try:
@@ -65,7 +70,7 @@ def _start(self):
 				_force_dummy=self._force_dummy)
 
 		self._imp.addExitListener(self._imp_exit)
-		self._imp.start()
+		yield self._imp.async_start()
 
 	def _imp_exit(self, imp):
 		# call exit listeners
@@ -183,6 +188,10 @@ class _LockProcess(AbstractPollTask):
 		('_acquired', '_kill_test', '_proc', '_files', '_unlock_future')
 
 	def _start(self):
+		self.scheduler.run_until_complete(self._async_start())
+
+	@coroutine
+	def _async_start(self):
 		in_pr, in_pw = os.pipe()
 		out_pr, out_pw = os.pipe()
 		self._files = {}
@@ -211,7 +220,7 @@ def _start(self):
 				fd_pipes={0:out_pr, 1:in_pw, 2:sys.__stderr__.fileno()},
 				scheduler=self.scheduler)
 		self._proc.addExitListener(self._proc_exit)
-		self._proc.start()
+		yield self._proc.async_start()
 		os.close(out_pr)
 		os.close(in_pw)
 

lib/_emerge/BinpkgExtractorAsync.py

@@ -1,4 +1,4 @@
-# Copyright 1999-2013 Gentoo Foundation
+# Copyright 1999-2020 Gentoo Authors
 # Distributed under the terms of the GNU General Public License v2
 
 import logging
@@ -10,6 +10,7 @@
 	compression_probe,
 	_compressors,
 )
+from portage.util.futures.compat_coroutine import coroutine
 from portage.process import find_binary
 from portage.util import (
 	shlex_split,
@@ -27,6 +28,10 @@ class BinpkgExtractorAsync(SpawnProcess):
 	_shell_binary = portage.const.BASH_BINARY
 
 	def _start(self):
+		self.scheduler.run_until_complete(self._async_start())
+
+	@coroutine
+	def _async_start(self):
 		tar_options = ""
 		if "xattr" in self.features:
 			process = subprocess.Popen(["tar", "--help"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
@@ -105,4 +110,4 @@ def _start(self):
 			portage._shell_quote(self.image_dir),
 			128 + signal.SIGPIPE)]
 
-		SpawnProcess._start(self)
+		yield SpawnProcess._async_start(self)

lib/_emerge/BinpkgFetcher.py

@@ -1,4 +1,4 @@
-# Copyright 1999-2018 Gentoo Foundation
+# Copyright 1999-2020 Gentoo Authors
 # Distributed under the terms of the GNU General Public License v2
 
 import functools
@@ -16,6 +16,7 @@
 from portage import os
 from portage.util._async.AsyncTaskFuture import AsyncTaskFuture
 from portage.util._pty import _create_pty_or_pipe
+from portage.util.futures.compat_coroutine import coroutine
 
 if sys.hexversion >= 0x3000000:
 	long = int
@@ -85,6 +86,10 @@ class _BinpkgFetcherProcess(SpawnProcess):
 	__slots__ = ("pkg", "pretend", "locked", "pkg_path", "_lock_obj")
 
 	def _start(self):
+		self.scheduler.run_until_complete(self._async_start())
+
+	@coroutine
+	def _async_start(self):
 		pkg = self.pkg
 		pretend = self.pretend
 		bintree = pkg.root_config.trees["bintree"]
@@ -158,7 +163,7 @@ def _start(self):
 		self.env = fetch_env
 		if settings.selinux_enabled():
 			self._selinux_type = settings["PORTAGE_FETCH_T"]
-		SpawnProcess._start(self)
+		yield SpawnProcess._async_start(self)
 
 	def _pipe(self, fd_pipes):
 		"""When appropriate, use a pty so that fetcher progress bars,

lib/_emerge/EbuildFetcher.py

@@ -1,4 +1,4 @@
-# Copyright 1999-2018 Gentoo Foundation
+# Copyright 1999-2020 Gentoo Authors
 # Distributed under the terms of the GNU General Public License v2
 
 import copy
@@ -15,6 +15,7 @@
 from portage.package.ebuild.fetch import _check_distfile, fetch
 from portage.util._async.AsyncTaskFuture import AsyncTaskFuture
 from portage.util._async.ForkProcess import ForkProcess
+from portage.util.futures.compat_coroutine import coroutine
 from portage.util._pty import _create_pty_or_pipe
 from _emerge.CompositeTask import CompositeTask
 
@@ -182,6 +183,10 @@ def _check_already_fetched(self, settings, uri_map):
 		return success
 
 	def _start(self):
+		self.scheduler.run_until_complete(self._async_start())
+
+	@coroutine
+	def _async_start(self):
 
 		root_config = self.pkg.root_config
 		portdb = root_config.trees["porttree"].dbapi
@@ -220,7 +225,7 @@ def _start(self):
 			settings["NOCOLOR"] = nocolor
 
 		self._settings = settings
-		ForkProcess._start(self)
+		yield ForkProcess._async_start(self)
 
 		# Free settings now since it's no longer needed in
 		# this process (the subprocess has a private copy).

lib/_emerge/SpawnProcess.py

@@ -1,4 +1,4 @@
-# Copyright 2008-2018 Gentoo Foundation
+# Copyright 2008-2020 Gentoo Authors
 # Distributed under the terms of the GNU General Public License v2
 
 try:
@@ -20,6 +20,7 @@
 from portage.output import EOutput
 from portage.util import writemsg_level
 from portage.util._async.PipeLogger import PipeLogger
+from portage.util.futures.compat_coroutine import coroutine
 
 class SpawnProcess(SubProcess):
 
@@ -42,7 +43,10 @@ class SpawnProcess(SubProcess):
 	_CGROUP_CLEANUP_RETRY_MAX = 8
 
 	def _start(self):
+		self.scheduler.run_until_complete(self._async_start())
 
+	@coroutine
+	def _async_start(self):
 		if self.fd_pipes is None:
 			self.fd_pipes = {}
 		else:
@@ -142,8 +146,8 @@ def _start(self):
 			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
+		yield self._pipe_logger.async_start()
 
 	def _can_log(self, slave_fd):
 		return True

lib/portage/dbapi/bintree.py

@@ -1,4 +1,4 @@
-# Copyright 1998-2019 Gentoo Authors
+# Copyright 1998-2020 Gentoo Authors
 # Distributed under the terms of the GNU General Public License v2
 
 from __future__ import unicode_literals
@@ -283,7 +283,7 @@ def unpack_contents(self, pkg, dest_dir):
 				logfile=settings.get('PORTAGE_LOG_FILE'),
 				scheduler=SchedulerInterface(loop))
 
-			extractor.start()
+			yield extractor.async_start()
 			yield extractor.async_wait()
 			if extractor.returncode != os.EX_OK:
 				raise PortageException("Error Extracting '{}'".format(pkg_path))

lib/portage/tests/util/futures/test_iter_completed.py

@@ -16,8 +16,12 @@
 class SleepProcess(ForkProcess):
 	__slots__ = ('future', 'seconds')
 	def _start(self):
+		self.scheduler.run_until_complete(self._async_start())
+
+	@coroutine
+	def _async_start(self):
 		self.addExitListener(self._future_done)
-		ForkProcess._start(self)
+		yield ForkProcess._async_start(self)
 
 	def _future_done(self, task):
 		if not self.future.cancelled():

lib/portage/util/_async/AsyncFunction.py

@@ -1,11 +1,12 @@
-# Copyright 2015 Gentoo Foundation
+# Copyright 2015-2020 Gentoo Authors
 # Distributed under the terms of the GNU General Public License v2
 
 import pickle
 import traceback
 
 from portage import os
 from portage.util._async.ForkProcess import ForkProcess
+from portage.util.futures.compat_coroutine import coroutine
 from _emerge.PipeReader import PipeReader
 
 class AsyncFunction(ForkProcess):
@@ -22,6 +23,10 @@ class AsyncFunction(ForkProcess):
 		'_async_func_reader', '_async_func_reader_pw')
 
 	def _start(self):
+		self.scheduler.run_until_complete(self._async_start())
+
+	@coroutine
+	def _async_start(self):
 		pr, pw = os.pipe()
 		self.fd_pipes = {}
 		self.fd_pipes[pw] = pw
@@ -31,7 +36,7 @@ def _start(self):
 			scheduler=self.scheduler)
 		self._async_func_reader.addExitListener(self._async_func_reader_exit)
 		self._async_func_reader.start()
-		ForkProcess._start(self)
+		yield ForkProcess._async_start(self)
 		os.close(pw)
 
 	def _run(self):

lib/portage/util/_async/FileDigester.py

@@ -1,9 +1,10 @@
-# Copyright 2013 Gentoo Foundation
+# Copyright 2013-2020 Gentoo Authors
 # Distributed under the terms of the GNU General Public License v2
 
 from portage import os
 from portage.checksum import perform_multiple_checksums
 from portage.util._async.ForkProcess import ForkProcess
+from portage.util.futures.compat_coroutine import coroutine
 from _emerge.PipeReader import PipeReader
 
 class FileDigester(ForkProcess):
@@ -18,6 +19,10 @@ class FileDigester(ForkProcess):
 		'_digest_pipe_reader', '_digest_pw')
 
 	def _start(self):
+		self.scheduler.run_until_complete(self._async_start())
+
+	@coroutine
+	def _async_start(self):
 		pr, pw = os.pipe()
 		self.fd_pipes = {}
 		self.fd_pipes[pw] = pw
@@ -27,7 +32,7 @@ def _start(self):
 			scheduler=self.scheduler)
 		self._digest_pipe_reader.addExitListener(self._digest_pipe_reader_exit)
 		self._digest_pipe_reader.start()
-		ForkProcess._start(self)
+		yield ForkProcess._async_start(self)
 		os.close(pw)
 
 	def _run(self):