[gentoo-commits] proj/portage:master commit in: lib/_emerge/, lib/portage/util/futures/executor/

[Zac Medico]

commit:     4df7a0a0c16c5ded65ad601d39840797b7704770
Author:     Zac Medico <zmedico <AT> gentoo <DOT> org>
AuthorDate: Sun Feb 23 21:44:58 2020 +0000
Commit:     Zac Medico <zmedico <AT> gentoo <DOT> org>
CommitDate: Mon Feb 24 02:35:15 2020 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=4df7a0a0

ForkExecutor: use async_start method

Also, fix AsynchronousTask.async_start to handle cancellation of the
_async_start coroutine, ensuring that start and exit listeners are
notified in this case (otherwise RetryForkExecutorTestCase will hang).

Signed-off-by: Zac Medico <zmedico <AT> gentoo.org>

 lib/_emerge/AsynchronousTask.py           | 15 +++++++++++++--
 lib/portage/util/futures/executor/fork.py | 21 ++++++++++++++++++---
 2 files changed, 31 insertions(+), 5 deletions(-)

diff --git a/lib/_emerge/AsynchronousTask.py b/lib/_emerge/AsynchronousTask.py
index d1e23cdf1..1e9e177cb 100644
--- a/lib/_emerge/AsynchronousTask.py
+++ b/lib/_emerge/AsynchronousTask.py
@@ -25,8 +25,19 @@ class AsynchronousTask(SlotObject):
 
 	@coroutine
 	def async_start(self):
-		yield self._async_start()
-		self._start_hook()
+		try:
+			if self._was_cancelled():
+				raise asyncio.CancelledError
+			yield self._async_start()
+			if self._was_cancelled():
+				raise asyncio.CancelledError
+		except asyncio.CancelledError:
+			self.cancel()
+			self._was_cancelled()
+			self._async_wait()
+			raise
+		finally:
+			self._start_hook()
 
 	@coroutine
 	def _async_start(self):

diff --git a/lib/portage/util/futures/executor/fork.py b/lib/portage/util/futures/executor/fork.py
index add7b3c9e..3549fdb31 100644
--- a/lib/portage/util/futures/executor/fork.py
+++ b/lib/portage/util/futures/executor/fork.py
@@ -13,6 +13,7 @@ import traceback
 
 from portage.util._async.AsyncFunction import AsyncFunction
 from portage.util.futures import asyncio
+from portage.util.futures.compat_coroutine import coroutine
 from portage.util.cpuinfo import get_cpu_count
 
 
@@ -51,11 +52,25 @@ class ForkExecutor(object):
 		while (not self._shutdown and self._submit_queue and
 			len(self._running_tasks) < self._max_workers):
 			future, proc = self._submit_queue.popleft()
-			future.add_done_callback(functools.partial(self._cancel_cb, proc))
-			proc.addExitListener(functools.partial(self._proc_exit, future))
 			proc.scheduler = self._loop
-			proc.start()
 			self._running_tasks[id(proc)] = proc
+			future.add_done_callback(functools.partial(self._cancel_cb, proc))
+			proc_future = asyncio.ensure_future(self._proc_coroutine(proc), loop=self._loop)
+			proc_future.add_done_callback(functools.partial(self._proc_coroutine_done, future, proc))
+
+	@coroutine
+	def _proc_coroutine(self, proc):
+		yield proc.async_start()
+		yield proc.async_wait()
+
+	def _proc_coroutine_done(self, future, proc, proc_future):
+		try:
+			proc_future.result()
+		except asyncio.CancelledError:
+			future.done() or future.cancel()
+			if proc.poll() is None:
+				proc.cancel()
+		self._proc_exit(future, proc)
 
 	def _cancel_cb(self, proc, future):
 		if future.cancelled():