[gentoo-commits] proj/portage:master commit in: pym/portage/util/futures/, pym/portage/tests/util/futures/asyncio/, ...

[gentoo-commits] proj/portage:master commit in: pym/portage/util/futures/, pym/portage/tests/util/futures/asyncio/, ...

[Zac Medico]

commit:     a78dca7e47f79ad48aee4909ee10688604996b86
Author:     Zac Medico <zmedico <AT> gentoo <DOT> org>
AuthorDate: Wed Apr 11 06:44:41 2018 +0000
Commit:     Zac Medico <zmedico <AT> gentoo <DOT> org>
CommitDate: Thu Apr 12 08:35:05 2018 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=a78dca7e

Implement AbstractEventLoopPolicy.get_child_watcher() (bug 649588)

Use a _PortageChildWatcher class to wrap portage's internal event loop
and implement asyncio's AbstractChildWatcher interface.

Bug: https://bugs.gentoo.org/649588

 .../util/futures/asyncio/test_child_watcher.py     | 45 +++++++++++
 pym/portage/util/_eventloop/EventLoop.py           |  7 +-
 pym/portage/util/futures/_asyncio.py               | 13 ++++
 pym/portage/util/futures/unix_events.py            | 90 ++++++++++++++++++++++
 4 files changed, 152 insertions(+), 3 deletions(-)

diff --git a/pym/portage/tests/util/futures/asyncio/test_child_watcher.py b/pym/portage/tests/util/futures/asyncio/test_child_watcher.py
new file mode 100644
index 000000000..dca01be56
--- /dev/null
+++ b/pym/portage/tests/util/futures/asyncio/test_child_watcher.py
@@ -0,0 +1,45 @@
+# Copyright 2018 Gentoo Foundation
+# Distributed under the terms of the GNU General Public License v2
+
+import os
+
+from portage.process import find_binary, spawn
+from portage.tests import TestCase
+from portage.util.futures import asyncio
+from portage.util.futures.unix_events import DefaultEventLoopPolicy
+
+
+class ChildWatcherTestCase(TestCase):
+	def testChildWatcher(self):
+		true_binary = find_binary("true")
+		self.assertNotEqual(true_binary, None)
+
+		initial_policy = asyncio.get_event_loop_policy()
+		if not isinstance(initial_policy, DefaultEventLoopPolicy):
+			asyncio.set_event_loop_policy(DefaultEventLoopPolicy())
+
+		try:
+			try:
+				asyncio.set_child_watcher(None)
+			except NotImplementedError:
+				pass
+			else:
+				self.assertTrue(False)
+
+			args_tuple = ('hello', 'world')
+
+			loop = asyncio.get_event_loop()
+			future = loop.create_future()
+
+			def callback(pid, returncode, *args):
+				future.set_result((pid, returncode, args))
+
+			with asyncio.get_child_watcher() as watcher:
+				pids = spawn([true_binary], returnpid=True)
+				watcher.add_child_handler(pids[0], callback, *args_tuple)
+
+				self.assertEqual(
+					loop.run_until_complete(future),
+					(pids[0], os.EX_OK, args_tuple))
+		finally:
+			asyncio.set_event_loop_policy(initial_policy)

diff --git a/pym/portage/util/_eventloop/EventLoop.py b/pym/portage/util/_eventloop/EventLoop.py
index d53a76ba1..12c199c76 100644
--- a/pym/portage/util/_eventloop/EventLoop.py
+++ b/pym/portage/util/_eventloop/EventLoop.py
@@ -25,7 +25,7 @@ import portage
 portage.proxy.lazyimport.lazyimport(globals(),
 	'portage.util.futures.futures:Future',
 	'portage.util.futures.executor.fork:ForkExecutor',
-	'portage.util.futures.unix_events:_PortageEventLoop',
+	'portage.util.futures.unix_events:_PortageEventLoop,_PortageChildWatcher',
 )
 
 from portage import OrderedDict
@@ -190,6 +190,7 @@ class EventLoop(object):
 		self._sigchld_src_id = None
 		self._pid = os.getpid()
 		self._asyncio_wrapper = _PortageEventLoop(loop=self)
+		self._asyncio_child_watcher = _PortageChildWatcher(self)
 
 	def create_future(self):
 		"""
@@ -424,8 +425,8 @@ class EventLoop(object):
 					self._sigchld_read, self.IO_IN, self._sigchld_io_cb)
 				signal.signal(signal.SIGCHLD, self._sigchld_sig_cb)
 
-		# poll now, in case the SIGCHLD has already arrived
-		self._poll_child_processes()
+		# poll soon, in case the SIGCHLD has already arrived
+		self.call_soon(self._poll_child_processes)
 		return source_id
 
 	def _sigchld_sig_cb(self, signum, frame):

diff --git a/pym/portage/util/futures/_asyncio.py b/pym/portage/util/futures/_asyncio.py
index 02ab59999..0f84f14b7 100644
--- a/pym/portage/util/futures/_asyncio.py
+++ b/pym/portage/util/futures/_asyncio.py
@@ -3,7 +3,9 @@
 
 __all__ = (
 	'ensure_future',
+	'get_child_watcher',
 	'get_event_loop',
+	'set_child_watcher',
 	'get_event_loop_policy',
 	'set_event_loop_policy',
 	'sleep',
@@ -62,6 +64,17 @@ def get_event_loop():
 	return get_event_loop_policy().get_event_loop()
 
 
+def get_child_watcher():
+    """Equivalent to calling get_event_loop_policy().get_child_watcher()."""
+    return get_event_loop_policy().get_child_watcher()
+
+
+def set_child_watcher(watcher):
+    """Equivalent to calling
+    get_event_loop_policy().set_child_watcher(watcher)."""
+    return get_event_loop_policy().set_child_watcher(watcher)
+
+
 class Task(Future):
 	"""
 	Schedule the execution of a coroutine: wrap it in a future. A task

diff --git a/pym/portage/util/futures/unix_events.py b/pym/portage/util/futures/unix_events.py
index ed4c6e519..6fcef45fa 100644
--- a/pym/portage/util/futures/unix_events.py
+++ b/pym/portage/util/futures/unix_events.py
@@ -2,9 +2,17 @@
 # Distributed under the terms of the GNU General Public License v2
 
 __all__ = (
+	'AbstractChildWatcher',
 	'DefaultEventLoopPolicy',
 )
 
+try:
+	from asyncio.unix_events import AbstractChildWatcher as _AbstractChildWatcher
+except ImportError:
+	_AbstractChildWatcher = object
+
+import os
+
 from portage.util._eventloop.global_event_loop import (
 	global_event_loop as _global_event_loop,
 )
@@ -68,6 +76,84 @@ class _PortageEventLoop(events.AbstractEventLoop):
 		return asyncio.Task(coro, loop=self)
 
 
+class AbstractChildWatcher(_AbstractChildWatcher):
+	def add_child_handler(self, pid, callback, *args):
+		raise NotImplementedError()
+
+	def remove_child_handler(self, pid):
+		raise NotImplementedError()
+
+	def attach_loop(self, loop):
+		raise NotImplementedError()
+
+	def close(self):
+		raise NotImplementedError()
+
+	def __enter__(self):
+		raise NotImplementedError()
+
+	def __exit__(self, a, b, c):
+		raise NotImplementedError()
+
+
+class _PortageChildWatcher(_AbstractChildWatcher):
+	def __init__(self, loop):
+		"""
+		@type loop: EventLoop
+		@param loop: an instance of portage's internal event loop
+		"""
+		self._loop = loop
+		self._callbacks = {}
+
+	def close(self):
+		pass
+
+	def __enter__(self):
+		return self
+
+	def __exit__(self, a, b, c):
+		pass
+
+	def _child_exit(self, pid, status, data):
+		self._callbacks.pop(pid)
+		callback, args = data
+		callback(pid, self._compute_returncode(status), *args)
+
+	def _compute_returncode(self, status):
+		if os.WIFSIGNALED(status):
+			return -os.WTERMSIG(status)
+		elif os.WIFEXITED(status):
+			return os.WEXITSTATUS(status)
+		else:
+			return status
+
+	def add_child_handler(self, pid, callback, *args):
+		"""
+		Register a new child handler.
+
+		Arrange for callback(pid, returncode, *args) to be called when
+		process 'pid' terminates. Specifying another callback for the same
+		process replaces the previous handler.
+		"""
+		source_id = self._callbacks.get(pid)
+		if source_id is not None:
+			self._loop.source_remove(source_id)
+		self._callbacks[pid] = self._loop.child_watch_add(
+			pid, self._child_exit, data=(callback, args))
+
+	def remove_child_handler(self, pid):
+		"""
+		Removes the handler for process 'pid'.
+
+		The function returns True if the handler was successfully removed,
+		False if there was nothing to remove.
+		"""
+		source_id = self._callbacks.pop(pid, None)
+		if source_id is not None:
+			return self._loop.source_remove(source_id)
+		return False
+
+
 class _PortageEventLoopPolicy(events.AbstractEventLoopPolicy):
 	"""
 	Implementation of asyncio.AbstractEventLoopPolicy based on portage's
@@ -87,5 +173,9 @@ class _PortageEventLoopPolicy(events.AbstractEventLoopPolicy):
 		"""
 		return _global_event_loop()._asyncio_wrapper
 
+	def get_child_watcher(self):
+		"""Get the watcher for child processes."""
+		return _global_event_loop()._asyncio_child_watcher
+
 
 DefaultEventLoopPolicy = _PortageEventLoopPolicy